sdis1819-t7g02

package service;

import channels.BackupChannel;

import channels.ControlChannel;

import channels.Message;

import channels.MessageHeader;

import channels.RestoreChannel;

import protocols.SendChunkMessageProtocol;

import java.io.*;

import java.nio.file.Files;

import java.nio.file.Path;

import java.nio.file.Paths;

import java.util.*;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ConcurrentHashMap.KeySetView;

import java.util.concurrent.ConcurrentLinkedQueue;

//https://www.geeksforgeeks.org/serialization-in-java/

public class Cloud implements java.io.Serializable

{

        private static final long serialVersionUID = 1L;

        /**

         * stores received V and backedUp chunks

         * key = <fileID>

         */

        private ConcurrentHashMap<String, Chunk> storedChunks;

        /**

         * stores the number of chunks taht a file that was backed up in peer was splited into

         * key = <fileID>

         */

        private ConcurrentHashMap<String, Integer> numberOfFileChunks;

        /**

         * registers the peers that have stored chunks sent by this peer

         * key = <fileID>:<ChunkNo>

         */

        private ConcurrentHashMap<String, HashSet<Integer>> colaborativePeers;

        /**

         * stores the number of times a chunk was stored

         * key = <fileID>:<ChunkNo>

         */

        private ConcurrentHashMap<String, Integer> numberOfChunksConfirmations;

        /**

         * key = <fileID>:<ChunkNo>

         */

        private ConcurrentHashMap<String, Integer> chunksToRestore;

        /**

         * Holds chucks restored by restore protocol

         * key = <fileID>:<ChunkNo>

         */

        private ConcurrentHashMap<String, Chunk> recoveredChunks;

        /**

         * stores the number of chunks a file was splited into

         * key = <fileID>

         */

        private ConcurrentLinkedQueue<String> numberOfFileChunksToBeRestored;

        private ArrayList<FileChunker> data;

        private int serverID;

        private String protocol_version;

        public ControlChannel controlRoom;

        public BackupChannel backupCh;

        public RestoreChannel restoreCh;

        long spaceRemaining;

        public Cloud(String id, String version)

        {

                numberOfFileChunks = new ConcurrentHashMap<String, Integer>();

                storedChunks = new ConcurrentHashMap<String, Chunk>();

                colaborativePeers = new ConcurrentHashMap<String, HashSet<Integer>>();

                numberOfChunksConfirmations = new ConcurrentHashMap<String, Integer>();

                this.data = new ArrayList<>();

                this.numberOfFileChunksToBeRestored = new ConcurrentLinkedQueue<String>();

                this.recoveredChunks = new  ConcurrentHashMap<String, Chunk>();

                this.spaceRemaining = 1073741824; //1GB

                serverID = Integer.parseInt(id);

                protocol_version = version;

        }

        public synchronized boolean sendBackupMessage(Message m) {

                return backupCh.sendMessage(m);

        }

        public synchronized boolean sendRestoreMessage(Message m) {

                return restoreCh.sendMessage(m);

        }

        public Cloud getPeerTools(String id, String version) {

                return new Cloud(id, version);

        }

        public int getChunkCurrentRepDegree(String fileId, String chunkNo) {

                String key = fileId + ':' + chunkNo;

                if(numberOfChunksConfirmations.containsKey(key)) {

                        return numberOfChunksConfirmations.get(key);

                }

                return -1;

        }

        public byte[] reverseList(byte[] a) {

                ArrayList<Byte> aux = new ArrayList<Byte>(a.length);

                for(byte b : a) {

                        aux.add(0, b);

                }

                byte[] toReturn = new byte[a.length];

                int i = 0;

                for(byte b : aux) {

                        toReturn[i] = b;

                        i++;

                }

                return toReturn;

        }

        public boolean buildFile(String file_id, int maxchunks, String fileName) throws IOException

