sdis1819-t7g02

package service;

import java.io.IOException;

import java.nio.file.Files;

import java.nio.file.Path;

import java.nio.file.Paths;

import java.rmi.registry.Registry;

import java.rmi.registry.LocateRegistry;

import java.rmi.server.UnicastRemoteObject;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import channels.Message;

import channels.MessageHeader;

import protocols.BackupChunkProtocol;

import protocols.RestoreChunksProtocol;

public class Peer implements RMI {

        ScheduledThreadPoolExecutor exec;

        private static Cloud cloud;

        public Peer() {}

        public static void main(String[] args) {

                try 

                {

                        Peer peer = new Peer();

                        RMI rmi = (RMI) UnicastRemoteObject.exportObject(peer, 0);

                        String acess_point = args[2];

                        Registry registry = LocateRegistry.getRegistry();

                        registry.bind(acess_point, rmi);

                        System.err.println("Peer ready");

                } 

                catch (Exception e) {

                        System.err.println("Peer exception: " + e.toString());

                        return;

                }

                if(args.length != 9) {

                        printHelp();

                        return;

                }

                try {

                        Cloud aux = new Cloud(args[0], args[1]);

                        cloud = aux.getPeerTools(args[0], args[1]);

                        cloud.createControlRoom(args[3], args[4]);

                        cloud.createBackupChannel(args[5], args[6]);

                        cloud.createRestoreChannel(args[7], args[8]);

                        String path = Constants.DATABASE_DIR;

                        Path p = Paths.get(path);

                        if (Files.exists(p)) 

                        {

                                try {

                                        cloud.loadFromDisk();

                                } catch (Exception e) {

                                        e.printStackTrace();

                                }

                        }

                        //                        System.out.println(cloud.getStoredChunks().size());

                } catch (NumberFormatException | IOException e) {

                        System.out.println(e.getMessage());

                        return;

                }

                if(Cloud.checkOrCreateDirectory(Constants.CHUNKS_DIR + args[0] + "/") == -1) {

                        System.out.println("Couldn't create peer chunks dir!");

                }

                cloud.activateChannels();

                System.out.println("Channels activated");

        }

        // the name of each multicast channel consists of the ip multicast address and

        // port, passed as cmd line arguments, followed by protocol version, server id

        // and service access point

        private static void printHelp() {

                System.out.println("java service.Peer SERVER_id VERSION ACCESS_POINT MC_ip MC_port MDB_ip MDB_port MDR_ip MDR_port");

        }

        public void saveState() {

                try {

                        cloud.saveToDisk();

                } catch (IOException e) {

                        e.printStackTrace();

                }

        }

        public synchronized String backup(String file_path, int replication_degree) 

        {          

                FileChunker fc;

                Message m;

                MessageHeader h;

                int numberOfChunks;

                ExecutorService WORKER_THREAD_POOL;

                List<Callable<Object>> callables = new ArrayList<Callable<Object>>();

                List<Future<Object>> futures = null;

                ArrayList<Chunk> chunkHolder = null;

                String[] headerElements = new String[Constants.PUTCHUNK_N_ARGS];

                Chunk aux;

                int numberOfSteps;

                System.out.println("BACKUP request received");

                try {

                        fc = new FileChunker(file_path, replication_degree);

                        //                cloud.createDirectoryFile(fc);

                } 

                catch (IOException e) {

                        System.out.println("ERROR chunking file -> " + e.getMessage());

                        return "ERROR chunking file -> " + e.getMessage();

                }

                chunkHolder = fc.getChunks();

                numberOfChunks = chunkHolder.size();

                cloud.registerNumberOfFileChunks(fc.getFileID(), numberOfChunks);

                System.out.println("File splited into " + numberOfChunks + " chunks.");

                headerElements[0] = Constants.PUTCHUNK;

                headerElements[1] = cloud.getProtocolVersion();

                headerElements[2] = Integer.toString(cloud.getID());

                numberOfSteps = numberOfChunks % Constants.MAX_NUMBER_OF_THREADS == 0 ? (int)(numberOfChunks / Constants.MAX_NUMBER_OF_THREADS):(int)(numberOfChunks / Constants.MAX_NUMBER_OF_THREADS) + 1;

                for(int step = 0, first, last; step < numberOfSteps; step++, futures.clear(), callables.clear()) {

                        first = step * Constants.MAX_NUMBER_OF_THREADS;

                        last = numberOfChunks - first < Constants.MAX_NUMBER_OF_THREADS ? numberOfChunks : first + Constants.MAX_NUMBER_OF_THREADS;

                        System.out.println("Sending from " + first + " to " + last);

                        WORKER_THREAD_POOL = Executors.newFixedThreadPool(last-first);

                        for(int chunkNo = first; chunkNo < last; chunkNo++) {

                                aux = chunkHolder.get(chunkNo);

                                headerElements[3] = fc.getFileID();

                                headerElements[4] = aux.getChunkNumber();

                                headerElements[5] = Integer.toString(replication_degree);

                                try {

                                        h = new MessageHeader(headerElements.clone());

                                        m = new Message(h, aux.getChunkContent());

                                        callables.add(new BackupChunkProtocol(cloud, aux, m ,replication_degree));

                                } 

                                catch(IllegalArgumentException e) {

                                        System.out.println(e.getMessage());

                                        return "Backup failed preparing chunks";

                                }

                        }

                        try {

                                futures = WORKER_THREAD_POOL.invokeAll(callables);

                                WORKER_THREAD_POOL.shutdown();

                                if(WORKER_THREAD_POOL.awaitTermination(35, TimeUnit.SECONDS)) {

                                        for(Future<Object> p: futures) {

                                                if(!(boolean)p.get()) {

                                                        System.out.println("One of the chunks could not be stored.");

                                                        delete(file_path); //order successfully stored chunks to be deleted

                                                        return "Backup failed preparing chunks";

                                                }

                                        }

                                }

                        }

                        catch (InterruptedException | ExecutionException e) {

                                return "Error in peer: " + e.getMessage();

                        }         

                }

                System.out.println("backup peer has " + cloud.getNumberOfStoredChunks());

                saveState();

                return "Backup successfull.";

        }

