universalsocket.java

来自「java语言开发的P2P流媒体系统」· Java 代码 · 共 361 行

/* Stream-2-Stream - Peer to peer television and radio
 * Project homepage: http://s2s.sourceforge.net/
 * Copyright (C) 2005-2006 Jason Hooks
 * ---------------------------------------------------------------------------
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 * ---------------------------------------------------------------------------
 */
package stream2stream.network;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.*;

import p2pradio.logging.Logger;
import p2pradio.packets.Packet;
import p2pradio.packets.PacketFactory;

/**
 * This class is a wrapper for 3 different kinds of sockets (TCP unicast, UDP unicast, UDP multicast).
 * By using this class we can code in a transfer independent way.   
 * This means you will only need to change the network code once, 
 * rather than changing for each transport type.
 * 
 * @author jhooks
 *
 */

public class UniversalSocket {
	public static final int TCP = 0;
	//Datagrams
	public static final int UDP = 1;
	
	private int mode;
	private byte[] buffer;
	private Socket tcpSocket;
	private DatagramSocket udpSocket;
	private DatagramPacket udpPacket;
	private SocketAddress address;
	private DataInputStream inputStream;
	private DataOutputStream outputStream;
	private Packet initialPacket;
	public UniversalSocket(Packet initialPacket, int mode) throws SocketException, IOException
	{
		this(mode);
		this.initialPacket = initialPacket;
		
	}
	public UniversalSocket(int localport, int mode) throws SocketException, IOException
	{
		this(mode);
		if (mode == UDP)
			udpSocket = new DatagramSocket(localport);
		else if (mode == TCP)
		{
			tcpSocket = new Socket();
			tcpSocket.bind(new InetSocketAddress(localport));
		}
			
	}
	public UniversalSocket(int mode) throws SocketException, IOException
	{
		this.mode = mode;
		if (mode == TCP)
			tcpSocket = new Socket();
		else if (mode == UDP)
			udpSocket = new DatagramSocket();
		
	}
	public UniversalSocket(Socket tcpSocket)
	{
		this.tcpSocket = tcpSocket;
		mode = TCP;
	}
	public UniversalSocket(DatagramSocket udpSocket)
	{
		this.udpSocket = udpSocket;
		mode = UDP;
	}
	public UniversalSocket(Socket tcpSocket, DatagramSocket udpSocket, int mode)
	{
		this.tcpSocket = tcpSocket;
		this.udpSocket = udpSocket;
		this.mode = mode;
	}
	
