peer.java

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

		}
		
		Logger.fine("Peer", "Peer.LEAVING"); //$NON-NLS-1$ //$NON-NLS-2$
			
		// Das n鋍hste mal den Strom nicht wiederaufnehmen
		buffer.resetResumeSeqNr();
		
		// Den fremden Public Key l鰏chen
		signatureChecker = null;
		
		
		// Nachricht an die Kinder senden
		
		// In der Nachricht steht auch noch die Adresse des sich verabschiedenden
		// Peers (mit korrekter Portnummer), damit es nicht ganz so einfach ist,
		// diese Nachricht zu f鋖schen
		
		for (int elementNr=0; elementNr < children.size(); elementNr++)
		{
			RemotePeer child = (RemotePeer)children.get(elementNr);
			if (child.isMulticastAddress())
				continue;
			try
			{
				Logger.finer("Peer", "Peer.SAYING_GOODBYE_TO_CHILD", child); //$NON-NLS-1$ //$NON-NLS-2$
			
				// Umleitung des Kinds auf den Zulieferer
				LeaveAndRedirectPacket packet = new LeaveAndRedirectPacket(redirection);
				RemotePeerData childData = getChildData(child);
				
				childData.getSocket().send(packet);
			}
			catch (IOException e)
			{
				Logger.fine("Peer", "IO_ERROR", e); //$NON-NLS-1$ //$NON-NLS-2$
				
				// Nicht zur點kkehren - restliche Kinder benachrichtigen
			}
			finally
			{
				removeChild(child);
			}
		}
	}

	private void sayGoodbyeToSupplier(RemotePeer supplier)
	{
		// Nachricht an den Zulieferer senden
		if (supplier.isMulticastAddress())
			return;
		try
		{
			Logger.finer("Peer", "Peer.SAYING_GOODBYE_TO_SUPPLIER", supplier); //$NON-NLS-1$ //$NON-NLS-2$
			
			LeavePacket packet = new LeavePacket();
			supplierSocket.send(packet);
		}
		catch (IOException e)
		{
			Logger.fine("Peer", "IO_ERROR", e); //$NON-NLS-1$			 //$NON-NLS-2$
		}
	}
	public synchronized boolean canServePeers(int peers)
	{
		int nmaxUpload = (int) maxUploadBandwidth;
		int remoteChildren = this.remoteChildren.size();
		int byterate = ui.getMetadata().getAverageByterate();
		int childrenServe;
		if (nmaxUpload < 0)
		{
			childrenServe = -nmaxUpload;
		}
		else
		{
			//					
			childrenServe = (int)(maxUploadBandwidth / ((double)byterate / 1024));
		}
		childrenServe -= remoteChildren;
		return childrenServe >= peers;
		
		// "+1", weil ein Kind dazukommen will
		//      bandwidth of current children + 1 additional            
	}
	public synchronized boolean canServeAnotherPeer()
	{
		return canServePeers(1);
	}
	
	public synchronized void addChild(RemotePeer peer) throws PeerException
	{
		if (equals(peer) || (!isServer && server.equals(peer)))
		{
			throw new IllegalArgumentException();
		}
		
		if (isChild(peer))
		{
			throw new PeerException(Messages.getString("Peer.PEER_IS_ALREADY_CHILD", peer)); //$NON-NLS-1$
		}

		children.add(peer);

		RemotePeerData remotePeerData = new RemotePeerData();
		remotePeerData.setTimeToConnect(System.currentTimeMillis() + StreamDispatcherForChildren.TIME_FOR_CONNECTION_ESTABLISHMENT);

		childrenData.put(peer, remotePeerData);
		
		Logger.finer("Peer", "Peer.NEW_PROVISIONAL_CHILD", peer); //$NON-NLS-1$ //$NON-NLS-2$
	}
	public void addLocalChild (RemotePeer peer)
	{
		addToList(peer, localChildren);
	}
	public boolean isLocalChild(RemotePeer peer)
	{
		Iterator iter = remoteChildren.iterator();
		while (iter.hasNext())
		{
			if (peer.equals(iter.next()))
				return true;
		}
		return false;
	}
	public void addRemoteChild(RemotePeer peer) 
	{
		addToList(peer, remoteChildren);
	}
	private void addToList(RemotePeer peer, List list)
	{
		Iterator iter = list.iterator();
		int i = 0;
		while (iter.hasNext())
		{	
			if (peer.equals((RemotePeer)iter.next()))
				return;
		}
		list.add(peer);
	}
	
	public synchronized boolean isChild(RemotePeer peer)
	{
		return children.contains(peer);
	}
	
	public synchronized void removeChild(RemotePeer child)
	{
		if (!children.contains(child))
		{
			throw new IllegalArgumentException(Messages.getString("Peer.PEER_IS_NO_CHILD", child)); //$NON-NLS-1$
		}
		
		RemotePeerData data = getChildData(child);
		
		// Das muss vorgezogen werden, damit die Threads merken,
		// dass dieses Kind gegangen ist
		data.disconnect();
		childrenData.remove(child);
		children.remove(child);
		localChildren.remove(child);
		remoteChildren.remove(child);
		if (data.hasBeenConnected())
		{		
			try
			{
				//System.out.println("peer1");
				data.getSocket().close();
			}
			catch (IOException e)
			{
			}
		}
		
		Logger.finer("Peer", "Peer.CHILD_REMOVED", child); //$NON-NLS-1$ //$NON-NLS-2$
	}
	
	// Diese Methode darf nur verwendet werden, wenn der
	// Peer sich "fies" verh鋖t
	public synchronized void removeAllChildren()
	{
		for (int i=0; i < children.size(); i++)
		{
			removeChild((RemotePeer)children.get(i));
		}
	}
	
	public synchronized RemotePeer getRedirectionChild() throws PeerException
	{
		if (remoteChildren.size() == 0)
		{
			throw new PeerException(Messages.getString("Peer.NO_CHILD_AVAILABLE")); //$NON-NLS-1$
		}
		else
		{
			// Zuf鋖liges Kind zur點kgeben
			return (RemotePeer)remoteChildren.get(random.nextInt(remoteChildren.size()));
		}
	}
	
	public synchronized RemotePeer getRedirectionChild(RemotePeer notThisPeer) throws PeerException
	{
		Vector allowedChildren = (Vector)remoteChildren.clone();
		allowedChildren.remove(notThisPeer);
		
		if (allowedChildren.size() == 0)
		{
			throw new PeerException(Messages.getString("Peer.NO_CHILD_AVAILABLE")); //$NON-NLS-1$
		}
		else
		{
			// Zuf鋖liges Kind zur點kgeben
			return (RemotePeer)allowedChildren.get(random.nextInt(allowedChildren.size()));
		}
	}
		
	public void finalize()
	{
		if (isServer)
		{
			Logger.fine("Peer", "Peer.SERVER_SHUTDOWN"); //$NON-NLS-1$ //$NON-NLS-2$
		}
		
	}
	
	/**
	 * Returns whether this peer is a server/source.
	 * 
	 * @return <code>true</code>, if this peer is a server, <code>false</code> otherwise
	 */
	public boolean isServer()
	{
		return isServer;
	}
	
	/**
	 * 
	 * Returns the server of the P2P network this peer is connected to.
	 * 
	 * @return The server of the P2P network, or <code>null</code> if this peer is a server itself
	 */
	public RemotePeer getServer()
	{
		return server;
	}

	/**
	 * Changes the supplier. If the supplier is missing, this method has no effect.
	 *   
	 * @param informSupplier Whether the old supplier should be informed
	 * @param reportFreeloader Whether the old supplier should be reported as freeloader
	 */
	public synchronized void changeSupplier(boolean informSupplier, boolean reportFreeloader)
	{
		changeSupplier(null, informSupplier, reportFreeloader);
	}

	public synchronized void changeSupplier(RemotePeer startPeer, boolean informSupplier, boolean reportFreeloader)
	{
		// Ist gar kein Zulieferer da?
		if (supplier == null)
		{
			return;
		}
		
		// Zulieferer entfernen
		removeSupplier(informSupplier);
		
		// Freeloader melden
		if (reportFreeloader)
		{
			reportFreeloader();
		}
		
		// Kinder benachrichtigen
		getBuffer().put(new StandByPacket());
		
		
		// Neuen Zulieferer suchen
		if (startPeer != null)
		{
			join(startPeer);
		}
		else
		{
			join();
		}
	}
	
	public synchronized void removeSupplier(boolean informSupplier)
	{
		if (isServer || (supplier == null))
		{
			// Ein Server braucht keinen Zulieferer
			// oder es gibt gar keinen Zulieferer
			Logger.warning("Peer", "Peer.INVALID_CALL_OF_REMOVESUPPLIER"); //$NON-NLS-1$ //$NON-NLS-2$
			return;
		}
		
		Logger.finer("Peer", "Peer.LOST_SUPPLIER"); //$NON-NLS-1$ //$NON-NLS-2$
			
	
		// Der Zulieferer trennt die Verbindung sofort, falls er benachrichtigt
		// wird. Daher muss "supplier" auf null gesetzt werden, damit die Threads
		// dieses Peers erkennen k鰊nen, dass der Zulieferer weg ist
		
		RemotePeer oldSupplier = supplier;
		supplier = null;
		
		// Zulieferer-Thread beenden
		streamDispatcherForSupplier.shutdown();
		
		if (informSupplier)
		{
			// Vom Zulieferer verabschieden
			sayGoodbyeToSupplier(oldSupplier);
		}
	
				
		if (supplierSocket != null)
		{
			try
			{
				//System.out.println("peer2");
				supplierSocket.close();
			}
			catch (IOException e)
			{
			}
		}
		
		// Alles M鰃liche auf "null" setzen
		// Schon gemacht: supplier = null;
		supplierFather = null;
		supplierSocket = null;
	}
	
	// Diese Methode NICHT synchronized machen (Deadlock)!
	public void waitUntilSupplierDispatcherDies()
	{
		if (streamDispatcherForSupplier != null)
		{
			try
			{	
				streamDispatcherForSupplier.join();
			}
			catch (InterruptedException e)
			{
			}
		}
	}
	
	public synchronized void addSupplier(RemotePeer supplier, RemotePeer supplierFather, UniversalSocket supplierSocket)
	{		
		if (isServer || (this.supplier != null))
		{
			// Ein Server braucht keinen Zulieferer
			// oder es gibt schon einen Zulieferer
			Logger.warning("Peer", "Peer.INVALID_CALL_OF_ADDSUPPLIER"); //$NON-NLS-1$ //$NON-NLS-2$
			return;
		}
		
		wasOnceConnected = true;
		
		this.supplier = supplier;
		this.supplierFather = supplierFather;
		this.supplierSocket = supplierSocket;
		
		// getMonitor().report(Events.NEW_SUPPLIER, supplier);
		
		// Den Public Key anfordern, falls dieser Peer ihn noch nicht hat
		
		//TODO - jhooks - look over verification
		
		if (verifyStreamPackets && (signatureChecker == null))
		{
			requestPublicKey();
		}
		
		// Dispatcher f黵 den Zulieferer starten
		streamDispatcherForSupplier = new StreamDispatcherForSupplier(this, supplier, supplierSocket);
		streamDispatcherForSupplier.start();
		
		// Falls der UDP-Dispatcher noch nicht l鋟ft: Ihn starten
		
		//jhooks - Start our Listener objects to serve other peers
		if ((listener1 == null && listener2 == null) || (!listener1.isAlive() && !listener2.isAlive()))
			createListeners();
	}
	

	/**
	 * Returns the current supplier.
	 * 
	 * @return The current supplier, or <code>null</code> if it's missing.
	 */
	public RemotePeer getSupplier()
	{
		return supplier;
	}
	
	
	/**
	 * Returns a vector of all children. The vector will contain {@link RemotePeer} objects.
	 * 
	 * @return A vector of all children
	 */
	public Vector getChildren()
	{
		return children;
	}
	
	/**
	 * Returns a vector of all banned peers. The vector will contain {@link RemotePeer} objects.
	 *
	 * @return A vector of all banned peers
	 */
	public Vector getBannedPeers()
	{
		return bannedPeers;
	}
	
	public synchronized void processComplainAboutFreeloader(RemotePeer sender, RemotePeer freeloader) throws PeerException
	{
		// Sollte man nachschauen, ob sich der Sender schon mal bei diesem
		// Peer anmelden wollte?
				
		if (!isChild(freeloader) || sender.equals(freeloader) || freeloader.isMulticastAddress())
		{
			// Da will uns jemand ver鋚peln
			throw new IllegalArgumentException(Messages.getString("Peer.INVALID_COMPLAINT")); //$NON-NLS-1$
		}
		
		if (canServeAnotherPeer())
		{
			// Solange noch Kinder aufgenommen werden k鰊nen werden -- As long as children can be accepted, there are no complaints 
			// keine Beschwerden akzeptiert
			return;
		}
		
		RemotePeerData data = getChildData(freeloader);
		data.addFreeloaderComplainant(sender);
		
		if (data.getFreeloaderComplaints() >= MAX_FREELOADER_COMPLAINTS)
		{
			bannedPeers.add(freeloader);
					
			// Nachricht an den Freeloader senden
			Logger.fine("Peer", "Peer.REDIRECTING_FREELOADER", freeloader); //$NON-NLS-1$ //$NON-NLS-2$
			
			RemotePeer redirection = null;
			try
			{
				redirection = getRedirectionChild(freeloader);
			}
			catch (PeerException e)
			{
				// Es ist kein Kind verf黦bar
				// In diesem Fall wird der Freeloader zu jenem Peer umgeleitet,
				// der die Beschwerde gesendet hat
				redirection = sender;
			}
				
			FreeloaderRedirectPacket packet = new FreeloaderRedirectPacket(redirection);
			
			RemotePeerData childData = getChildData(freeloader);