sdis1819-t7g02

<?xml version="1.0" encoding="UTF-8"?>

<projectDescription>

	<name>service</name>

	<comment></comment>

	<projects>

	</projects>

	<buildSpec>

		<buildCommand>

			<name>org.eclipse.jdt.core.javabuilder</name>

			<arguments>

			</arguments>

		</buildCommand>

	</buildSpec>

	<natures>

		<nature>org.eclipse.jdt.core.javanature</nature>

	</natures>

</projectDescription>

package service;

import java.util.ArrayList;

public class Chunk implements java.io.Serializable {

	private int chunkNr;

    private String fileID;

    private String filePathName = null;

    private int size;

    private int actual_replication_degree = 0;

    private int desired_replication_degree;

    private byte[] chunkContent;

//    each chunk is identified by the pair <fileID, chunkNo>

    public Chunk(int n, byte[] c, String s, int desiredRepDegree) throws Exception

    {

      this.chunkNr = n;

      this.chunkContent = getFiltredData(c);

      if(chunkContent.length > Constants.MAX_CHUNK_SIZE) {

    	  throw new Exception("Invalid chunk size -> " + chunkContent.length);

      }

      this.size = chunkContent.length;

      desired_replication_degree = desiredRepDegree;

      fileID = s;

    }

    private byte[] getFiltredData(byte[] c) {

    	ArrayList<Byte> aux = new ArrayList<Byte>();

    	boolean toAdd = false;

		for(byte b : c) {

			aux.add(b);

		}

		byte[] toReturn = new byte[aux.size()];

		for(int i = aux.size() - 1, k = 0; i >= 0 ; i--) {

			if(aux.get(i) == 0) {

				if(!toAdd) {

					continue;

				}

				else {

					toReturn[k] = aux.get(i);

					k++;

				}

			}

			else {

				toReturn[k] = aux.get(i);

				toAdd = true;

				k++;

			}	

		}

		return toReturn;

	}

	public int getReplicationDegree() {

    	return desired_replication_degree;

    }

    public void setActualReplicationDegree(int rd)

    {

      this.actual_replication_degree = rd;

    }

    public String getChunkNumber()

    {

      return Integer.toString(this.chunkNr);

    }

    public String getChunkId()

    {

      return this.fileID;

    }

    public int getChunkSize()

    {

      return this.size;

    }

    public int getActualReplicationDegree()

    {

      return this.actual_replication_degree;

    }

    public byte[] getChunkContent()

    {

      return this.chunkContent;

    }

    public int compareTo(Chunk c)

    {

         return(this.chunkNr - c.chunkNr);

    }

	@Override

	public boolean equals(Object obj) {

		if (this == obj) {

			return true;

		}

		if (obj == null) {

			return false;

		}

		if (!(obj instanceof Chunk)) {

			return false;

		}

		Chunk other = (Chunk) obj;

		if (chunkNr != other.chunkNr) {

			return false;

		}

		if (fileID == null) {

			if (other.fileID != null) {

				return false;

			}

		} else if (!fileID.equals(other.fileID)) {

			return false;

		}

		return true;

	}

	public String getFilePathName() {

		return filePathName;

	}

	public void setFilePathName(String filePathName) {

		this.filePathName = filePathName;

	}

}

package service;

import java.rmi.registry.LocateRegistry;

import java.rmi.registry.Registry;

public class Client {

    private Client() {}

    public static void main(String[] args) { 

    	System.out.println("CLIENT");

    	try 

        {      	

        	int argc = args.length;

            if(argc < 2 || argc > 4)

            {

              print_usage();

              return;

            }

            String aux = args[0];

            String[] acess_point = aux.split(":");

            String ip = acess_point[0];

            String id = acess_point[1];

            String sub_protocol = args[1];

            Registry registry = LocateRegistry.getRegistry(ip);

            RMI rmi = (RMI) registry.lookup(id);

            String file_path;

            String response;

            switch(sub_protocol)

            {

              case "BACKUP":

                if(argc != 4)

                {

                	print_error_ops(sub_protocol);

                }

                file_path = args[2];

                int replication_degree = Integer.parseInt(args[3]);

                response = rmi.backup(file_path, replication_degree);

                System.out.println("response: " + response);

                break;

              case "RESTORE":

                if(argc != 3)

                {

                	print_error_ops(sub_protocol);

                }

                file_path = args[2];

                response = rmi.restore(file_path);

                System.out.println("response: " + response);

                break;

              case "DELETE":

                if(argc != 3)

                {

                	print_error_ops(sub_protocol);

                }

                file_path = args[2];

                response = rmi.delete(file_path);

                System.out.println("response: " + response);

                break;

              case "RECLAIM":

                if(argc != 3)

                {

                  print_error_ops(sub_protocol);

                }

                long disk_space = Long.parseLong(args[2]);

                response = rmi.reclaim(disk_space);

                System.out.println("response: " + response);

                break;

              case "STATE":

                if(argc != 2)

                {

                	print_error_ops(sub_protocol);

                }

                response = rmi.state();

                System.out.println("response: " + response);

                break;

            }

        } catch (Exception e) {

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

        }

    }

    public static void print_usage()

    {

      System.out.println("Usage: :\n");

      System.out.println("  java Client <peer_ap> <operation> <opnd_1> <opnd_2> \n");

      System.out.println("  - <peer_ap> -> peer's acess point\n");

      System.out.println("  - <operation> -> Subprotocol ops: BACKUP, RESTORE, DELETE, RECLAIM;  Enhancements: ops with ENH in the end. State: STATE \n");

      System.out.println("  - <opnd_1> -> path name of the file or in case of op RECLAIM the max amout of disk space\n");

      System.out.println("  - <opnd_2>* -> integer that specifies the desired replication degree\n");

    }

