peer.java

java语言开发的P2P流媒体系统

			
			try
			{
				childData.getSocket().send(packet);
			}
			catch (IOException e)
			{
				Logger.fine("Peer", "IO_ERROR", e); //$NON-NLS-1$ //$NON-NLS-2$
			}
			
			removeChild(freeloader);
		}
	}
	
	// Den aktuellen Zulieferer melden
	public synchronized void reportFreeloader()
	{
		// Ist der Zulieferer die Wurzel?
		if (supplierFather != null)
		{
			reportFreeloader(supplier, supplierFather);
		}
	}
	
	public void reportFreeloader(RemotePeer freeloader, RemotePeer freeloaderSupplier)
	{	
		if (freeloader.isMulticastAddress())
			return;
		// Eigenes UDP-Socket erstellen
		DatagramSocket mySocket = null;
		try
		{
			mySocket = new DatagramSocket();
		}
		catch (Exception e)
		{
			Logger.warning("Peer", "COULD_NOT_CREATE_UDP_SOCKET", e); //$NON-NLS-1$ //$NON-NLS-2$
			return; 
		}
		
		// Nachricht an den Zulieferer des Freeloader senden
		
		DatagramPacket outPacket = null; 
		try
		{
			Logger.fine("Peer", "Peer.REPORTING_FREELOADER", new RemotePeer[]{freeloader, freeloaderSupplier}); //$NON-NLS-1$ //$NON-NLS-2$
						
			outPacket = PacketFactory.createUDPPacket(new FreeloaderReportPacket(getSocketAddress().getPort(), freeloader), freeloaderSupplier);
			mySocket.send(outPacket);
		}
		catch (Exception e)
		{
			Logger.fine("Peer", "Peer.ERROR_WHILE_REPORTING_FREELOADER", e); //$NON-NLS-1$ //$NON-NLS-2$
			
			mySocket.close();
			return;
		}
			
		mySocket.close();
	}
	
	/**
	 * Returns the data belonging to the specified child.
	 */
	public synchronized RemotePeerData getChildData(RemotePeer child)
	{
		return (RemotePeerData)childrenData.get(child);
	}
	
	/**
	 * Tells whether the specified peer is banned.
	 */
	public synchronized boolean isPeerBanned(RemotePeer peer)
	{
		return bannedPeers.contains(peer);
	}
	
	public synchronized RemotePeer searchChildWithAddress(InetSocketAddress address)
	{
		for (int i=0; i < children.size(); i++)
		{
			RemotePeer peer = (RemotePeer)children.get(i);
			RemotePeerData data = getChildData(peer);
			
			if (!data.hasBeenConnected())
			{
				if (peer.getSocketAddress().equals(address))
				{
					return peer;
				}
			}
		}
		
		return null;
	}
	
	public synchronized void forgetSomeFreeloaders()
	{
		for (int i=0; i < children.size(); i++)
		{
			RemotePeer remotePeer = (RemotePeer)children.get(i);
			RemotePeerData data = getChildData(remotePeer);
			
			data.removeRandomFreeloaderComplainant();
		}
	}
	
	public synchronized void forgetSomeBannedPeers()
	{
		if (bannedPeers.size() != 0)
		{
			bannedPeers.remove(random.nextInt(bannedPeers.size()));
		}
	}
	
	/**
	 * Returns the buffer of this peer.
	 */
	public Buffer getBuffer()
	{
		return buffer;
	}
	
	public synchronized void removeUnconnectedChildren()
	{
		// Alle Kinder entfernen, die nach einer gewissen Zeit
		// noch keine TCP-Verbindung aufgebaut haben
		
		long now = System.currentTimeMillis();
		
		for (int i=0; i < children.size(); i++)
		{
			RemotePeer child = (RemotePeer)children.get(i);
			if (child.isMulticastAddress())
				continue;
			RemotePeerData data = getChildData(child);
			
			if (!data.hasBeenConnected())
			{
				if (data.getTimeToConnect() < now)
				{
					Logger.fine("Peer", "Peer.CHILD_REMOVED_CONNECT_TIMEOUT", child); //$NON-NLS-1$ //$NON-NLS-2$
					removeChild(child);
					
					// Und gleich noch auf die schwarze Liste setzen
					bannedPeers.add(child);
				}
			}
		}
	}
	
	/**
	 * Returns the maximal upload bandwidth.
	 */
	public synchronized double getMaxUploadBandwidth()
	{
		return maxUploadBandwidth;
	}
	
	/**
	 * Sets the maximal upload bandwidth that this peer can use.
	 * 
	 * @param KBperSec The maximal upload bandwidth (in KiloBytes/s).
	 */
	public synchronized void setMaxUploadBandwidth(double KBperSec)
	{
		maxUploadBandwidth = KBperSec;
	}
	public void setMaxUploadBandwidth()
	{
		if(verifiedBandwidth)
		{
			setMaxUploadBandwidth(internetUpload);
		}
		else
		{
			setMaxUploadBandwidth(LANUpload);
		}
	}
	
	/**
	 * Returns the monitor log handler of this peer, or <code>null</code> if there's none.
	 */
	public MonitorLogHandler getMonitorLogHandler()
	{
		return monitorLogHandler;
	}
	
	public synchronized boolean getMisbehavior(int index)
	{
		return misbehavior[index];
	}
	
	public synchronized void setMisbehavior(int index, boolean misbehave)
	{
		misbehavior[index] = misbehave;
	}
	
	/**
	 * Returns whether this peer is disconnected from the P2P
	 * network. A peer is disconnected if it has no supplier and
	 * it isn't trying to connect.
	 * <P>
	 * If the peer is a server, this method always returns <code>false</code>.
	 */
	public synchronized boolean isDisconnected()
	{
		// Ein Server ist nie unverbunden
		if (isServer())
		{
			return false;
		}
		else
		{
			return ((getSupplier() == null) && (joinThread != null) && !joinThread.isAlive());
		}
	}
	
	public synchronized boolean wasOnceConnected()
	{
		return wasOnceConnected;
	}
	
	/**
	 * Returns the signature generator of this server peer.
	 * If this peer isn't a server, this method returns <code>null</code>.
	 */
	public SignatureGenerator getSignatureGenerator()
	{
		return signatureGenerator;
	}
	
	/**
	 * Returns the signature checker (verifier) of this server peer.
	 * If this peer is a server, this method returns <code>null</code>.
	 */
	public SignatureChecker getSignatureChecker()
	{
		return signatureChecker;
	}
	
	public void requestPublicKey()
	{
		// Public Key direkt vom Sender anfordern
		RemotePeer publicKeyOwner = getServer();
		DatagramSocket udpSocket;
		
		try
		{
			// Eigenes UDP-Socket erstellen
			udpSocket = new DatagramSocket();
			udpSocket.setSoTimeout(JoinThread.JOIN_REQUEST_SOCKET_TIMEOUT);
		}
		catch (Exception e)
		{
			Logger.warning("Peer", "COULD_NOT_CREATE_UDP_SOCKET", e); //$NON-NLS-1$ //$NON-NLS-2$
			return;
		}
		
		DatagramPacket requestPublicKeyPacket = null; 
		try
		{
			requestPublicKeyPacket = PacketFactory.createUDPPacket(new PublicKeyRequestPacket(), publicKeyOwner);
		}
		catch (SocketException e)
		{
			Logger.warning("Peer", "COULD_NOT_CREATE_UDP_PACKET", e); //$NON-NLS-1$ //$NON-NLS-2$
			return;
		}
		
		
		boolean packetReceived = false;
		int connectTryNumber = 0;
		
		byte[] buffer = new byte[PublicKeyPacket.getMaxLength()];
		DatagramPacket inPacket = new DatagramPacket(buffer, buffer.length);
			
		while (!packetReceived && (connectTryNumber <= PacketFactory.UDP_RETRIES))
		{
			try
			{
				Logger.finer("Peer", "Peer.REQUESTING_PUBLIC_KEY"); //$NON-NLS-1$ //$NON-NLS-2$
					
				udpSocket.connect(publicKeyOwner.getSocketAddress());
				udpSocket.send(requestPublicKeyPacket);
				udpSocket.receive(inPacket);
				udpSocket.disconnect();
					
				packetReceived = true;
			}
			catch(SocketTimeoutException e)
			{
				Logger.fine("Peer", "Peer.SOCKET_TIMEOUT_PUBLIC_KEY"); //$NON-NLS-1$ //$NON-NLS-2$
					
				packetReceived = false;
				connectTryNumber++;
			}
			catch(Exception e)
			{
				Logger.fine("Peer", "Peer.PUBLIC_KEY_REQUEST_ERROR", e); //$NON-NLS-1$ //$NON-NLS-2$
									
				break;
			}	
		}
		
		if (!packetReceived)
		{
			Logger.warning("Peer", "Peer.SERVER_DID_NOT_ANSWER_PUBLIC_KEY_REQUEST"); //$NON-NLS-1$ //$NON-NLS-2$
			return;
		}
	
		Packet response = PacketFactory.createPacket(inPacket);
	
		if (response instanceof PublicKeyPacket)
		{
			byte[] publicKey = ((PublicKeyPacket)response).getPublicKey();
			
			if (publicKey.length == 0)
			{
				// Server signiert die Pakete nicht
				// Dummy-Signierer verwenden, sonst wird bei jedem
				// Zuliefererwechsel erneut nach dem Public Key
				// gefragt
				signatureChecker = new SignatureCheckerDummy();
			}
			else
			{
				signatureChecker = new SignatureChecker(publicKey);
			}
		}
		else
		{
			// Nicht erwartetes Paket erhalten
			Logger.fine("Peer", "Peer.PUBLIC_KEY_REQUEST_UNEXPECTED_PACKET_RECEIVED", response); //$NON-NLS-1$ //$NON-NLS-2$
		}
	}
	
	public BroadcastBuffer getBroadcastBuffer()
	{
		if (!isServer)
		{
			return null;
		}
		
		if (broadcastBuffer != null)
		{
			return broadcastBuffer;
		}
		else
		{
			startTime = System.currentTimeMillis();  //jhooks - get current time in milliseconds, used to show how long
			 //The server has been serving and to calculate lag.
			broadcastBuffer = new BroadcastBuffer(buffer, getSocketAddress().getPort(), this);  //jhooks - New Call, pass in the server's time started
			broadcastBuffer.start();
			
			return broadcastBuffer;	
		}
	}
	
	public ListenBuffer getListenBuffer()
	{
		return new ListenBuffer(getBuffer(), ui.getMetadata());
	}
	public long getStartTime()
	{
		return startTime;
	}
	public long Get_Time_Server_Has_Been_Running_At_Startup()
	{
		return Time_Server_Has_Been_Running_At_Startup;
	}
	public boolean getLagIsValid()
	{
		return lagIsValid;
	}
	public boolean getVerifiedBandwidth()
	{
		return verifiedBandwidth;
	}
	public LinkedList getRemoteChildren()
	{
		return remoteChildren;
	}
	public LinkedList getLocalChildren()
	{
		return localChildren;
	}
	
	public UI getUI()
	{
		return ui;
	}
	public SettingsXML getXML()
	{
		return xml;
	}
	public int getPort()
	{
		return port;
	}
	public int getMessageMode()
	{
		return messageMode;
	}
	public void setMessageMode(int mode)
	{
		messageMode = mode;
	}
	public void setStreamMode(int mode)
	{
		streamMode = mode;
	}
	public int getStreamMode()
	{
		return streamMode;
	}
	
	public void setStartTime(long startTime)
	{
		this.startTime = startTime;
	}
	public void setLagIsValid(boolean lagIsValid)
	{
		this.lagIsValid = lagIsValid;
	}
	public void Set_Time_Server_Has_Been_Running_At_Startup(long time)
	{
		Time_Server_Has_Been_Running_At_Startup = time;
	}
	public void setVerifiedBandwidth(boolean verifiedBandwidth)
	{
		this.verifiedBandwidth = verifiedBandwidth;
	}
	public void setMulticastAddress(InetSocketAddress multicastAddress)
	{
		this.multicastAddress = multicastAddress;
	}
	public InetSocketAddress getMulticastAddress()
	{
		return multicastAddress;
	}
	public void setMessageSocket(UniversalSocket socket)
	{
		messageSocket = socket;
	}
	public UniversalSocket getMessageSocket()
	{
		return messageSocket;
	}
	public int getMulticastTTL()
	{
		return multicastTTL;
	}
	public void setMulticastTTL(int ttl)
	{
		multicastTTL = ttl;
	}
	public UDPPeerUpdater getUdpUpdater()
	{
		return udpPeerUpdater;
	}
}