	public void setBuffer(byte[] buffer) 
	{
		this.buffer = buffer;
		if (mode == UDP)
			udpPacket = new DatagramPacket(buffer, buffer.length);
	}
	public void setSoTimeout(int timeout) throws IOException
	{
		if (tcpSocket != null)
			tcpSocket.setSoTimeout(timeout);
		if (udpSocket != null) 
			udpSocket.setSoTimeout(timeout);
	}
	//Do not change to void receive(packet), address does not get passed back (packet is null)
	public Packet receive() throws IOException
	{
		if (initialPacket != null)
		{
			Packet returnPacket = initialPacket;
			initialPacket = null;
			//if (returnPacket == null)
				//System.out.println("CLEARED");
			return returnPacket;
		}
		Packet packet = null;
		if (mode == TCP)
		{
			if (inputStream == null)
				inputStream = new DataInputStream(tcpSocket.getInputStream());
			int packetLength = 0;
			packetLength = inputStream.readInt();
			
			if ((packetLength < 0) || (packetLength > PacketFactory.STREAM_PACKET_MAX_SIZE))
			{
				Logger.fine("TCPDispatcherForSupplier", "MESSAGE_WITH_WRONG_SIZE_INFORMATION_RECEIVED"); //$NON-NLS-1$ //$NON-NLS-2$
				return null;
			}
			byte[] buffer = new byte[packetLength];
			while (packet == null)
			{
				inputStream.readFully(buffer);
				packet = PacketFactory.createPacket(buffer, packetLength);
			}
		}
		else //if (mode == UDP || mode == MULTICAST)
		{
			
			udpPacket = new DatagramPacket(buffer, buffer.length);
			udpSocket.receive(udpPacket);
			byte[] data = udpPacket.getData();
			int subLength = udpPacket.getLength();
			if (data[0] == Packet.UDP_STREAM)
			{
				//System.out.println("UDPSTREAM");
				
				
				//Get the length of the message
				byte[] blength = new byte[4];
				System.arraycopy(data, 1, blength, 0, 4);
				ByteArrayInputStream byte_in = new ByteArrayInputStream (blength); 
		 		DataInputStream data_in = new DataInputStream (byte_in);
				int length = data_in.readInt(); 
				byte[] streamInfo = new byte[length];
				
				
				//Get the data from the first datagram
				int offset = subLength - 5;
				System.arraycopy(data, 5, streamInfo, 0, offset);
				//System.out.println("LENGTH:" + length + "BUFFER:" + subLength + "OFFSET:" + offset);
				//					2853					2000					1995
				//Length:853
				//5
				
				//
				int udpPacketLength;
				//Write the rest of the datagrams
				//length = 512 offset = 507
				while ((udpPacketLength = length - offset) > 0)
				{
					udpSocket.receive(udpPacket);
					if (udpPacketLength > udpPacket.getLength())
						udpPacketLength = udpPacket.getLength();
					//System.out.println("Length:" + udpPacketLength);
					data = udpPacket.getData();
					
					System.arraycopy(data, 0, streamInfo, offset, udpPacketLength);
					offset += udpPacketLength;
				}
				packet = PacketFactory.createPacket(streamInfo, length);
			    
			}
			else
			{
				byte[] streamInfo = new byte[subLength];
				System.arraycopy(data, 0, streamInfo, 0, subLength);
				packet = PacketFactory.createPacket(streamInfo, streamInfo.length);
			}
		}
		return packet;
	}
	public void send(Packet packet) throws IOException
	{
		if (mode == TCP)
		{
			if (outputStream == null)
				outputStream = new DataOutputStream(new BufferedOutputStream(tcpSocket.getOutputStream(), buffer.length));
			packet.send(outputStream);
		}
		else
		{
			DatagramPacket udpPacket;
			//System.out.println("PacketLength" + packet.getContentLength());
			if (packet.getContentLength() <= buffer.length )
			{
				//System.out.println("Smaller");
				udpPacket = PacketFactory.createUDPPacket(packet, address);
				udpSocket.send(udpPacket);
			}
			//If the packet is bigger than the buffer (datagram size), split into multiple datagrams 
			else
			{
				//System.out.println("BIGGER");
				int subLength = buffer.length;
				
				byte[] content = packet.getContent();
				byte[] subContent = new byte[subLength];
				int length = packet.getContentLength();
				
				
				//Write a byte to indicate that the message will be split into multiple datagrams
				subContent[0] = Packet.UDP_STREAM;
				ByteArrayOutputStream byte_out = new ByteArrayOutputStream (); 
		 		DataOutputStream data_out = new DataOutputStream (byte_out);
		 		
		 		//Write the length of the message onto the first datagram
		 		data_out.writeInt(length);
		 		System.arraycopy(byte_out.toByteArray(), 0, subContent, 1, 4);
		 		//Write part of the message on the first datagram
		 		
				System.arraycopy(content, 0, subContent, 5, subLength - 5);
				int offset = subLength;
				
				//Send the first datagram
				udpPacket = new DatagramPacket(subContent, subLength, address);
				udpSocket.send(udpPacket);
				int udpPacketLength;
				
				//Write the rest of the datagrams
				while ((udpPacketLength = length - offset + 5) > 0)
				{
					
					if (udpPacketLength > subLength)
						udpPacketLength = subLength;
					//System.out.println("ServerLength" + udpPacketLength);
					subContent = new byte [udpPacketLength];
					
					System.arraycopy(content, offset, subContent, 0, udpPacketLength);
					udpPacket = new DatagramPacket(subContent, udpPacketLength, address);
					udpSocket.send(udpPacket);
					offset += udpPacketLength;
				}
				
				
				
				
				
				
			}
		}
	}
	public void close() throws IOException
	{
		//System.out.println("A socket is being closed");
		if (inputStream != null)
			inputStream.close();
		if (outputStream != null)
			outputStream.close();
		if (tcpSocket != null)
			tcpSocket.close(); 
		if (udpSocket != null) 
			udpSocket.close();
	}
	public void bind(SocketAddress a) throws IOException
	{
		if (mode == TCP)
			tcpSocket.bind(a);
		else //if (mode == UDP || mode == MULTICAST)
			udpSocket.bind(a);
	}
	public void setMode(int mode)
	{
		if (this.mode == mode)
			return;
		this.mode = mode;
	}
	public void setTcpSocket(Socket tcpSocket)
	{
		this.tcpSocket = tcpSocket;
	}
	public void connect(SocketAddress endpoint) throws IOException
	{
		address = endpoint;
		if (mode == TCP)
			tcpSocket.connect(endpoint);
		//if (udpSocket != null)
			//udpSocket.connect(addr);
		
	}
	public void connect(SocketAddress endpoint, int timeout) throws IOException
	{
		address = endpoint;
		if (tcpSocket != null && !tcpSocket.isConnected())
			tcpSocket.connect(endpoint, timeout);
		
	}
	public void disconnect()
	{
		if (mode == UDP)
			udpSocket.disconnect();
	}
	public int getMode()
	{
		return mode;
	}
	public InetAddress getInetAddress()
	{
		if (mode == TCP)
			return tcpSocket.getInetAddress();
		else if (mode == UDP)
			return udpSocket.getInetAddress();
		return null;
	}
	public SocketAddress getRemoteSocketAddress()
	{
		if (mode == TCP)
			return tcpSocket.getRemoteSocketAddress();
		else if (mode == UDP)
			return address;
		return null;
	}
	public void setReceiveBufferSize (int size) throws SocketException
	{
		if (mode == UDP)
			udpSocket.setReceiveBufferSize(size);
	}
	public int getLocalPort()
	{
		if (mode == TCP)
			return tcpSocket.getLocalPort();
		return udpSocket.getLocalPort();
	}
	public void setReceivedAddress() throws IOException, SocketException
	{
		if (udpPacket == null)
			throw new SocketException();
		connect(udpPacket.getSocketAddress());
	}
	
}