    public static void print_error_ops(String error)

    {

      switch(error)

      {

        case "BACKUP":

          System.out.println("ERROR on BACKUP format. Must be:\n");

          System.out.println("> BACKUP <file_path> <replication_degree>");

          break;

        case "RESTORE":

          System.out.println("ERROR on RESTORE format. Must be:\n");

          System.out.println("> RESTORE <file_path>");

          break;

        case "DELETE":

          System.out.println("ERROR on DELETE format. Must be:\n");

          System.out.println("> DELETE <file_path>");

          break;

        case "RECLAIM":

          System.out.println("ERROR on RECLAIM format. Must be:\n");

          System.out.println("> RECLAIM <disk_space>");

          break;

        case "STATE":

          System.out.println("ERROR on STATE format. Must be:\n");

          System.out.println("> RECLAIM <disk_space>");

          break;

         default:

           System.out.println("ERROR. Unkown op");

      }

    }

}

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;

	}

}

package channels;

import java.io.IOException;

import protocols.SendSoredMessageProtocol;

import service.Cloud;

import service.Constants;

public class BackupChannel extends Channel {

	private Message receivedMessage;

	public BackupChannel(String addr, String p, Cloud f) throws IOException {

		super(addr, p, f.getID(), f);

	}

	public void printLog(String msg) {

		System.out.println("MDB " + serverId + msg);

	}

//	This message is used to ensure that the chunk is backed up with the desired replication degree as follows. The initiator-peer collects the confirmation messages during a time interval of one second. If the number of confirmation messages it received up to the end of that interval is lower than the desired replication degree, it retransmits the backup message on the MDB channel, and doubles the time interval for receiving confirmation messages. This procedure is repeated up to a maximum number of five times, i.e. the initiator will send at most 5 PUTCHUNK messages per chunk.

//

//	Hint: Because UDP is not reliable, a peer that has stored a chunk must reply with a STORED message to every PUTCHUNK message it receives. Furthermore, the initiator-peer needs to keep track of which peers have responded.

//

//	A peer should also count the number of confirmation messages for each of the chunks it has stored and keep that count in non-volatile memory. This information can be useful if the peer runs out of disk space: in that event, the peer may try to free some space by evicting chunks whose actual replication degree is higher than the desired replication degree.

	@Override

	public void run () {

		printLog(": UP");

		int messageSenderId = -1;

		String messageType;

		try {

			joinChannelGroup();

		} catch (IOException e1) {

			e1.printStackTrace();

		}

		do {

			try {

				receivedMessage = captureData();

				if(receivedMessage == null) {

					printLog(" passing message.");

					continue;

				}

				messageType = receivedMessage.getHeader().getMessageType();

				messageSenderId = Integer.parseInt(receivedMessage.getHeader().getSenderId());

				if(messageType.equals(Constants.PUTCHUNK) && messageSenderId != getServerId()) {

					//printLog(" received: " + receivedMessage.getHeader().toString());

					if(father.storeChunk(receivedMessage)) {

						SendSoredMessageProtocol mp = new SendSoredMessageProtocol(buildMessage(), father);

						mp.start();

					}

					else {

						printLog(" ERROR: couldn't store chunk");

					}

				}

				else {

					//printLog(" ERROR: Invalid message header received -> " + receivedMessage.getHeader().toString());

				}

//				leaveGroupChannel();			

			}

			catch (IOException e) {

				printLog(" ERROR: " + "IOException -> " + e.getMessage());

			}

		} while(true);

	}

	public MessageHeader buildMessage() {

		MessageHeader response;

		response = buildResponseMessage();

		return response;

	}

	public MessageHeader buildResponseMessage() {

		MessageHeader response;

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

	    headerElements[0] = Constants.STORED;

	    headerElements[1] = father.getProtocolVersion();

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

	    headerElements[3] = receivedMessage.getHeader().getFileId();

	    headerElements[4] = receivedMessage.getHeader().getChunkNumber();

	    response = new MessageHeader(headerElements);

		return response;

	}

}

package service;

public class Constants {

	public static final int MAX_CHUNK_SIZE = 64 * 1000;

	public static final int MAX_MESSAGE_SIZE = MAX_CHUNK_SIZE + 1024;

	public static final String CHUNKS_DIR = "../Project 1 -- Distributed Backup Service/peers/peer";

	public static final String DATABASE_DIR = "../Project 1 -- Distributed Backup Service/database/";

	public static final int MAX_NUMBER_OF_THREADS = 8;

	public static final String NORMAL_VERSION = "1.0";

	// Available header types

	public static final String PUTCHUNK = "PUTCHUNK";

	public static final String STORED = "STORED";

	public static final String GETCHUNK = "GETCHUNK";

	public static final String CHUNK = "CHUNK";

	public static final String DELETE = "DELETE";

	public static final String REMOVED = "REMOVED";

	// headers necessary arguments

	public static final int PUTCHUNK_N_ARGS = 6;

	public static final int STORED_N_ARGS = 5;

	public static final int GETCHUNK_N_ARGS = 5;

	public static final int CHUNK_N_ARGS = 5;

	public static final int DELETE_N_ARGS = 4;

	public static final int REMOVED_N_ARGS = 5;

	//'0xD''0xA'

	public static final byte CR = 0xD;

	public static final byte LF = 0xA;

	public static final int MAX_PUTCHUNK_MESSAGES = 5;

}

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;