Handling file transfer progress (#39)

* Modifying how the sockets connections is processed * Adding factory changes * Excluding communication api dependency * Removing unnecesary files * Updating dhash layer * Adding dhash changes * Adding new transfer communication methods * Adding receive method * Integration tests worked * Defining MessageStream and resources classes * Changing resources approach, removing checksum * New classes and serialization is working, integration tests passes * Async operations. Integrations tests passed * Refactoring * Refactoring * Refactoring * Refactoring * Message refactoring * Changing log api * Refactoring. Integration tests working * Adding close to Resource interface * Adding unit tests * Adding unit tests * Using LocalFileResource * Defining versions and publishing * Updating readme
estigma88 · Oct 5, 2018 · e1eb9a1 · e1eb9a1
1 parent 05e9f89
commit e1eb9a1
Show file tree

Hide file tree

Showing 137 changed files with 2,781 additions and 2,843 deletions.
diff --git a/README.md b/README.md
@@ -13,7 +13,7 @@ jDHTUQ is a peer-to-peer DHT system based on Chord algorithm, but built to gener
 
 ## Download last versions
 
-[![Github Releases (by Asset)](https://img.shields.io/github/downloads/estigma88/jdhtuq/latest/desktop-structure-gui-2.0.2.jar.svg)](https://github.com/estigma88/jdhtuq/releases/download/v2.0.4/desktop-structure-gui-2.0.2.jar)     [![Github Releases (by Asset)](https://img.shields.io/github/downloads/estigma88/jdhtuq/latest/desktop-network-gui-2.0.4.jar.svg)](https://github.com/estigma88/jdhtuq/releases/download/v2.0.4/desktop-network-gui-2.0.4.jar)     [![Github Releases (by Asset)](https://img.shields.io/github/downloads/estigma88/jdhtuq/latest/standalone-network-2.0.4.jar.svg)](https://github.com/estigma88/jdhtuq/releases/download/standalone-network-v2.0.4/standalone-network-2.0.4.jar)
+[![Github Releases (by Asset)](https://img.shields.io/github/downloads/estigma88/jdhtuq/desktop-structure-gui-v2.0.3/desktop-structure-gui-2.0.3.jar.svg)](https://github.com/estigma88/jdhtuq/releases/download/desktop-structure-gui-v2.0.3/desktop-structure-gui-2.0.3.jar)     [![Github Releases (by Asset)](https://img.shields.io/github/downloads/estigma88/jdhtuq/desktop-network-gui-v2.0.5/desktop-network-gui-2.0.5.jar.svg)](https://github.com/estigma88/jdhtuq/releases/download/desktop-network-gui-v2.0.5/desktop-network-gui-2.0.5.jar)     [![Github Releases (by Asset)](https://img.shields.io/github/downloads/estigma88/jdhtuq/standalone-network-v2.0.5/standalone-network-2.0.5.jar.svg)](https://github.com/estigma88/jdhtuq/releases/download/standalone-network-v2.0.5/standalone-network-2.0.5.jar)
 
 Data structure and network applications.
 - Execute with double click or

diff --git a/communication/communication-api/build.gradle b/communication/communication-api/build.gradle
@@ -9,13 +9,11 @@ repositories {
 }
 
 group = 'com.github.estigma88'
-version = '2.0.0'
+version = '3.0.0'
 
 
 dependencies {
     compileOnly 'org.projectlombok:lombok:1.16.20'
-
-    compile group: 'log4j', name: 'log4j', version: '1.2.16'
 }
 
 task sourcesJar(type: Jar) {

diff --git a/...munication/message/SequenceGenerator.java → ...ls/communication/message/IdGenerator.java b/...munication/message/SequenceGenerator.java → ...ls/communication/message/IdGenerator.java
@@ -21,11 +21,11 @@
 /**
  * Sequence generator
  */
-public interface SequenceGenerator {
+public interface IdGenerator {
     /**
      * Get the next sequence value
      *
      * @return next sequence value
      */
-    long getSequenceNumber();
+    String newId();
 }
diff --git a/...ommunication-api/src/main/java/co/edu/uniquindio/utils/communication/message/Message.java b/...ommunication-api/src/main/java/co/edu/uniquindio/utils/communication/message/Message.java
@@ -64,48 +64,27 @@ public enum SendType {
     @Singular
     private final Map<String, String> params;
 
-    /**
-     * Hash map of names with datas
-     */
-    @Singular
-    private final Map<String, byte[]> datas;
-
     /**
      * Sequence number
      */
-    private final long sequenceNumber;
-
+    private final String id;
+
+    public static Message.MessageBuilder with(Message message){
+        return Message.builder()
+                .id(message.id)
+                .sendType(message.sendType)
+                .messageType(message.messageType)
+                .address(message.address)
+                .params(message.params);
+    }
 
     /**
      * This method is used for getting a specific data from the message
      *
      * @return Returns the data that is stored in the given position
      */
     public String getParam(String name) {
-        if (!params.containsKey(name)) {
-            throw new IllegalArgumentException("The message type "
-                    + messageType.getName() + " not contains param '" + name
-                    + "'");
-        } else {
-            return params.get(name);
-        }
-    }
-
-
-    /**
-     * Gets data by name
-     *
-     * @param name Data name
-     * @return Data
-     */
-    public byte[] getData(String name) {
-        if (!datas.containsKey(name)) {
-            throw new IllegalArgumentException("The big message type "
-                    + messageType.getName() + " not contains param '" + name
-                    + "'");
-        } else {
-            return datas.get(name);
-        }
+        return params.get(name);
     }
 
     /**

diff --git a/...cation-api/src/main/java/co/edu/uniquindio/utils/communication/message/MessageStream.java b/...cation-api/src/main/java/co/edu/uniquindio/utils/communication/message/MessageStream.java
@@ -0,0 +1,21 @@
+package co.edu.uniquindio.utils.communication.message;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+
+@Builder
+@Data
+public class MessageStream implements Closeable{
+    private final Message message;
+    private transient final InputStream inputStream;
+    private final Long size;
+
+    @Override
+    public void close() throws IOException {
+        this.inputStream.close();
+    }
+}
diff --git a/...cation/message/SequenceGeneratorImpl.java → .../communication/message/UUIDGenerator.java b/...cation/message/SequenceGeneratorImpl.java → .../communication/message/UUIDGenerator.java
@@ -18,12 +18,14 @@
 
 package co.edu.uniquindio.utils.communication.message;
 
+import java.util.UUID;
+
 /**
- * The {@code SequenceGenerator} class is responsible for generating the sequence
+ * The {@code IdGenerator} class is responsible for generating the sequence
  * numbers for the messages. This create a sequence number between 1 and Long.MAX_VALUE.
  * When the sequence number is bigger than Long.MAX_VALUE the sequence number starts at 0
  * again. The sequence number increments in one every time that {@code
- * SequenceGenerator.getSequenceNumber()} is called
+ * IdGenerator.newId()} is called
  * 
  * 
  * @author Daniel Pelaez
@@ -33,27 +35,14 @@
  * @since 1.0
  * 
  */
-public class SequenceGeneratorImpl implements SequenceGenerator{
-	/**
-	 * The next sequence number, star at 0
-	 */
-	private static long number = 0;
-
-	/**
-	 * The max number that SequenceGenerator can generate, its value is Long.MAX_VALUE
-	 */
-	private static long maxNumber = Long.MAX_VALUE;
-
+public class UUIDGenerator implements IdGenerator {
 	/**
 	 * This method return the next sequence number
 	 * 
 	 * @return Returns the next sequence number
 	 */
-	public synchronized long getSequenceNumber() {
-		if (number > maxNumber) {
-			number = 0;
-		}
-
-		return ++number;
+	@Override
+	public String newId() {
+		return UUID.randomUUID().toString();
 	}
 }
diff --git a/...pi/src/main/java/co/edu/uniquindio/utils/communication/transfer/CommunicationManager.java b/...pi/src/main/java/co/edu/uniquindio/utils/communication/transfer/CommunicationManager.java
@@ -20,6 +20,11 @@
 
 import co.edu.uniquindio.utils.communication.Observer;
 import co.edu.uniquindio.utils.communication.message.Message;
+import co.edu.uniquindio.utils.communication.message.MessageStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 
 /**
  * The {@code CommunicationManager} interface is used to send messages
@@ -31,7 +36,7 @@
  * @version 1.0.2, 17/06/2010
  * @since 1.0.2
  */
-public interface CommunicationManager {
+public interface CommunicationManager extends StreamManager{
 
     /**
      * Creates and sends a message specifying its type, the type of the response
@@ -41,7 +46,7 @@ public interface CommunicationManager {
      * @param typeReturn The type of the response
      * @return An object <T> of the specified type.
      */
-    <T> T sendMessageUnicast(Message message, Class<T> typeReturn);
+    <T> T send(Message message, Class<T> typeReturn);
 
     /**
      * Creates and sends a message specifying its type, the type of the response
@@ -52,42 +57,43 @@ public interface CommunicationManager {
      * @param paramNameResult Param name of result
      * @return An object <T> of the specified type.
      */
-    <T> T sendMessageUnicast(Message message, Class<T> typeReturn,
-                             String paramNameResult);
+    <T> T send(Message message, Class<T> typeReturn,
+               String paramNameResult);
+
 
     /**
      * Sends a message specifying its type, the type of the response and the
      * data. Used Communicator instance called unicastManager to send message
      *
      * @param message Messages to send
      */
-    void sendMessageUnicast(Message message);
+    void send(Message message);
 
     /**
      * Creates and sends a multicast message specifying its type, the type of
      * the response and the data. Used
-     * <code>sendMessageMultiCast(message)</code> to send message
+     * <code>sendMultiCast(message)</code> to send message
      *
      * @param <T>        Type return
      * @param message    Message
      * @param typeReturn Type return
      * @return Response
      */
-    <T> T sendMessageMultiCast(Message message, Class<T> typeReturn);
+    <T> T sendMultiCast(Message message, Class<T> typeReturn);
 
     /**
      * Creates and sends a multicast message specifying its type, the type of
      * the response and the data. Used
-     * <code>sendMessageMultiCast(message)</code> to send message
+     * <code>sendMultiCast(message)</code> to send message
      *
      * @param <T>             Type return
      * @param message         Message
      * @param typeReturn      Type return
      * @param paramNameResult Param name of result
      * @return Response
      */
-    <T> T sendMessageMultiCast(Message message, Class<T> typeReturn,
-                               String paramNameResult);
+    <T> T sendMultiCast(Message message, Class<T> typeReturn,
+                        String paramNameResult);
 
     /**
      * Sends a multicast message specifying its type, the type of the response
@@ -96,12 +102,7 @@ <T> T sendMessageMultiCast(Message message, Class<T> typeReturn,
      *
      * @param message Messages to send
      */
-    void sendMessageMultiCast(Message message);
-
-    /**
-     * Stop all process
-     */
-    void stopAll();
+    void sendMultiCast(Message message);
 
     /**
      * Adds observer to communication
@@ -124,6 +125,28 @@ <T> T sendMessageMultiCast(Message message, Class<T> typeReturn,
      */
     void removeObserver(String name);
 
+
+    /**
+     * Adds observer to communication
+     *
+     * @param observer Observer to add
+     */
+    void addStreamObserver(Observer<MessageStream> observer);
+
+    /**
+     * Remove observer to communication
+     *
+     * @param observer Observer to remove
+     */
+    void removeStreamObserver(Observer<MessageStream> observer);
+
+    /**
+     * Remove observer by name
+     *
+     * @param name Observer name
+     */
+    void removeStreamObserver(String name);
+
     /**
      * Add a message processor to handle any kind of message
      *
@@ -132,15 +155,35 @@ <T> T sendMessageMultiCast(Message message, Class<T> typeReturn,
      */
     void addMessageProcessor(String name, MessageProcessor messageProcessor);
 
+    /**
+     * Add a message processor to handle any kind of message with additional input stream
+     *
+     * @param name             of the message processor
+     */
+    void addMessageStreamProcessor(String name, MessageStreamProcessor messageStreamProcessor);
+
     /**
      * Remove a message processor
      *
      * @param name of the message processor
      */
     void removeMessageProcessor(String name);
 
+    /**
+     * Remove a message processor
+     *
+     * @param name of the message processor
+     */
+    void removeMessageStreamProcessor(String name);
+
     /**
      * Initialize communication manager.
      */
     void init();
+
+    /**
+     * Stop all process
+     */
+    void stopAll();
+
 }
diff --git a/...n-api/src/main/java/co/edu/uniquindio/utils/communication/transfer/ConnectionHandler.java b/...n-api/src/main/java/co/edu/uniquindio/utils/communication/transfer/ConnectionHandler.java
@@ -0,0 +1,7 @@
+package co.edu.uniquindio.utils.communication.transfer;
+
+import java.net.Socket;
+
+public interface ConnectionHandler {
+    void handle(Socket socket);
+}
diff --git a/...quindio/dhash/resource/BytesResource.java → ...tion/transfer/MessageStreamProcessor.java b/...quindio/dhash/resource/BytesResource.java → ...tion/transfer/MessageStreamProcessor.java
@@ -1,5 +1,5 @@
 /*
- *  DHash project implement a storage management
+ *  Communication project implement communication point to point and multicast
  *  Copyright (C) 2010 - 2018  Daniel Pelaez
  *
  *  This program is free software: you can redistribute it and/or modify
@@ -16,28 +16,14 @@
  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-package co.edu.uniquindio.dhash.resource;
+package co.edu.uniquindio.utils.communication.transfer;
 
-import co.edu.uniquindio.storage.resource.Resource;
+import co.edu.uniquindio.utils.communication.message.Message;
+import co.edu.uniquindio.utils.communication.message.MessageStream;
 
-import java.io.Serializable;
+import java.io.InputStream;
+import java.io.OutputStream;
 
-public class BytesResource implements Resource, Serializable {
-
-    private final String id;
-    private final byte[] bytes;
-
-    public BytesResource(String id, byte[] bytes) {
-        this.id = id;
-        this.bytes = bytes;
-    }
-
-    @Override
-    public String getId() {
-        return id;
-    }
-
-    public byte[] getBytes() {
-        return bytes;
-    }
+public interface MessageStreamProcessor {
+    MessageStream process(MessageStream messageStream);
 }
diff --git a/.../src/main/java/co/edu/uniquindio/utils/communication/transfer/ProgressStatusTransfer.java b/.../src/main/java/co/edu/uniquindio/utils/communication/transfer/ProgressStatusTransfer.java
@@ -0,0 +1,5 @@
+package co.edu.uniquindio.utils.communication.transfer;
+
+public interface ProgressStatusTransfer {
+    void status(String name, Long current, Long size);
+}
diff --git a/...ation-api/src/main/java/co/edu/uniquindio/utils/communication/transfer/StreamManager.java b/...ation-api/src/main/java/co/edu/uniquindio/utils/communication/transfer/StreamManager.java
@@ -0,0 +1,10 @@
+package co.edu.uniquindio.utils.communication.transfer;
+
+import co.edu.uniquindio.utils.communication.message.Message;
+import co.edu.uniquindio.utils.communication.message.MessageStream;
+
+public interface StreamManager {
+    MessageStream receive(Message message, ProgressStatusTransfer progressStatusTransfer);
+
+    void send(MessageStream messageStream, ProgressStatusTransfer progressStatusTransfer);
+}