        {

                String restore_path = Constants.CHUNKS_DIR + this.serverID + "/restored/";

                String key;

                Chunk aux;

                int total = 0;

                checkOrCreateDirectory(restore_path);

                File restoredFile = new File(restore_path + fileName);

                restoredFile.createNewFile();

                FileOutputStream fos =  new FileOutputStream(restoredFile);

                for(int i = 0; i < maxchunks; i++)

                {

                        key = file_id + ':' + Integer.toString(i);

                        if(recoveredChunks.containsKey(key)) {

                                aux = recoveredChunks.get(key);

                                if(aux != null) {

                                        fos.write(reverseList(aux.getChunkContent()));

                                        total += aux.getChunkContent().length;

                                }

                                else {

                                        System.out.println("NULL chunk file might be corrupted");

                                }

                        }

                }

                try

                {

                        fos.close();

                }

                catch(IOException e) {

                        recoveredChunks.clear();

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

                }

                recoveredChunks.clear();

                System.out.println("Restored file " + fileName + " with " + total + " bytes.");

                return false;

        }

        public ArrayList<FileChunker> getData()

        {

                return this.data;

        }

        public ConcurrentHashMap<String, Integer> getChunksToRestore()

        {

                return this.chunksToRestore;

        }

        void addFile(FileChunker fc)

        {

                this.data.add(fc);

        }

        public void addChunk(Chunk c) {

                String key = c.getChunkId() + ":" + c.getChunkNumber();

                storedChunks.put(key, c);

        }

        public void deleteChunks(FileChunker fc, ArrayList<FileChunker> fcs)

        {

                String file_id = fc.getFileID();

                String path = Constants.CHUNKS_DIR + this.serverID + "/backup/" + file_id + '/';

                File f = new File(path);

                String[] entries = f.list();

                for(String s: entries)

                {

                        File currentFile = new File(f.getPath(), s);

                        currentFile.delete();

                }

                f = new File(path);

                f.delete();

        }

        public boolean storeRestoreChunk(Message m) {

                Chunk toStore = null;

                byte[] chunkData;

                try {

                        chunkData = m.getChunkContent();

                        toStore = new Chunk(Integer.parseInt(m.getHeader().getChunkNumber()), chunkData, m.getHeader().getFileId(), Integer.parseInt(m.getHeader().getDegree()));

                } catch (Exception e) {

                        System.out.println("Error in chunk: " + e.getMessage());

                        return false;

                }

                addToRecoveredChunks(toStore);

                return true;

        }

        public boolean storeChunk(Message m) {

                Chunk toStore = null;

                String filename, path;

                FileOutputStream out = null;

                File dir = null;

                byte[] chunkData;

                try {

                        chunkData = m.getChunkContent();

                        toStore = new Chunk(Integer.parseInt(m.getHeader().getChunkNumber()), chunkData, m.getHeader().getFileId(), Integer.parseInt(m.getHeader().getDegree()));

                } catch (Exception e) {

                        System.out.println("Error in chunk: " + e.getMessage());

                        return false;

                }

                if(toStore.getChunkContent().length > spaceRemaining) {

                        System.out.println("There is no available space");

                        return false;

                }

                this.addChunk(toStore);

                spaceRemaining -= toStore.getChunkSize();

                path = Constants.CHUNKS_DIR + this.serverID + "/backup/" + toStore.getChunkId() + "/";

                filename = getFileName(Integer.parseInt(toStore.getChunkNumber()));

                if(!doesDirExists(path)) {

                        dir = new File(path);

                        if(!dir.mkdirs()) {

                                return false;

                        }

                }

                try {

                        dir = new File(path + filename);

                        dir.createNewFile();

                        out = new FileOutputStream(dir);

                        out.write(toStore.getChunkContent());

                        out.close();

                } catch (IOException e) {

                        System.out.println("Error writing file to disk: " + e.getMessage());

                        return false;

                }

                return true;

        }

        public boolean checkIfChunkWasRestored(String fileid, String chunknr)

        {

                return this.recoveredChunks.containsKey(fileid + ':' + chunknr);

        }

        public void addToRecoveredChunks(Chunk c) {

                String key = c.getChunkId() + ":" + c.getChunkNumber();

                if(!recoveredChunks.containsKey(key)) {

                        recoveredChunks.put(key, c);

                }

        }

        public void registerFileChunkToRestore(String file_id, String chunknr)

