import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class cdht {

	private static String selfID;
	private static String stringFirstSuc;
	private static String stringSecondSuc;
	private static int port;
	private static int firstSuc;
	private static int secondSuc;
	private static int firstPort;
	private static int secondPort;
	private static int firstPred = 0;
	private static int secondPred = 0;
	private static int tmpPred = 0;
	private static int firstPredPort;
	private static int secondPredPort;
	private static int playDead = 0;
	public  Runnable step3Server;
	public  Runnable step3Client;

	public cdht(String[] args) {
		selfID = args[0];
		stringFirstSuc = args[1];
		stringSecondSuc = args[2];
		port = 50000 + Integer.parseInt(selfID);
		firstSuc = Integer.parseInt(stringFirstSuc);
		secondSuc = Integer.parseInt(stringSecondSuc);
		firstPort = firstSuc + 50000;
		secondPort = secondSuc + 50000;


		//=========================STEP 3===================================================
		//Server side
		//to run the sub class
		final ExecutorService serverProcessingPool = Executors.newFixedThreadPool(10);
		step3Server = new Runnable() {
		public void run() {
			try {
				// create server socket
				ServerSocket welcomeSocket = new ServerSocket(port);
				while(true) {
				    // accept connection from client
				    Socket connectionSocket = welcomeSocket.accept();
				    //launch class in new thread
				    serverProcessingPool.submit(new ServerTask(connectionSocket));
		    	}
			} catch (IOException e){
				System.out.println(e.getMessage());
			}
		}
	};
		//Client side
		step3Client = new Runnable() {
			public void run() {
				while(true) {
					try {
						String serverName = "localhost";
						InetAddress IPAddress = InetAddress.getByName(serverName);
						// create socket which connects to server
						Socket clientSocket = new Socket(IPAddress, firstPort);
						// get input from keyboard
						BufferedReader inFromUser =
						new BufferedReader(new InputStreamReader(System.in));
						//thread blocked here, waiting to read in input
						String line = inFromUser.readLine();
						String[] linePart = line.split(" ");
						if (linePart[0].equals("request")) {
							String file = linePart[1];
							// write to server
							DataOutputStream outToServer = new DataOutputStream(clientSocket.getOutputStream());
							outToServer.writeBytes(selfID + "-" + file+ "-" + selfID); //selfID so the final peer knows who to reply directly to
							System.out.println("File request message for " + file +  " has been sent to my successor.");
							clientSocket.close();
						} else if(linePart[0].equals("quit")){
							int playDead = 1; //mark the peer as quit
							//update first and second ports
							firstPredPort = firstPred + 50000;
							secondPredPort = secondPred + 50000;
							//start connection with predecessors
							Socket firstPredSocket = new Socket(IPAddress, firstPredPort);
							Socket secondPredSocket = new Socket(IPAddress, secondPredPort);
							//tell predecessors you are leaving and their new successors
							DataOutputStream alertFirst = new DataOutputStream(firstPredSocket.getOutputStream());
							alertFirst.writeBytes(selfID + "-"+ secondSuc);
							DataOutputStream alertSecond = new DataOutputStream(secondPredSocket.getOutputStream());
							alertSecond.writeBytes(selfID + "-"+ firstSuc);
							//close both sockets
							firstPredSocket.close();
							secondPredSocket.close();
						}
			    	} catch (IOException e) {
			    		System.out.println(e.getMessage());
					}
				}
			}
		};
	}

	//private class that will spit the server thread
	//Only used by the server from step 3
	private class ServerTask implements Runnable {
        private final Socket connectionSocket;
        private ServerTask(Socket connectionSocket) {
            this.connectionSocket = connectionSocket;
        }
        @Override
        public void run() {
        	try {
	            InetAddress IPAddress = InetAddress.getByName("localhost");
	            // create read stream to get input
			    BufferedReader inFromClient = new BufferedReader(new InputStreamReader(connectionSocket.getInputStream()));
			    String objectFile;
			    objectFile = inFromClient.readLine();
		    	String[] parts = objectFile.split("-");
		    	if(parts.length == 3){
				    String returnPeer = parts[0]; // peer whom to reply directly to
				    String fileName = parts[1]; // file name
				    String pred = parts[2]; //predecessor
				    int hash = Integer.parseInt(fileName) % 256;

				    //case when returnPeer receives an acknowledgement
				    if(returnPeer.equals(selfID)){
			    		System.out.println("Received a response message from peer "+pred+", which has the file "+fileName+".");
				    }

					//takes the lowest number peer into account when highest peer meets lowest peer
					// ie (if predecessor is higher selfID)
				    else if(
							//for files within peer circle (excuding the lowest)
				    		(hash <= Integer.parseInt(selfID)&& hash > Integer.parseInt(pred))
				    		//for files larger than largest peer
				    		||((Integer.parseInt(pred)>Integer.parseInt(selfID))&& hash > Integer.parseInt(pred))
				    		//for files less than or equal to lowest peer
				    		|| ((Integer.parseInt(pred)>Integer.parseInt(selfID))&& hash <= Integer.parseInt(selfID))) {

				    	System.out.println("File "+fileName+" is here.");
				    	System.out.println("A response message, destined for peer " + returnPeer + ", has been sent.");

				    	//set up connection with original peer
				    	int returnPort = Integer.parseInt(returnPeer) + 50000;
				    	Socket clientReturn = new Socket(IPAddress, returnPort);
				    	DataOutputStream outToClient = new DataOutputStream(clientReturn.getOutputStream());
				    	outToClient.writeBytes(returnPeer + "-" + fileName + "-" + selfID);
						clientReturn.close();
				    } else {
				    	System.out.println("File " + fileName + " is not stored here.");
				    	System.out.println("File request message has been forwarded to my successor.");
				    	Socket clientForward = new Socket(IPAddress, firstPort);
				    	DataOutputStream outToClient = new DataOutputStream(clientForward.getOutputStream());
				    	outToClient.writeBytes(returnPeer + "-" + fileName + "-" + selfID);
						clientForward.close();
				    }
				//==================================================================================
			    //=========================STEP 4===================================================
		    	} else if(parts.length ==2){
		    		//for step 4 quit
		    		String leavingPeer = parts[0].trim();
		    		System.out.println("Peer "+leavingPeer+" will depart from the network.");
    				if(firstSuc == Integer.parseInt(leavingPeer)){
    					firstSuc = secondSuc;
					}
    				secondSuc = Integer.parseInt(parts[1]);
    				firstPort = firstSuc + 50000;
    				secondPort = secondSuc + 50000;
    				System.out.println("My first successor is now peer "+firstSuc+".");
					System.out.println("My second successor is now peer "+secondSuc+".");
		    	}
                connectionSocket.close();
                //==================================================================================
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


	public static void main(String[] args) throws Exception {

		//multithreading
		cdht cdht = new cdht(args);
		new Thread(cdht.step3Server).start();
		Thread.sleep(1000);
		new Thread(cdht.step3Client).start();

		selfID = args[0];
		stringFirstSuc = args[1];
		stringSecondSuc = args[2];
		port = 50000 + Integer.parseInt(selfID);
		firstSuc = Integer.parseInt(stringFirstSuc);
		secondSuc = Integer.parseInt(stringSecondSuc);
		firstPort = firstSuc + 50000;
		secondPort = secondSuc + 50000;


		//=========================STEP 2===================================================
		//create IPAddress and socket for sending REQUEST (as client)
		InetAddress IPAddress = InetAddress.getByName("localhost");
		DatagramSocket socket = new DatagramSocket(); //client

		//create socket for receiving RESPONSE (as server)
	    DatagramSocket socket2 = new DatagramSocket(port); //server

    	//packet timeout
	    socket.setSoTimeout(3000);
	    socket2.setSoTimeout(3000);

		while (playDead != 1) {

    	 try {
		 	// send REQUEST ping to first successor
		    byte dataByte[] = selfID.getBytes();
		    DatagramPacket sendData1 = new DatagramPacket(dataByte,dataByte.length, IPAddress, firstPort);
		    socket.send(sendData1);

			// send REQUEST ping to second successor
		    DatagramPacket sendData2 = new DatagramPacket(dataByte,dataByte.length, IPAddress, secondPort);
		    socket.send(sendData2);

		    //receiving REQUEST (as server)
    		byte[] buffer = new byte[1024];
    		DatagramPacket receiveDataOne = new DatagramPacket(buffer,1024);
    		socket2.receive(receiveDataOne);
    		String bufferString = new String(buffer, "UTF-8");
    		String newBufferString = bufferString.trim();
    		Integer predNum = Integer.parseInt(newBufferString);
			System.out.println("A ping request message was received from Peer " + bufferString + ".\n");
			//=========================STEP 4===================================================
			//discover predecessors for step 4
			//let the largest predecessor be the closest predecessor
			if(predNum != firstPred && predNum != secondPred){
				if(predNum > tmpPred){
					secondPred = tmpPred;
					tmpPred = predNum;
				} else {
					secondPred = predNum;
				}
				if(tmpPred > firstPred) {

					firstPred = tmpPred;
				}
				System.out.println("firstPredecessor is " + firstPred + " secondPredecessor is " + secondPred);
			}
			//in the case where closest predecessor is smaller than furthest predecessor
			else if(firstPred > Integer.parseInt(selfID) && secondPred < Integer.parseInt(selfID)) {
				tmpPred = firstPred;
				firstPred = secondPred;
				secondPred = tmpPred;
			System.out.println("firstPredecessor is " + firstPred + " secondPredecessor is " + secondPred);
			}
			//==================================================================================
			//send RESPONSE (as server)
 			InetAddress clientHost = receiveDataOne.getAddress();
 	        int clientPort = receiveDataOne.getPort();
 	        byte[] response = selfID.getBytes();
 	        DatagramPacket reply = new DatagramPacket(response, response.length, clientHost, clientPort);
 	        socket2.send(reply);

 	        //receive RESPONSE as client
 	        byte[] buffer2 = new byte[1024];
 	        DatagramPacket response1 = new DatagramPacket(buffer2, 1024);
 	        socket.receive(response1);
 	        buffer2 = response1.getData();
 	        System.out.println("A ping response message was received from Peer " + new String(buffer2) + ".\n");

			Thread.sleep((int) (10000)); //remember to update report

			} catch (SocketTimeoutException e) {
	    		 continue;
	    	 }
    	 //==================================================================================
		}
	}
}