        public synchronized String restore(String file_path)

        {

                System.out.println("RESTORE request received");

                FileChunker fc;

                List<Callable<Object>> callables = new ArrayList<Callable<Object>>();

                List<Future<Object>> futures = null;

                int numberOfChunks;

                MessageHeader h;

                int numberOfSteps;

                ExecutorService WORKER_THREAD_POOL;

                String[] headerElements = new String[Constants.GETCHUNK_N_ARGS];

                headerElements[0] = Constants.GETCHUNK;

                headerElements[1] = cloud.getProtocolVersion();

                headerElements[2] = Integer.toString(cloud.getID());

                try

                {

                        fc = new FileChunker(file_path);

                }

                catch (IOException e) 

                {

                        System.out.println("ERROR restoring file -> " + e.getMessage());

                        return "ERROR restoring file -> " + e.getMessage();

                }

                headerElements[3] = fc.getFileID();

                String file_id = fc.getFileID();

                numberOfChunks = cloud.getNumberOfChunks(file_id);

                if(numberOfChunks < 0) {

                        return file_path + " wasn't backep up by this peer.";

                }

                numberOfSteps = numberOfChunks % Constants.MAX_NUMBER_OF_THREADS == 0 ? (int)(numberOfChunks / Constants.MAX_NUMBER_OF_THREADS):(int)(numberOfChunks / Constants.MAX_NUMBER_OF_THREADS) + 1;      

                for(int step = 0, first, last; step < numberOfSteps; step++, futures.clear(), callables.clear()) {

                        first = step * Constants.MAX_NUMBER_OF_THREADS;

                        last = numberOfChunks - first < Constants.MAX_NUMBER_OF_THREADS ? numberOfChunks : first + Constants.MAX_NUMBER_OF_THREADS;

                        System.out.println("Sending from " + first + " to " + last);

                        WORKER_THREAD_POOL = Executors.newFixedThreadPool(last-first);

                        for(int chunkNo = first; chunkNo < last; chunkNo++) 

                        {

                                headerElements[4] = Integer.toString(chunkNo);

                                try 

                                {

                                        h = new MessageHeader(headerElements.clone());

                                        cloud.registerFileChunkToRestore(file_id, headerElements[4]);

                                        callables.add(new RestoreChunksProtocol(cloud, h));

                                } 

                                catch(IllegalArgumentException e) {

                                        System.out.println(e.getMessage());

                                        return "Restore failed preparing chunks";

                                }

                        }

                        try {

                                futures = WORKER_THREAD_POOL.invokeAll(callables);

                                WORKER_THREAD_POOL.shutdown();

                                if(WORKER_THREAD_POOL.awaitTermination(35, TimeUnit.SECONDS)) {

                                        for(Future<Object> p: futures) {

                                                if(!(boolean)p.get()) {

                                                        System.out.println("One of the chunks could not be restored.");

                                                        return "Restore failed preparing chunks";

                                                }

                                        }

                                }

                        }

                        catch (InterruptedException | ExecutionException e) {

                                return "Error in peer: " + e.getMessage();

                        }         

                }

                System.out.println("Colected all file parts. Building file.");

                try {

                        cloud.buildFile(file_id, numberOfChunks, Paths.get(file_path).getFileName().toString());

                } 

                catch (IOException e) {

                        System.out.println("ERROR builidng file -> " + e.getMessage());

                }

                saveState();

                return "Restored file successfully.";

        }