        {

                if(!this.numberOfFileChunksToBeRestored.contains(file_id + ':' + chunknr))

                {

                        this.numberOfFileChunksToBeRestored.add(file_id + ':' + chunknr);

                }

        }

        public boolean isFileChunkRegistered(String fileId, String chunknr) {

                return this.numberOfFileChunksToBeRestored.contains(fileId + ':' + chunknr);

        }

        public void createDirectoryFile(FileChunker fc) throws IOException

        {

                String file_id = fc.getFileID();

                String path = Constants.CHUNKS_DIR + this.serverID + "/backup/" + file_id + '/';

                if(checkOrCreateDirectory(path) == -1)

                {

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

                }

                for(int i = 0; i < fc.getChunks().size(); i++)

                {

                        int nr = i + 1;

                        String filename = "chk" + Integer.toString(nr);

                        File dir = new File(path + filename);

                        dir.createNewFile();

                        try(FileOutputStream out = new FileOutputStream(dir))

                        {

                                out.write(fc.getChunks().get(i).getChunkContent());

                        }

                }

        }

        private static String getFileName(int chunkNumber) {

                return "chk" + Integer.toString(chunkNumber);

        }

        private static boolean doesDirExists(String path) {

                File f = new File(path);

                return f.exists();

        }

        public boolean checkIfPeerStoredFile(String filename) {

                return numberOfFileChunks.containsKey(filename);

        }

        public static int checkOrCreateDirectory(String dirName)

        {

                File f = new File(dirName);

                int result = -1;

                if(!f.exists())

                {

                        //                        System.out.println("Creating directory " + f.getName());

                        try

                        {

                                f.mkdirs();

                                result = 1;

                        }

                        catch(SecurityException se)

                        {

                                //                                 System.err.println("Dir exception: " + se.toString());

                                se.printStackTrace();

                                result = -1;

                        }

                        if(result == 1)

                        {

                                //                                System.out.println("DIR " + f.getName() + " created");

                        }

                }

                else if(f.exists())

                {

                        //                        System.out.println("Directory already exists, proceeding");

                        result = 2;

                }

                else

                {

                        result = -1;

                }

                return result;

        }

        public void createControlRoom(String ip, String port) throws IOException {

                controlRoom = new ControlChannel(ip, port, this);

        }

        public void createBackupChannel(String ip, String port) throws IOException {

                backupCh = new BackupChannel(ip, port, this);

        }

        public void createRestoreChannel(String ip, String port) throws IOException {

                restoreCh = new RestoreChannel(ip, port, this);

        }

        public void activateChannels() {

                controlRoom.start();

                backupCh.start();

                restoreCh.start();

        }

        public int getID() {

                return serverID;

        }

        public boolean storeChunk(Chunk c)

        {

                if(!this.storedChunks.containsKey(c.getChunkId() + ":" + c.getChunkNumber())) {

                        addChunk(c);

                        spaceRemaining -= c.getChunkSize();

                }

                return true;

        }

        public String getProtocolVersion() {

                return protocol_version;

        }

        public ConcurrentHashMap<String, Chunk> getStoredChunks()

        {

                return this.storedChunks;

        }

        public synchronized void registerNumberOfFileChunks(String fileID, Integer NFileChunks) {

                System.out.println(fileID + " registered.");

                numberOfFileChunks.put(fileID, NFileChunks);

        }

        public synchronized boolean insertAwaitingStoreFile(MessageHeader h) {

                String key = h.getFileId() + ':' + h.getChunkNumber();

                if(numberOfFileChunks.containsKey(h.getFileId())) {

                        colaborativePeers.put(key, new HashSet<Integer>());

                        numberOfChunksConfirmations.put(key, 0);

                        return true;

                }

                else {

                        System.out.println("Cloud hasn't got " + h.getFileId() + " registered.");

                        return false;

                }

        }

        public boolean wasTheRepDegreeMatched(Chunk c) {

                return c.getReplicationDegree() <= numberOfChunksConfirmations.get(c.getChunkId() + ':' + c.getChunkNumber());

        }

