sdis1819-t2g03

package handlers;

import protocols.*;

import server.Peer;

import java.util.Random;

import java.util.concurrent.*;

import chunk.Chunk;

import chunk.FileManager;

public class Handler implements Runnable {

        private BlockingQueue<Message> msgQueue;

        private ScheduledExecutorService agent;

        private Peer peer;

        private Chunk peerInformation;

        private int delayLimit;

        private Random random;

        public Handler(Peer peer) {

                this.peer = peer;

                this.peerInformation = peer.getPeerInformation();

                msgQueue = new LinkedBlockingQueue<>();

                agent = Executors.newScheduledThreadPool(5);

                delayLimit = 300;

                this.random = new Random();

        }

        @Override

        public void run() {

                Message msg;

                while (true) {

                        try {

                                msg = msgQueue.take();

                                if (msg == null) {

                                        System.out.println("No msg received!");

                                        return;

                                }

                                System.out.println("R:: " + msg.toString());

                                switch (msg.getType()) {

                                case PUTCHUNK:

                                        Backup backup = new Backup(peer, msg);

                                        agent.execute(backup);

                                        break;

                                case STORED:

                                        peerInformation.addChunkReplication(msg.getFileID(), msg.getChunkNo());

                                        break;

                                case REMOVED:

                                        FileManager database = peer.getDatabase();

                                        String fileID = msg.getFileID();

                                        int chunkNo = msg.getChunkNo();

                                        Chunk chunkInfo = database.getChunkInformation(fileID, chunkNo);

                                        int perceivedReplication = database.getRecognizedReplication(fileID, chunkNo);

                                        int targetReplication = chunkInfo.getReplication();

                                        if (perceivedReplication < targetReplication) {

                                                byte[] chunkData = peer.loadChunk(fileID, chunkNo);

                                                agent.schedule(

                                                                new RemovedChunkHandler(peer, chunkInfo, chunkData),

                                                                this.random.nextInt(delayLimit + 1),

                                                                TimeUnit.MILLISECONDS

                                                                );

                                        }

                                        break;

                                case DELETE:

                                        Delete delete = new Delete(peer, msg);

                                        agent.execute(delete);

                                        break;

                                default:

                                        return;

                                }

                        } catch (InterruptedException e) {

                                e.printStackTrace();

                        }

                }

        }

        public void pushMessage(byte[] data, int length) {

                Message msgParsed = new Message(data, length);

                msgQueue.add(msgParsed);

        }

        // Getters and setters

        public BlockingQueue<Message> getMsgQueue() {

                return msgQueue;

        }

        public ScheduledExecutorService getAgent() {

                return agent;

        }

        public Peer getPeer() {

                return peer;

        }

        public Chunk getPeerInformation() {

                return peerInformation;

        }

        public int getDelayLimit() {

                return delayLimit;

        }

        public Random getRandom() {

                return random;

        }

        public void setMsgQueue(BlockingQueue<Message> msgQueue) {

                this.msgQueue = msgQueue;

        }

        public void setAgent(ScheduledExecutorService agent) {

                this.agent = agent;

        }

        public void setPeer(Peer peer) {

                this.peer = peer;

        }

        public void setPeerInformation(Chunk peerInformation) {

                this.peerInformation = peerInformation;

        }

        public void setDelayLimit(int delayLimit) {

                this.delayLimit = delayLimit;

        }

        public void setRandom(Random random) {

                this.random = random;

        }

}