SDIS1819-T2G01

import java.util.Random;

import java.rmi.registry.Registry;

import java.rmi.registry.LocateRegistry;

import java.rmi.RemoteException;

import java.rmi.server.UnicastRemoteObject;

import java.util.Map;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.concurrent.Executors;

import java.util.concurrent.CopyOnWriteArrayList;

import java.util.ArrayList;

import java.util.List;

import java.util.Comparator;

import java.util.HashSet;

import java.util.Collections;

import java.io.*;

import java.net.DatagramPacket;

import java.nio.file.*;

public class Interpreter implements Runnable {

    public Random rand = new Random();

    private static final byte CR = 0xD;

    private static final byte LF = 0xA;

    public Server peer;

    public String raw;

    public String header;

    public byte[] body;

    public String version;

    public String senderId;

    public String fileId;

    public String chunkNo;

    public String replication;

    public AtomicBoolean elseSentChunk = new AtomicBoolean(false);

    public AtomicBoolean elseSentChunkReclaim = new AtomicBoolean(true);

    public Message.MessageType type;

    public Interpreter() {

    }

    public Interpreter(Server peer, DatagramPacket packet) {

        // Set peer

        this.peer = peer;

        // Seperate header and body

        messageSplitHeaderBody(packet.getData(), packet.getLength());

        System.out.println(header);

        // Get parts

        String[] parts = header.split(" ");

        // Identify type and specific arguments

        switch (parts[0]) {

        case "PUTCHUNK":

            type = Message.MessageType.PUTCHUNK;

            chunkNo = parts[4];

            replication = parts[5];

            break;

        case "STORED":

            type = Message.MessageType.STORED;

            chunkNo = parts[4];

            break;

        case "GETCHUNK":

            type = Message.MessageType.GETCHUNK;

            chunkNo = parts[4];

            break;

        case "CHUNK":

            type = Message.MessageType.CHUNK;

            chunkNo = parts[4];

            break;

        case "DELETE":

            type = Message.MessageType.DELETE;

            break;

        case "REMOVED":

            type = Message.MessageType.REMOVED;

            chunkNo = parts[4];

            break;

        default:

            break;

        }

        // Version, SenderId and FileId are shared

        version = parts[1];

        senderId = parts[2];

        fileId = parts[3];

    }

    // Split message header and body

    public void messageSplitHeaderBody(byte[] message, int size) {

        // Find index of CRLF

        int i = 0;

        for (i = 0; i < size; i++) {

            if (i <= size - 5) {

                if (message[i] == CR && message[i + 1] == LF && message[i + 2] == CR && message[i + 3] == LF) {

                    break;

                }

            }

        }

        // Get header

        byte[] headerByte = new byte[i];

        System.arraycopy(message, 0, headerByte, 0, i);

        header = new String(headerByte);

        header = header.trim();

        // Get body

        if (size > i + 3) {

            body = new byte[size - i - 4];

            System.arraycopy(message, i + 4, body, 0, size - i - 4);

        } else {

            body = null;

        }

    }

    @Override