        public synchronized boolean intrepertStoredMessage(MessageHeader h) {

                String key = h.getFileId() + ':' + h.getChunkNumber();

                if(numberOfFileChunks.containsKey(h.getFileId())) {

                        if(colaborativePeers.get(key).add(Integer.parseInt(h.getSenderId()))) {

                                numberOfChunksConfirmations.put(key, numberOfChunksConfirmations.get(key) + 1);

                        }

                        return true;

                }

                else {

                        if(storedChunks.containsKey(key)) {

                                if(!colaborativePeers.containsKey(key)) {

                                        colaborativePeers.put(key, new HashSet<Integer>());

                                        colaborativePeers.get(key).add(getID());

                                }

                                colaborativePeers.get(key).add(Integer.parseInt(h.getSenderId()));

                                storedChunks.get(key).setActualReplicationDegree(colaborativePeers.get(key).size());

                                return true;

                        }

                        return false;

                }

        }

        public synchronized boolean deleteFile(MessageHeader h) {

                String fileId = h.getFileId();

                String path = Constants.CHUNKS_DIR + this.serverID + "/backup/" + fileId + "/";

                File dir = new File(path);

                boolean somethingDeleted = false;

                if(dir.isDirectory()) {

                        for(File f : dir.listFiles()) {

                                f.delete();

                                somethingDeleted = true;

                        }

                        if(!dir.delete()) {

                                System.out.println("Couldn't erase " + h.getFileId() + " from file system");

                                return false;

                        }

                }

                for(String key : storedChunks.keySet()) {

                        Chunk c = storedChunks.get(key);

                        if(c.getChunkId().equals(fileId)) {

                                storedChunks.remove(key);

                                spaceRemaining += c.getChunkSize();

                                somethingDeleted = true;

                        }

                }

                if(somethingDeleted) {

                        System.out.println(h.getFileId() + " erased from file system.");

                }

                return true;

        }

        public KeySetView<String, Integer> getBackedUpFileNames() {

                return numberOfFileChunks.keySet();

        }

        public String getPathNameFromFileId(String fileId) {

                if(storedChunks.containsKey(fileId + ":" + "0")) {

                        return storedChunks.get(fileId + ":" + "0").getFilePathName();

                }

                return null;

        }

        public int getBackedUpChunkNecessaryRepDegree(String fileId) {

                if(storedChunks.containsKey(fileId + ":" + "0")) {

                        return storedChunks.get(fileId + ":" + "0").getReplicationDegree();

                }

                return -1;

        }

        public int getBackepUpChunkPerceivedRepDegree(String fileId, int id) {

                if(colaborativePeers.containsKey(fileId + ":" + Integer.toString(id))) {

                        return colaborativePeers.get(fileId + ":" + Integer.toString(id)).size();

                }

                return 0;

        }

        public long getAvailableSpace() {

                return this.spaceRemaining;

        }

        public int getNumberOfStoredChunks() {

                return storedChunks.size();

        }

        public long getChunkSpace() {

                long total = 0;

                for(String key : storedChunks.keySet()) {

                        Chunk c = storedChunks.get(key);

                        total += c.getChunkSize();

                }

                return total;

        }

        public long getTotalSpace() {

                return this.spaceRemaining + getChunkSpace();

        }

        public void setAvailableSpace(long s) {

                this.spaceRemaining = s-getChunkSpace();

        }

        public void sendRemovedMessage(String fileId, String chunkNumber) {

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

                MessageHeader h;

                headerElements[0] = Constants.REMOVED;

                headerElements[1] = getProtocolVersion();

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

                headerElements[3] = fileId;

                headerElements[4] = chunkNumber;

                h = new MessageHeader(headerElements);

                controlRoom.sendHeader(h);

        }

        public void sendHeader(MessageHeader mh)

        {

                controlRoom.sendHeader(mh);

        }

        public boolean freeUpSpace(long desiredSpace) {

                String fileId, chunkNo, path;

                File file;

                for(String key : storedChunks.keySet()) {

                        Chunk c = storedChunks.get(key);

                        if(getChunkSpace() <= desiredSpace) {

                                break;

                        }

                        fileId = c.getChunkId();

                        chunkNo = c.getChunkNumber();

                        path = Constants.CHUNKS_DIR + this.serverID + "/backup/" + fileId + "/chk" + chunkNo;

                        file = new File(path);

                        if(file.delete()) {

                                sendRemovedMessage(fileId, chunkNo);

                                storedChunks.remove(key);

                        }

                }

                setAvailableSpace(desiredSpace);

                return true;

        }