        public synchronized String delete(String file_path)

        {    

                FileChunker fc;

                MessageHeader h;

                String[] headerElements = new String[Constants.DELETE_N_ARGS];

                System.out.println("DELETE request.");

                try {

                        fc = new FileChunker(file_path);

                        headerElements[0] = Constants.DELETE;

                        headerElements[1] = cloud.getProtocolVersion();

                        headerElements[2] = Integer.toString(cloud.getID());

                        headerElements[3] = fc.getFileID();  

                        h = new MessageHeader(headerElements.clone());

                        if(cloud.checkIfPeerStoredFile(headerElements[3])) {

                                for(int i = 0; i < 3; i++) {

                                        cloud.controlRoom.sendHeader(h);

                                        Thread.sleep(1000);

                                }          

                        }

                        else {

                                return "File was not uploaded here. Delete failed.";

                        }

                }

                catch(Exception e) {

                        System.out.println("ERROR deleting file -> " + e.getMessage());

                        return "ERROR deleting file -> " + e.getMessage();

                }

                saveState();

                return "Delete successfull.";

        }

        public synchronized String reclaim(long disk_space)

        {

                if(disk_space < 0) {

                        return "Space_disk can't be lower than 0!";

                }

                System.out.println("Space reclaim request to " + disk_space + " bytes.");

                if(disk_space >= cloud.getChunkSpace()) {

                        cloud.setAvailableSpace(disk_space);

                }

                else {

                        cloud.freeUpSpace(disk_space);

                }

                saveState();

                return "Space reclaim successfull";

        }

        public synchronized String state()

        {

                System.out.println("STATE OPERATION RECEIVED");

                String answer = "\n\nPeer " + cloud.getID() + " state: \n";

                answer += "\tBacked-up files:\n";

                String pathName;

                int repDegree, nChunks;

                Chunk c;

                System.out.println("Number of files = " + cloud.getBackedUpFileNames().size());

                for(String key : cloud.getBackedUpFileNames()) {

                        pathName = cloud.getPathNameFromFileId(key);

                        repDegree = cloud.getBackedUpChunkNecessaryRepDegree(key);

                        if(pathName != null && repDegree > -1) {

                                answer += "\t\tFILE PATHNAME: " + pathName + "    FILE ID: " + key + "    DESIRED REPPLICATION DEGREE:" + repDegree + "\n";

                        }

                        else {

                                System.out.println("Error in key: " + key + " pathname " + pathName + " degree " + repDegree);

                                continue;

                        }

                        nChunks = cloud.getNumberOfChunks(key);

                        if(nChunks < 0) {

                                System.out.println("Error getting number of chunks in key: " + key + " pathname " + pathName + " degree " + repDegree);

                                continue;

                        }

                        answer += "\t\t\tChunks:\n";

                        for(int i = 0; i < nChunks; i++) {

                                answer += "\t\t\t\tchk" + i + "   REP: DEGREE " + cloud.getBackepUpChunkPerceivedRepDegree(key, i) + "\n";

                        }

                        answer += "\n";

                }

                answer += "\tStored chunks:\n";

                for(String key : cloud.getStoredChunks().keySet()) {

                        c = cloud.getStoredChunks().get(key);

                        if(c.getFilePathName() == null) { //only backep up files have a non-null file path name

                                answer += "\t\t" +c.getChunkId() + ":" + c.getChunkNumber() + "   " + "SIZE: " + c.getChunkSize() / 1024 + " KBytes    " + " PERCEIVED REP. DEGREE: " + c.getActualReplicationDegree() + "\n";

                        }

                }

                answer += "\n\n\t Total peer disk space: " + cloud.getTotalSpace() / 1024 + " KBytes   Space used by chunks : " + cloud.getChunkSpace() / 1024 + " KBytes   Available space: " + cloud.getAvailableSpace() / 1024 + " KBytes.\n";

                return answer;

        }

}