    public void run() {

        // Do whatever the message asks

        try {

            // Identify Type

            switch (type) {

            case PUTCHUNK:

                // Someone(NOT US!) sent a CHUNK, cancel it

                if (this.elseSentChunkReclaim.get() && senderId != this.peer.id)

                    this.elseSentChunkReclaim.set(true);

                // Dont backup self

                if (peer.doingBackUp)

                    return;

                // Check peer has enough size available

                if (peer.used + body.length > peer.limit)

                    return;

                // Get path to peer

                Path currentRelativePath = Paths.get("");

                String peerDir = currentRelativePath.toAbsolutePath().toString() + "\\peer" + peer.id;

                peerDir += "\\backup";

                peerDir += "\\" + fileId;

                File file = new File(peerDir);

                // Create path

                file.mkdirs();

                // Store chunk in file system

                peerDir += "\\chk" + chunkNo + ".chk";

                File chunk = new File(peerDir);

                FileOutputStream out = new FileOutputStream(chunk);

                out.write(body);

                out.close();

                // Store chunk in Peer cache(hashmap)

                peer.chunks.put(fileId + "-" + chunkNo,

                        new Chunk(fileId, Integer.parseInt(chunkNo), body, Integer.parseInt(replication)));

                // Add to used memory

                peer.used += body.length;

                // Random delay

                Thread.sleep(rand.nextInt(400));

                // Send STORED to MC

                Message msg = new Message(Message.MessageType.STORED, peer.id, fileId, Integer.parseInt(chunkNo), 0,

                        new byte[0]);

                DatagramPacket packet = msg.packit(Server.MC.address, Server.MC.port);

                Server.MC.socket.send(packet);

                break;

            case STORED:

                // If doing backup then add to replications

                if (peer.doingBackUp && peer.file.equals(fileId)) {

                    peer.replications++;

                }

                // Chunk that was stored

                String keyStored = fileId + "-" + chunkNo;

                // If we have the chunk

                if (this.peer.chunks.get(keyStored) != null) {

                    this.peer.chunks.get(keyStored).cReps++;

                }

                break;

            case GETCHUNK:

                // Find chunk in store

                Chunk requestedChunk = peer.chunks.get(fileId + "-" + chunkNo);

                // Chunk not found, just return

                if (requestedChunk == null)

                    return;

                // Set flag to lookout for CHUNK

                this.elseSentChunk.set(false);

                // Wait random delay

                Thread.sleep(rand.nextInt(400));

                // Someone else sent the CHUNK, just return

                if (this.elseSentChunk.get())

                    return;

                // Send CHUNK to MDR

                Message msgGetChunk = new Message(Message.MessageType.CHUNK, peer.id, fileId, Integer.parseInt(chunkNo),

                        0, requestedChunk.bytes);

                DatagramPacket packetGetChunk = msgGetChunk.packit(Server.MDR.address, Server.MDR.port);

                Server.MDR.socket.send(packetGetChunk);

                break;

            case CHUNK:

                // Find chunk in store

                Chunk receivedChunk = peer.chunks.get(fileId + "-" + chunkNo);

                // Someone(NOT US!) sent a CHUNK we were going to send, cancel it

                if (receivedChunk != null && senderId != this.peer.id) {

                    this.elseSentChunk.set(true);

                }

                // Add to recovered if not there yet

                if (this.peer.restoring) {

                    if (!this.peer.recoveredChunks.containsKey(chunkNo)) {

                        this.peer.recoveredChunks.put(chunkNo, new Chunk(fileId, Integer.parseInt(chunkNo), body, 0));

                        // Chunk received, continue asking for rest

                        if (peer.waitRestoreReply.get())

                            peer.waitRestoreReply.set(false);

                    }

                }

                break;

            case DELETE:

                boolean found1 = false;

                // Delete cache chunks

                for (String keyDelete : this.peer.chunks.keySet()) {

                    if (keyDelete.contains(this.fileId)) {

                        found1 = true;

                        this.peer.chunks.remove(keyDelete);

                    }

                }

                // Delete filesystem chunks

                // If at least one in cache was found then surely it exists in file system and

                // we should delete

                if (found1) {

                    Path current = Paths.get("");

                    String fileDir = current.toAbsolutePath().toString() + "\\peer" + peer.id;

                    fileDir += "\\backup";

                    fileDir += "\\" + fileId;

                    File fileDirFile = new File(fileDir);

                    FileManager.deleteDir(fileDirFile);

                }

                break;

            case REMOVED:

                // Chunk requested

                String key = fileId + "-" + chunkNo;

                // If we have the chunk

                if (this.peer.chunks.get(key) != null) {

                    // Remove replication

                    this.peer.chunks.get(key).cReps--;

                    System.out.println(this.peer.chunks.get(key).cReps);

                    System.out.println(this.peer.chunks.get(key).reps);

                    // If bellow replication degree

                    if (this.peer.chunks.get(key).cReps < this.peer.chunks.get(key).reps) {

                        // Wait random delay

                        this.elseSentChunkReclaim.set(false);

                        Thread.sleep(rand.nextInt(400));

                        // Someone already started backup of this chunk

                        if (this.elseSentChunkReclaim.get())

                            return;

                        // Prepare message

                        Message msgRemovedChunk = new Message(Message.MessageType.PUTCHUNK, this.peer.id, fileId,

                                this.peer.chunks.get(key).chunkN, this.peer.chunks.get(key).reps,

                                this.peer.chunks.get(key).bytes);

                        // Pack in datagram

                        DatagramPacket packetRemovedChunk = msgRemovedChunk.packit(this.peer.MDB.address,

                                this.peer.MDB.port);

                        // Send message

                        this.peer.MDB.socket.send(packetRemovedChunk);

                    }

                }

                break;

            default:

                break;

            }

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}