        public boolean intrepertRemovedMessage(MessageHeader h) {

                String fileId, chunkNo, senderId, key;

                fileId = h.getFileId();

                chunkNo = h.getChunkNumber();

                senderId = h.getSenderId();

                key = fileId + ":" + chunkNo;

                if(checkIfPeerStoredFile(fileId)) {

                        if(numberOfChunksConfirmations.containsKey(key)) {

                                numberOfChunksConfirmations.put(key, numberOfChunksConfirmations.get(key) - 1);

                        }

                        if(this.colaborativePeers.containsKey(key)) {

                                this.colaborativePeers.get(key).remove(Integer.parseInt(senderId));

                        }

                        if(storedChunks.containsKey(key)) {

                                return storedChunks.get(key).getReplicationDegree() < numberOfChunksConfirmations.get(key);

                        }

                }

                return true;

        }

        public int getNumberOfChunks(String file_id)

        {

                if(numberOfFileChunks.containsKey(file_id)) {

                        return numberOfFileChunks.get(file_id);

                }

                return -1;

        }

        public Chunk getFileChunkRepDegree(String fileId, String chunkNo) {

                if(storedChunks.containsKey(fileId + ":" + chunkNo)) {

                        return storedChunks.get(fileId + ":" + chunkNo);

                }

                return null;

        }

        public void intrepertGetChunkMessage(MessageHeader header) {

                String key = header.getFileId() + ":" + header.getChunkNumber();

                Chunk toSend;

                SendChunkMessageProtocol p;

                if(storedChunks.containsKey(key)) {

                        System.out.println("Going to send chunk no " + header.getChunkNumber());

                        toSend = storedChunks.get(key);

                        p = new SendChunkMessageProtocol(toSend, this);

                        p.run();

                }

                else {

                        System.out.println("I dont have that chunk");

                }

        }

        @SuppressWarnings({ "unchecked" })

        public void loadFromDisk() throws IOException, Exception

        {

                String path = Constants.DATABASE_DIR + "peer" + this.serverID + '/';

                String path1 = path + "storedChunks.ser";

                Path p = Paths.get(path1);

                if (Files.exists(p)) {

                        FileInputStream fis = new FileInputStream(path1);

                        ObjectInputStream ois = new ObjectInputStream(fis);

                        this.setStoredChunks( (ConcurrentHashMap<String, Chunk>)ois.readObject());

                        fis.close();

                        ois.close();

                }

                String path2 = path + "numberOfFileChunks.ser";

                Path p2 = Paths.get(path2);

                if (Files.exists(p2)) {

                        FileInputStream fis2 = new FileInputStream(path2);

                        ObjectInputStream ois2 = new ObjectInputStream(fis2);

                        this.setNumberOfFileChunks((ConcurrentHashMap<String, Integer>)ois2.readObject());

                        fis2.close();

                        ois2.close();

                }

                String path3 = path + "colaborativePeers.ser";

                Path p3 = Paths.get(path3);

                if (Files.exists(p3)) {

                        FileInputStream fis3 = new FileInputStream(path3);

                        ObjectInputStream ois3 = new ObjectInputStream(fis3);

                        this.setColaborativePeers((ConcurrentHashMap<String, HashSet<Integer>>)ois3.readObject());

                        fis3.close();

                        ois3.close();

                }

                String path4 = path + "numberOfChunksConfirmations.ser";

                Path p4 = Paths.get(path4);

                if (Files.exists(p4)) {

                        FileInputStream fis4 = new FileInputStream(path4);

                        ObjectInputStream ois4 = new ObjectInputStream(fis4);

                        this.setNumberOfChunksConfirmations((ConcurrentHashMap<String, Integer>)ois4.readObject());

                        fis4.close();

                        ois4.close();

                }

                String path5 = path + "chunksToRestore.ser";

                Path p5 = Paths.get(path5);

                if (Files.exists(p5)) {

                        FileInputStream fis5 = new FileInputStream(path5);

                        ObjectInputStream ois5 = new ObjectInputStream(fis5);

                        this.setChunksToRestore((ConcurrentHashMap<String, Integer>)ois5.readObject());

                        fis5.close();

                        ois5.close();

                }

                String path6 = path + "recoveredChunks.ser";

                Path p6 = Paths.get(path6);

                if (Files.exists(p6)) {

                        FileInputStream fis6 = new FileInputStream(path6);

                        ObjectInputStream ois6 = new ObjectInputStream(fis6);

                        this.setRecoveredChunks((ConcurrentHashMap<String, Chunk>)ois6.readObject());

                        fis6.close();

                        ois6.close();

                }

                String path7 = path + "numberOfFileChunksToBeRestored.ser";

                Path p7 = Paths.get(path7);

                if (Files.exists(p7)) {

                        FileInputStream fis7 = new FileInputStream(path7);

                        ObjectInputStream ois7 = new ObjectInputStream(fis7);

                        this.setNumberOfFileChunksToBeRestored((ConcurrentLinkedQueue<String>)ois7.readObject());

                        fis7.close();

                        ois7.close();

                }

        }

