sdis1819-t7g05

import java.io.File;

import java.io.FileOutputStream;

import java.io.IOException;

import java.net.DatagramPacket;

import java.net.DatagramSocket;

import java.net.InetAddress;

import java.util.Random;

import java.util.concurrent.ConcurrentHashMap;

/**

 * PutChunkTask

 */

public class PutChunkTask implements Runnable {

    private static final String CRLF = Integer.toHexString(0xD) + Integer.toHexString(0xA);

    private String version;

    private int senderId;

    private String fileId;

    private int chunkNo;

    private int replicationDegree;

    private byte[] body;

    private InetAddress MCaddress;

    private int MCport;

    private MDBListener mdb;

    private Peer peer;

    PutChunkTask(String[] args, InetAddress MCaddress, int MCport, MDBListener mdb, Peer peer, byte[] body) {

        this.version = args[1];

        this.senderId = Integer.parseInt(args[2]);

        this.fileId = args[3];

        this.chunkNo = Integer.parseInt(args[4]);

        this.replicationDegree = Integer.parseInt(args[5]);

        this.body = body;

        this.MCaddress = MCaddress;

        this.MCport = MCport;

        this.mdb = mdb;

        this.peer = peer;

    }

    @Override

    public void run() {

        File dir =  new File("Peer-" + this.peer.getSenderId() + File.separator + "Chunks" + File.separator + this.fileId);

        File file = new File(dir.getPath() + File.separator + this.chunkNo);

        dir.mkdirs();

        try {

            //STORED <Version> <SenderId> <FileId> <ChunkNo> <CRLF><CRLF>

            String message = "STORED "+this.version+" "+this.senderId+" "+this.fileId+" "+this.chunkNo+" "+CRLF+CRLF;

            DatagramSocket socket = new DatagramSocket();

            DatagramPacket packet = new DatagramPacket(message.getBytes(), message.length(), this.MCaddress, this.MCport);

            Random random = new Random();

            //Random delay between [0-400] ms to avoid overlapping messages

            Thread.sleep(random.nextInt(401));

            //Protocol enhancement - Only store the chunk if the desired replication degree hasn't been achieved yet to avoid using unnecessary storage space

            if (this.version.equals("2.0")) {

                if (!this.peer.repDegMap.containsKey(this.fileId))

                    this.peer.repDegMap.put(this.fileId, new ConcurrentHashMap<>());

                if(!this.peer.repDegMap.get(this.fileId).containsKey(this.chunkNo))

                    this.peer.repDegMap.get(this.fileId).put(this.chunkNo, 0);

                this.mdb.print("RepDegMap("+this.chunkNo+") = "+this.peer.repDegMap.get(this.fileId).get(this.chunkNo));

                if (this.peer.repDegMap.get(this.fileId).get(this.chunkNo) < this.replicationDegree){

                    FileOutputStream stream = new FileOutputStream(file);

                    stream.write(this.body);

                    this.peer.addStorage(this.body.length);

                    stream.close();

                    if (this.peer.chunksMap.get(this.fileId) == null)

                        this.peer.chunksMap.put(this.fileId, new BackupFile(this.fileId));

                    this.peer.chunksMap.get(this.fileId).addChunk(this.chunkNo, new Chunk(this.peer.chunksMap.get(this.fileId), this.chunkNo, this.body));

                    socket.send(packet);

                }

            }

            //Normal protocol - Always store the chunk, even if the replication degree might have been achieved already

            else {

                FileOutputStream stream = new FileOutputStream(file);

                stream.write(this.body);

                        this.peer.addStorage(this.body.length);

                stream.close();

                if (this.peer.chunksMap.get(this.fileId) == null)

                    this.peer.chunksMap.put(this.fileId, new BackupFile(this.fileId));

                this.peer.chunksMap.get(this.fileId).getChunksMap().put(this.chunkNo, new Chunk(this.peer.chunksMap.get(this.fileId), this.chunkNo, this.body));

                socket.send(packet);

            }

            socket.close();

        } catch (IOException | InterruptedException e) {

            e.printStackTrace();

            return;

        }

        this.mdb.print("PUTCHUNK protocol finished");

    }

}