jointhread.java

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

					packetReceived = true;
				}
				catch(SocketTimeoutException e)
				{
					if (shutdown)
						return;
					Logger.fine("JoinThread", "JoinThread.SOCKET_TIMEOUT_JOIN", possibleSupplier); //$NON-NLS-1$ //$NON-NLS-2$
					if (shutdown)
						return;
					
					packetReceived = false;
					connectTryNumber++;
				}
				catch(Exception e)
				{
					if (shutdown)
						return;
					Logger.fine("JoinThread", "JoinThread.REQUEST_ERROR", e); //$NON-NLS-1$ //$NON-NLS-2$
					if (shutdown)
						return;
					break;
				}	
			}
			}
			if (shutdown)
				return;
	
			if (!packetReceived)
			{
				// peer.getMonitor().report(Events.JOIN_NO_ANSWER, possibleSupplier);
				
				// Freeloader melden
				if (possibleSupplierFather != null)
				{
					if (shutdown)
						return;
					peer.reportFreeloader(possibleSupplier, possibleSupplierFather);
					if (shutdown)
						return;
				}
				
				Logger.finer("JoinThread", "JoinThread.RESTARTING_AT_SOURCE"); //$NON-NLS-1$ //$NON-NLS-2$
				
				//Start over at the top of the tree, 1 traversion has been used
				treeTraversionNumber++;
				possibleSupplier = rootPeer;
				possibleSupplierFather = null;
				alreadySeenPeers.clear();
				
			
				continue;
			}
			
			// debugMessage("L鋘ge der Paketdaten: " + inPacket.getLength());
			
			
			
			if (inPacket instanceof OKPacket)
			{
				try
				{
					
					String errorMsg = null;
					
					try
					{
						errorMsg = "JoinThread.COULD_NOT_ESTABLISH_TCP_CONNECTION"; //$NON-NLS-1$
						
						if (shutdown)
							return;
						int mode = peer.getStreamMode();
						if (mode == SettingsXML.MULTICAST_PLUS)
							mode = SettingsXML.UDP;
						supplierSocket = new UniversalSocket(mode);
						// Zum m鰃lichen Zulieferer verbinden
						supplierSocket.connect(possibleSupplier.getSocketAddress(), StreamDispatcherForChildren.TIME_FOR_CONNECTION_ESTABLISHMENT);
						if (shutdown)
							return;
						
						errorMsg = "COULD_NOT_SET_SOCKET_TIMEOUT"; //$NON-NLS-1$
									
						if (shutdown)
							return;
						// Timeout setzen
						supplierSocket.setSoTimeout(StreamDispatcherForSupplier.SUPPLIER_CONNECTION_TIMEOUT);
						supplierSocket.setBuffer(new byte[PacketFactory.UDP_MAX_DATAGRAM_SIZE]);
						//supplierSocket.setReceiveBufferSize(65509);
						
						errorMsg = "COULD_NOT_CREATE_TCP_STREAM"; //$NON-NLS-1$
												
						// Str鰉e erstellen
										
						
						errorMsg = "JoinThread.COULD_NOT_SEND_IDENTIFICATION_PACKET"; //$NON-NLS-1$
						
						// Identifizierungs-Paket senden

						IdentificationPacket identificationPacket = null;
						
						// Warten, bis der Zulieferer-Thread beendet ist.
						// Sehr wichtig! Sonst kann es sein, dass die resumeSeqNr
						// bestimmt wird, aber nachher noch ein Paket vom Thread
						// in den Puffer eingef黦t wird. Ist wirklich vorgekommen!
						if (shutdown)
							return;
						peer.waitUntilSupplierDispatcherDies();
						if (shutdown)
							return;

						// resumeSeqNr bestimmen						 
						if (peer.getBuffer().isResumeSeqNrValid())
						{
							long resumeSeqNr = peer.getBuffer().getResumeSeqNr() + 1;
							long headerSeqNr = peer.getBuffer().getNewestHeaderPacket().getSeqNr();
							long metadataSeqNr;
							
							// Es ist m鰃lich, dass noch gar kein Metadaten-Paket empfangen wurde
							MetadataPacket metadataPacket = peer.getBuffer().getNewestMetadataPacket();
							if (metadataPacket != null)
							{
								metadataSeqNr = metadataPacket.getSeqNr();
							}
							else
							{
								metadataSeqNr = 0;
							}
							
							identificationPacket = new IdentificationSeqNrPacket(peer.getSocketAddress().getPort(), resumeSeqNr, headerSeqNr, metadataSeqNr);
						}
						else
						{
							identificationPacket = new IdentificationPacket(peer.getSocketAddress().getPort());
						}
						if (shutdown)
							return;
						supplierSocket.send(identificationPacket);
						if (shutdown)
							return;
						
					
						// Ein Zulieferer wurde gefunden  
						//A supplier was found
						supplier = possibleSupplier;
						supplierFather = possibleSupplierFather;
						
						if (shutdown)
							return;
						Logger.finer("JoinThread", "JoinThread.NEW_SUPPLIER_FOUND", supplier); //$NON-NLS-1$		 //$NON-NLS-2$
						if (shutdown)
							return;
						
						if (inPacket instanceof OKUseMonitorPacket)
						{
							//jhooks - It seems that in addition to the server, each individual peer must choose whether or not they want to
							//be "monitored".
							if (Radio.enableMonitor)
							{ 
		 						InetSocketAddress monitorAddress = ((OKUseMonitorPacket)inPacket).getMonitorAddress();
								peer.getMonitorLogHandler().setMonitorAddress(monitorAddress);
							}
							else
							{
								if (shutdown)
									return;
								Logger.fine("JoinThread", "JoinThread.MONITOR_ADDRESS_IGNORED"); //$NON-NLS-1$ //$NON-NLS-2$
								if (shutdown)
									return;
							}
						}
					}
					catch (Exception e)
					{
						if (shutdown)
							return;
						Logger.fine("JoinThread", errorMsg, e); //$NON-NLS-1$
						if (shutdown)
							return;
						Logger.fine("JoinThread", "JoinThread.JOIN_FAILED", possibleSupplier); //$NON-NLS-1$ //$NON-NLS-2$
						if (shutdown)
							return;
												
						// M鰃lichen Freeloader melden
						if (possibleSupplierFather != null)
						{
							if (shutdown)
								return;
							peer.reportFreeloader(possibleSupplier, possibleSupplierFather);
							if (shutdown)
								return;
						}
					 	
					 	treeTraversionNumber++;
						possibleSupplier = rootPeer;
						possibleSupplierFather = null;
						alreadySeenPeers.clear();
							
						continue;
					}
				}
				finally
				{
					if (supplier == null)
					{
						// Aufr鋟men
						try
						{
							socket.close();
						}
						catch (IOException e)
						{
						}
					}
				}
			}
			else if (inPacket instanceof RedirectPacket)
		    {
				// Umleitung auf einen anderen Zulieferer
				RemotePeer Old_Possible_Supplier_Father = possibleSupplierFather; //jhooks - NEW CODE
				possibleSupplierFather = possibleSupplier;
				//jhooks - New peer that possibleSupplierFather hinted us to try next
				//We will talk to this peer in the next loop cycle if we manage to get past the two following if statements
				possibleSupplier = ((RedirectPacket)inPacket).getRedirection();
		 
				if (peer.getServer().equals(possibleSupplier))
				{
					// Der andere Peer will bescheissen
					// und verweist auf die Quelle
		 
					// M鰃lichen Freeloader melden
					if (possibleSupplierFather != null)
					{
						if (shutdown)
							return;
						peer.reportFreeloader(possibleSupplierFather, Old_Possible_Supplier_Father); 
						if (shutdown)
							return;
					}
					if (shutdown)
						return;
					Logger.fine("JoinThread", "JoinThread.REDIRECTED_TO_SOURCE"); //$NON-NLS-1$ //$NON-NLS-2$
					if (shutdown)
						return;
					treeTraversionNumber++;
					possibleSupplier = rootPeer;
					possibleSupplierFather = null;
					alreadySeenPeers.clear();
					continue;
				}

				if (alreadySeenPeers.contains(possibleSupplier))
				{
					// Es ist ein Zyklus entstanden, nochmals von
					// vorne beginnen. Aber keinen Freeloader melden,
					// denn es ist nicht klar, wer denn der Schuldige war
					if (shutdown)
						return;
					Logger.fine("JoinThread", "JoinThread.CYCLE_DETECTED"); //$NON-NLS-1$ //$NON-NLS-2$
					if (shutdown)
						return;
					
					treeTraversionNumber++;
					possibleSupplier = rootPeer;
					possibleSupplierFather = null;
					alreadySeenPeers.clear();
			
					continue;
				}
				else
				{
					alreadySeenPeers.add(possibleSupplierFather);
				}
			}
			else
			{	
				// Nicht erwartetes Paket erhalten
				Logger.fine("JoinThread", "UNEXPECTED_PACKET_RECEIVED_FROM_PEER", new Object[]{possibleSupplier, inPacket}); //$NON-NLS-1$ //$NON-NLS-2$
				
				// M鰃lichen Freeloader melden
				if (possibleSupplierFather != null)
				{
					if (shutdown)
						return;
					peer.reportFreeloader(possibleSupplier, possibleSupplierFather);
					if (shutdown)
						return;
				}
					
				treeTraversionNumber++;
				possibleSupplier = rootPeer;
				possibleSupplierFather = null;
				alreadySeenPeers.clear();
							
				continue;
			}
		}
	}
	public void shutdown()
	{
		shutdown = true;
		if (socket != null)
		{
			try {
				socket.close();
			} catch (IOException e) {
			}
		}
		if (supplierSocket != null)
		{
			try {
				socket.close();
			}
			catch (IOException e){
			}
		}
	}
}