        @SuppressWarnings("resource")

        public synchronized void saveToDisk() throws IOException

        {

                String path = Constants.DATABASE_DIR + "peer" + this.serverID + '/';

                checkOrCreateDirectory(path);

                String path1 = path + "storedChunks.ser";

                FileOutputStream fos1 = new FileOutputStream(path1);

                ObjectOutputStream oos1 = new ObjectOutputStream(fos1);

                oos1.writeObject(storedChunks);

                String path2 = path + "numberOfFileChunks.ser";

                FileOutputStream fos2 = new FileOutputStream(path2);

                ObjectOutputStream oos2 = new ObjectOutputStream(fos2);

                oos2.writeObject(numberOfFileChunks);

                String path3 = path + "colaborativePeers.ser";

                FileOutputStream fos3 = new FileOutputStream(path3);

                ObjectOutputStream oos3 = new ObjectOutputStream(fos3);

                oos3.writeObject(colaborativePeers);

                String path4 = path + "numberOfChunksConfirmations.ser";

                FileOutputStream fos4 = new FileOutputStream(path4);

                ObjectOutputStream oos4 = new ObjectOutputStream(fos4);

                oos4.writeObject(numberOfChunksConfirmations);

                String path5 = path + "chunksToRestore.ser";

                FileOutputStream fos5 = new FileOutputStream(path5);

                ObjectOutputStream oos5 = new ObjectOutputStream(fos5);

                oos5.writeObject(chunksToRestore);

                String path6 = path + "recoveredChunks.ser";

                FileOutputStream fos6 = new FileOutputStream(path6);

                ObjectOutputStream oos6 = new ObjectOutputStream(fos6);

                oos6.writeObject(recoveredChunks);

                String path7 = path + "numberOfFileChunksToBeRestored.ser";

                FileOutputStream fos7 = new FileOutputStream(path7);

                ObjectOutputStream oos7 = new ObjectOutputStream(fos7);

                oos7.writeObject(numberOfFileChunksToBeRestored);

        }

        public void setStoredChunks(ConcurrentHashMap<String, Chunk> sc)

        {

                storedChunks = sc;

        }

        public void setNumberOfFileChunks(ConcurrentHashMap<String, Integer> si)

        {

                numberOfFileChunks = si;

        }

        public void setColaborativePeers(ConcurrentHashMap<String, HashSet<Integer>> sh)

        {

                colaborativePeers = sh;

        }

        public void setNumberOfChunksConfirmations(ConcurrentHashMap<String, Integer> si)

        {

                numberOfChunksConfirmations = si;

        }

        public void setChunksToRestore(ConcurrentHashMap<String, Integer> si)

        {

                chunksToRestore = si;

        }

        public void setRecoveredChunks(ConcurrentHashMap<String, Chunk> sc)

        {

                recoveredChunks = sc;

        }

        public void setNumberOfFileChunksToBeRestored(ConcurrentLinkedQueue<String> s)

        {

                numberOfFileChunksToBeRestored = s;

        }

}