My task was similar to my last assignment but this time I had to do it with UDP instead of TCP. This basically means I had to emulate TCP over UDP.
Multithreading was an interesting problem as I had to simulate TCP's 3-way handshake.
I decided to use encapsulation and break up the file I'm transferring into 512 byte size blocks, except for the last block which is more likely going to be smaller. I then create my message object with the correct segmentID and a bytesToWrite. The bytesToWrite is needed to handle the last block, there is a better way to do this. I just got lazy here. I'll fix that when I have time.
Every segment received is acknowledged, after a 2x second timeout if there is no ACK then the segment is resent. In the event an ACK fails to get through the receiver will get a segment ID that is exactly -1 to the expected ID. In this way I know an ACK has failed and I resend it.
This has been tested locally using the built in fail simulator (2% chance a packet fails) and tested in the wild by asking a friend set the server up in London, England and I downloaded a 14.4mb file to my machine in Galway, Ireland. It took a long time and had 35 failed packets, 19 of which were failed ACKs.
Both the Server and Client use the UDPFileReceiver and UDPFileSender classes.
Server
/*****
* CT326 - Assignment 12 - [email protected] - 09101916
*/
package pkg12;
import java.io.*;
import java.net.*;
public class Server
{
private static int clientID = 1;
private static DatagramSocket serverSocket;
public static void main(String[] args) throws IOException
{
System.out.println("Server started.");
byte[] buffer = new byte[512];
/**
* ASSIGNMENT INSTRUCTION The server should be multi-threaded, and
* have one thread per connection.
*/
serverSocket = new DatagramSocket(8550);
while (true)
{
try
{
DatagramPacket packet = new DatagramPacket(buffer, buffer.length );
serverSocket.receive(packet);
System.out.println("SERVER: Accepted connection.");
System.out.println("SERVER: received"+new String(packet.getData(), 0, packet.getLength()));
//new socket created with random port for thread
DatagramSocket threadSocket = new DatagramSocket();
Thread t = new Thread(new CLIENTConnection(threadSocket, packet, clientID++));
t.start();
} catch (Exception e)
{
System.err.println("Error in connection attempt.");
}
}
}
}
CLIENTConnection thread class
/*****
* CT326 - Assignment 12 - [email protected] - 09101916
*/
package pkg12;
import java.io.File;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.StringTokenizer;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
*
* @author [email protected] - 09101916
*/
public class CLIENTConnection implements Runnable
{
private int clientID;
private byte[] buffer;
private int bytesToReceive;
private DatagramSocket clientSocket;
private DatagramPacket packet, initPacket;
private String userInput, filename, initString;
public CLIENTConnection(DatagramSocket clientSocket, DatagramPacket packet, int clientID) throws IOException
{
this.clientSocket = clientSocket;
this.packet = packet;
this.clientID = clientID;
}
@Override
public void run()
{
try
{
buffer = new byte[618];
System.out.println("THREAD: " + new String(packet.getData(), 0, packet.getLength()));
initString = new String(packet.getData(), 0, packet.getLength());
StringTokenizer t = new StringTokenizer(initString);
userInput = t.nextToken();
filename = t.nextToken();
if (t.hasMoreTokens())
{
bytesToReceive = new Integer(t.nextToken()).intValue();
}
switch (messageType.valueOf(userInput))
{
case put:
//sends a message gets the new port information to the client
send(packet.getAddress(), packet.getPort(), ("OK").getBytes());
//create Object to handle incoming file
new UDPFileReceiver(clientSocket);
break;
case get:
File theFile = new File(filename);
send(packet.getAddress(), packet.getPort(), ("OK").getBytes());
//create object to handle out going file
UDPFileSender fileHandler = new UDPFileSender(clientSocket, packet);
fileHandler.sendFile(theFile);
break;
default:
System.out.println("Incorrect command received.");
break;
}
} catch (IOException ex)
{
Logger.getLogger(CLIENTConnection.class.getName()).log(Level.SEVERE, null, ex);
}
System.out.println("*** Transfer for client " + clientID + " complete. ***");
}
private void send(InetAddress recv, int port, byte[] message) throws IOException
{
DatagramPacket packet = new DatagramPacket(message, message.length, recv, port);
clientSocket.send(packet);
}
private void send(byte[] message) throws IOException
{
DatagramPacket packet = new DatagramPacket(message, message.length);
clientSocket.send(packet);
}
public enum messageType
{
get, put;
}
}
Message class
/*****
* CT326 - Assignment 12 - [email protected] - 09101916
*/
package pkg12;
import java.io.Serializable;
public class Message implements Serializable {
private int segmentID;
private byte[] packet;
private int bytesToWrite;
public Message(){}
public Message(int segmentID, byte[] packet, int bytesToWrite) {
this.segmentID = segmentID;
this.packet = packet;
this.bytesToWrite = bytesToWrite;
}
public int getBytesToWrite() {
return bytesToWrite;
}
public int getSegmentID() {
return segmentID;
}
public byte[] getPacket() {
return packet;
}
}
Client
/*****
* CT326 - Assignment 12 - [email protected] - 09101916
*/
package pkg12;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.StringTokenizer;
/**
*
* @author [email protected] - 09101916
*/
public class Client
{
private static byte[] buffer;
private static int port = 8550;
private static DatagramSocket socket;
private static BufferedReader stdin;
private static StringTokenizer userInput;
private static DatagramPacket initPacket, packet;
public static void main(String[] args) throws IOException
{
socket = new DatagramSocket();
InetAddress address = InetAddress.getByName("localhost");
buffer = new byte[618];
stdin = new BufferedReader(new InputStreamReader(System.in));
String selectedAction = selectAction();
userInput = new StringTokenizer(selectedAction);
try
{
switch (messageType.valueOf(userInput.nextToken()))
{
case put:
packet = new DatagramPacket((selectedAction).getBytes(), (selectedAction).getBytes().length, address, port);
socket.send(packet);
File theFile = new File(userInput.nextToken());
initPacket = receivePacket();
//create object to handle out going file
UDPFileSender fileHandler = new UDPFileSender(socket, initPacket);
fileHandler.sendFile(theFile);
break;
case get:
packet = new DatagramPacket((selectedAction).getBytes(), (selectedAction).getBytes().length, address, 8550);
socket.send(packet);
initPacket = receivePacket();
socket.send(initPacket);
//create Object to handle incoming file
new UDPFileReceiver(socket);
break;
}
} catch (Exception e)
{
System.err.println("not valid input");
}
socket.close();
}
private static DatagramPacket receivePacket() throws IOException
{
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
socket.receive(packet);
return packet;
}
public static String selectAction() throws IOException
{
System.out.println("COMMANDS: get *filename*");
System.out.println("\t put *filename*");
System.out.println("\t example: put data.txt");
System.out.print("ftp> ");
return stdin.readLine();
}
public enum messageType
{
get, put;
}
}
UDPFileSender
/*****
* CT326 - Assignment 12 - [email protected] - 09101916
*/
package pkg12;
import java.io.*;
import java.net.*;
public class UDPFileSender
{
private int segmentID;
private int reSendCount;
private byte[] msg, buffer;
private FileInputStream fileReader;
private DatagramSocket datagramSocket;
private int fileLength, currentPos, bytesRead;
private final int packetOverhead = 106; // packet overhead
public UDPFileSender(DatagramSocket socket, DatagramPacket initPacket) throws IOException
{
msg = new byte[512];
buffer = new byte[512];
datagramSocket = socket;
//setup DatagramSocket with correct Inetaddress and port of receiver
datagramSocket.connect(initPacket.getAddress(), initPacket.getPort());
segmentID = 0;
}
public void sendFile(File theFile) throws IOException
{
fileReader = new FileInputStream(theFile);
fileLength = fileReader.available();
System.out.println("*** Filename: " + theFile.getName() + " ***");
System.out.println("*** Bytes to send: " + fileLength + " ***");
send((theFile.getName() + "::" + fileLength).getBytes());
DatagramPacket reply = new DatagramPacket(buffer, buffer.length);
datagramSocket.receive(reply);
//waits for receiver to indicate OK to send
if (new String(reply.getData(), 0, reply.getLength()).equals("OK"))
{
System.out.println("*** Got OK from receiver - sending the file ***");
//outer while to control when send operation comlete
//inner while to control ACK messages from receiver
while (currentPos < fileLength)
{
bytesRead = fileReader.read(msg);
Message message = new Message(segmentID, msg, bytesRead);
System.out.println("Sending segment " + message.getSegmentID() + " with " + bytesRead + " byte payload.");
byte[] test = serialize(message);
send(test, bytesRead + packetOverhead);
currentPos = currentPos + bytesRead;
//handle ACK of sent message object, timeout of 2 seconds. If segementID ACK is not received
//resend segment.
datagramSocket.setSoTimeout(2000);
boolean receiveACK = false;
while (!receiveACK)
{
try
{
datagramSocket.receive(reply);
} catch (SocketTimeoutException e)
{
send(test, bytesRead + packetOverhead);
System.out.println("*** Sending segment " + message.getSegmentID() + " with " + bytesRead + " payload again. ***");
reSendCount++;
}
if (new String(reply.getData(), 0, reply.getLength()).equals(Integer.toString(message.getSegmentID())))
{
System.out.println("Received ACK to segment" + new String(reply.getData(), 0, reply.getLength()));
segmentID++;
receiveACK = true;
}
}
}
System.out.println("*** File transfer complete...");
System.out.println(reSendCount + " segments had to be resent. ***");
} else
{
System.out.println("Recieved something other than OK... exiting");
}
}
private void send(byte[] message, int length) throws IOException
{
DatagramPacket packet = new DatagramPacket(message, length);
datagramSocket.send(packet);
}
private void send(byte[] message) throws IOException
{
DatagramPacket packet = new DatagramPacket(message, message.length);
datagramSocket.send(packet);
}
public byte[] serialize(Object obj) throws IOException
{
ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
ObjectOutputStream objectStream = new ObjectOutputStream(byteStream);
objectStream.writeObject(obj);
objectStream.flush();
return byteStream.toByteArray();
}
}
UDPFileReceiver
/*****
* CT326 - Assignment 12 - [email protected] - 09101916
*/
package pkg12;
import java.io.*;
import java.net.*;
import java.util.*;
public class UDPFileReceiver
{
private byte[] buffer;
private Message receiveMSG;
private DatagramSocket socket;
private String filename, initString;
private FileOutputStream fileWriter;
private DatagramPacket initPacket, receivedPacket;
private int bytesReceived, bytesToReceive, simulateBadConnection, expectedSegmentID;
private final boolean simulateMessageFail = false;//true if you want to simulate a bad connection
public UDPFileReceiver(DatagramSocket socket) throws IOException
{
this.socket = socket;
buffer = new byte[618]; //618 is the size of my message object. 512 payload, 106 overhead
System.out.println("*** Ready to receive file on port: " + socket.getLocalPort() + " ***");
initPacket = receivePacket();
initString = "Recieved-" + new String(initPacket.getData(), 0, initPacket.getLength());
//get the file name and byte size of file name
StringTokenizer t = new StringTokenizer(initString, "::");
filename = t.nextToken();
bytesToReceive = new Integer(t.nextToken()).intValue();
System.out.println("*** The file will be saved as: " + filename + " ***");
System.out.println("*** Expecting to receive: " + bytesToReceive + " bytes ***");
//tell the sender OK to send data
send(initPacket.getAddress(), initPacket.getPort(), ("OK").getBytes());
fileWriter = new FileOutputStream(filename);
//two while loops. First checks that there is still more data to receive
//and inner do/while is to error check on received packets and catch missing ACK
//sent to sender
while (bytesReceived < bytesToReceive)
{
receiveMSG = new Message();
do
{
receivedPacket = receivePacket();
try
{
receiveMSG = (Message) deserialize(receivedPacket.getData());
} catch (ClassNotFoundException ex)
{
System.out.println("*** Message packet failed. ***");
}
//logically if the last ACK sent fails to be received the UDPSender will resend the last segment.
//A simple check on the segemntID will catch this as it will be equal to expectedID - 1. Resending the ACK
//for this previous segment will eventually get through to the server to send the next expected segment.
if ((expectedSegmentID - 1) == receiveMSG.getSegmentID())
{
String ACK = Integer.toString(receiveMSG.getSegmentID());
send(initPacket.getAddress(), initPacket.getPort(), (ACK).getBytes());
System.out.println("*** Resending ACK for segment " + ACK + " ***");
}
if (simulateMessageFail)
{
simulateBadConnection = (Math.random() < 0.98) ? 0 : 1; //simulate a 2% chance a message object is lost
}
//by adding 1 to segmentIDExpected we can make the receiver determine a message object is lost
//as the server has a 2 second timeout before it will resend we can check the error control in this way.
} while (receiveMSG.getSegmentID() != (expectedSegmentID + simulateBadConnection));
expectedSegmentID++;
//handles the last byte segmentID size .getBytesToWrite()
fileWriter.write(receiveMSG.getPacket(), 0, receiveMSG.getBytesToWrite());
System.out.println("Received segmentID " + receiveMSG.getSegmentID());
//adding payload size for outer while condition
bytesReceived = bytesReceived + 512;
//simulate a 2% chance an ACK is lost
if (simulateMessageFail)
{
if ((Math.random() < 0.98))
{
String ACK = Integer.toString(receiveMSG.getSegmentID());
send(initPacket.getAddress(), initPacket.getPort(), (ACK).getBytes());
} else
{
System.out.println("*** failed to send ACK ***");
}
} else
{
String ACK = Integer.toString(receiveMSG.getSegmentID());
send(initPacket.getAddress(), initPacket.getPort(), (ACK).getBytes());
}
}
System.out.println("*** File transfer complete. ***");
fileWriter.close();
}
private DatagramPacket receivePacket() throws IOException
{
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
socket.receive(packet);
return packet;
}
private void send(InetAddress recv, int port, byte[] message) throws IOException
{
DatagramPacket packet = new DatagramPacket(message, message.length, recv, port);
socket.send(packet);
}
private Object deserialize(byte[] bytes) throws IOException, ClassNotFoundException
{
ByteArrayInputStream byteStream = new ByteArrayInputStream(bytes);
ObjectInputStream objectStream = new ObjectInputStream(byteStream);
return (Message) objectStream.readObject();
}
}
