Content Sending working between nodes

src/kademlia/core/KadServer.java

@@ -118,12 +118,6 @@ public class KadServer
 
     private void sendMessage(Node to, Message msg, int comm) throws IOException
     {
-
-        Class c = msg.getClass();
-        System.out.println(c.getSimpleName());
-        System.out.println(c.getName());
-
-
         /* Setup the message for transmission */
         ByteArrayOutputStream bout = new ByteArrayOutputStream();
         DataOutputStream dout = new DataOutputStream(bout);

src/kademlia/message/AcknowledgeMessage.java

@@ -6,7 +6,9 @@
 package kademlia.message;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import kademlia.node.Node;
 
@@ -21,19 +23,19 @@ public class AcknowledgeMessage implements Message
         this.origin = origin;
     }
 
-    public AcknowledgeMessage(DataInput in) throws IOException
+    public AcknowledgeMessage(DataInputStream in) throws IOException
     {
         this.fromStream(in);
     }
 
     @Override
-    public final void fromStream(DataInput in) throws IOException
+    public final void fromStream(DataInputStream in) throws IOException
     {
         this.origin = new Node(in);
     }
 
     @Override
-    public void toStream(DataOutput out) throws IOException
+    public void toStream(DataOutputStream out) throws IOException
     {
         origin.toStream(out);
     }

src/kademlia/message/ConnectMessage.java

@@ -6,7 +6,9 @@
 package kademlia.message;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import kademlia.node.Node;
 
@@ -21,19 +23,19 @@ public class ConnectMessage implements Message
         this.origin = origin;
     }
 
-    public ConnectMessage(DataInput in) throws IOException
+    public ConnectMessage(DataInputStream in) throws IOException
     {
         this.fromStream(in);
     }
 
     @Override
-    public final void fromStream(DataInput in) throws IOException
+    public final void fromStream(DataInputStream in) throws IOException
     {
         this.origin = new Node(in);
     }
 
     @Override
-    public void toStream(DataOutput out) throws IOException
+    public void toStream(DataOutputStream out) throws IOException
     {
         origin.toStream(out);
     }

src/kademlia/message/ContentStoreMessage.java

@@ -1,7 +1,9 @@
 package kademlia.message;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import kademlia.dht.KadContent;
 import kademlia.node.Node;
 
@@ -36,13 +38,13 @@ public class ContentStoreMessage implements Message
     }
 
     @Override
-    public void fromStream(DataInput in)
+    public void fromStream(DataInputStream in)
     {
 
     }
 
     @Override
-    public void toStream(DataOutput out)
+    public void toStream(DataOutputStream out)
     {
         /* @todo write the origin and the content to the stream */
     }

src/kademlia/message/MessageFactory.java

@@ -1,16 +1,17 @@
-/**
- * @author Joshua
- * @created
- * @desc
- */
 package kademlia.message;
 
-import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.IOException;
 import kademlia.core.KadServer;
 import kademlia.node.Node;
 import kademlia.operation.Receiver;
 
+/**
+ * Handles creating messages and receivers
+ *
+ * @author Joshua Kissoon
+ * @since 20140202
+ */
 public class MessageFactory
 {
 
@@ -21,7 +22,7 @@ public class MessageFactory
         this.localNode = local;
     }
 
-    public Message createMessage(byte code, DataInput in) throws IOException
+    public Message createMessage(byte code, DataInputStream in) throws IOException
     {
         switch (code)
         {
@@ -35,6 +36,8 @@ public class MessageFactory
                 return new NodeReplyMessage(in);
             case NodeLookupMessage.CODE:
                 return new NodeLookupMessage(in);
+            case StoreContentMessage.CODE:
+                return new StoreContentMessage(in);
             default:
                 System.out.println("No Message handler found for message. Code: " + code);
                 return new SimpleMessage(in);
@@ -53,6 +56,8 @@ public class MessageFactory
                 return new ConnectReceiver(server, this.localNode);
             case NodeLookupMessage.CODE:
                 return new NodeLookupReceiver(server, this.localNode);
+            case StoreContentMessage.CODE:
+                return new StoreContentReceiver(server, this.localNode);
         }
     }
 }

src/kademlia/message/NodeLookupMessage.java

@@ -6,7 +6,9 @@
 package kademlia.message;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import kademlia.node.Node;
 import kademlia.node.NodeId;
@@ -31,20 +33,20 @@ public class NodeLookupMessage implements Message
         this.lookupId = lookup;
     }
 
-    public NodeLookupMessage(DataInput in) throws IOException
+    public NodeLookupMessage(DataInputStream in) throws IOException
     {
         this.fromStream(in);
     }
 
     @Override
-    public final void fromStream(DataInput in) throws IOException
+    public final void fromStream(DataInputStream in) throws IOException
     {
         this.origin = new Node(in);
         this.lookupId = new NodeId(in);
     }
 
     @Override
-    public void toStream(DataOutput out) throws IOException
+    public void toStream(DataOutputStream out) throws IOException
     {
         this.origin.toStream(out);
         this.lookupId.toStream(out);

src/kademlia/message/NodeReplyMessage.java

@@ -6,7 +6,9 @@
 package kademlia.message;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
 import kademlia.node.Node;
@@ -24,13 +26,13 @@ public class NodeReplyMessage implements Message
         this.nodes = nodes;
     }
 
-    public NodeReplyMessage(DataInput in) throws IOException
+    public NodeReplyMessage(DataInputStream in) throws IOException
     {
         this.fromStream(in);
     }
 
     @Override
-    public final void fromStream(DataInput in) throws IOException
+    public final void fromStream(DataInputStream in) throws IOException
     {
         /* Read in the origin */
         this.origin = new Node(in);
@@ -47,7 +49,7 @@ public class NodeReplyMessage implements Message
     }
 
     @Override
-    public void toStream(DataOutput out) throws IOException
+    public void toStream(DataOutputStream out) throws IOException
     {
         /* Add the origin node to the stream */
         origin.toStream(out);

src/kademlia/message/SimpleMessage.java

@@ -6,7 +6,9 @@
 package kademlia.message;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 
 public class SimpleMessage implements Message
@@ -22,7 +24,7 @@ public class SimpleMessage implements Message
         this.content = message;
     }
 
-    public SimpleMessage(DataInput in)
+    public SimpleMessage(DataInputStream in)
     {
         System.out.println("Creating message from input stream.");
         this.fromStream(in);
@@ -35,7 +37,7 @@ public class SimpleMessage implements Message
     }
 
     @Override
-    public void toStream(DataOutput out)
+    public void toStream(DataOutputStream out)
     {
         try
         {
@@ -49,7 +51,7 @@ public class SimpleMessage implements Message
     }
 
     @Override
-    public final void fromStream(DataInput in)
+    public final void fromStream(DataInputStream in)
     {
         try
         {

src/kademlia/message/StoreContentMessage.java

@@ -0,0 +1,74 @@
+package kademlia.message;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import kademlia.dht.KadContent;
+import kademlia.node.Node;
+import kademlia.serializer.JsonSerializer;
+
+/**
+ * A StoreContentMessage used to send a store message to a node
+ *
+ * @author Joshua Kissoon
+ * @since 20140225
+ */
+public class StoreContentMessage implements Message
+{
+
+    public static final byte CODE = 0x55;
+
+    private KadContent content;
+    private Node origin;
+
+    /**
+     * @param origin  Where the message came from
+     * @param content The content to be stored
+     *
+     */
+    public StoreContentMessage(Node origin, KadContent content)
+    {
+        this.content = content;
+        this.origin = origin;
+    }
+
+    public StoreContentMessage(DataInputStream in) throws IOException
+    {
+        this.fromStream(in);
+    }
+
+    @Override
+    public void toStream(DataOutputStream out) throws IOException
+    {
+        this.origin.toStream(out);
+
+        /* Serialize the KadContent, then send it to the stream */
+        JsonSerializer serializer = new JsonSerializer();
+        serializer.write(content, out);
+    }
+
+    @Override
+    public void fromStream(DataInputStream in) throws IOException
+    {
+        this.origin = new Node(in);
+        try
+        {
+            this.content = new JsonSerializer().read(in);
+        }
+        catch (ClassNotFoundException e)
+        {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public byte code()
+    {
+        return CODE;
+    }
+
+    public String toString()
+    {
+        return "StoreMessage[origin=" + origin + ",content=" + content + "]";
+    }
+}

src/kademlia/message/StoreContentReceiver.java

@@ -0,0 +1,39 @@
+package kademlia.message;
+
+import kademlia.core.KadServer;
+import kademlia.node.Node;
+import kademlia.operation.Receiver;
+
+/**
+ * Receiver for incoming StoreContentMessage
+ *
+ * @author Joshua Kissoon
+ * @since 20140225
+ */
+public class StoreContentReceiver implements Receiver
+{
+
+    private final KadServer server;
+    private final Node localNode;
+
+    public StoreContentReceiver(KadServer server, Node localNode)
+    {
+        this.server = server;
+        this.localNode = localNode;
+    }
+
+    @Override
+    public void receive(Message incoming, int comm)
+    {
+        /* @todo - Insert the message sender into this node's routing table */
+        StoreContentMessage msg = (StoreContentMessage) incoming;
+        System.out.println(this.localNode + " - Received a store content message");
+        System.out.println(msg);
+    }
+
+    @Override
+    public void timeout(int comm)
+    {
+        /* @todo Do something if the request times out */
+    }
+}

src/kademlia/message/Streamable.java

@@ -1,7 +1,7 @@
 package kademlia.message;
 
-import java.io.DataInput;
-import java.io.DataOutput;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
 import java.io.IOException;
 
 /**
@@ -16,17 +16,27 @@ import java.io.IOException;
  * that classes implementing Streamble also provide a constructor of the form:
  * <p>
  * <code>Streamable(DataInput in) throws IOException;</code>
- **/
-public interface Streamable {
+ * */
+public interface Streamable
+{
+
     /**
      * Writes the internal state of the Streamable object to the output stream
      * in a format that can later be read by the same Streamble class using
      * the {@link #fromStream} method.
-     **/
-    public void toStream(DataOutput out) throws IOException;
+     *
+     * @param out
+     *
+     * @throws java.io.IOException
+     */
+    public void toStream(DataOutputStream out) throws IOException;
 
     /**
      * Reads the internal state of the Streamable object from the input stream.
-     **/
-    public void fromStream(DataInput out) throws IOException;
+     *
+     * @param out
+     *
+     * @throws java.io.IOException
+     */
+    public void fromStream(DataInputStream out) throws IOException;
 }

src/kademlia/node/Node.java

@@ -1,7 +1,9 @@
 package kademlia.node;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -45,7 +47,7 @@ public class Node implements Streamable
      *
      * @throws IOException
      */
-    public Node(DataInput in) throws IOException
+    public Node(DataInputStream in) throws IOException
     {
         this.fromStream(in);
     }
@@ -79,7 +81,7 @@ public class Node implements Streamable
     }
 
     @Override
-    public void toStream(DataOutput out) throws IOException
+    public void toStream(DataOutputStream out) throws IOException
     {
         /* Add the NodeId to the stream */
         this.nodeId.toStream(out);
@@ -97,7 +99,7 @@ public class Node implements Streamable
     }
 
     @Override
-    public final void fromStream(DataInput in) throws IOException
+    public final void fromStream(DataInputStream in) throws IOException
     {
         /* Load the NodeId */
         this.nodeId = new NodeId(in);

src/kademlia/node/NodeId.java

@@ -6,7 +6,9 @@
 package kademlia.node;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.math.BigInteger;
 import java.util.Arrays;
@@ -60,7 +62,7 @@ public class NodeId implements Streamable
      *
      * @throws IOException
      */
-    public NodeId(DataInput in) throws IOException
+    public NodeId(DataInputStream in) throws IOException
     {
         this.fromStream(in);
     }
@@ -186,14 +188,14 @@ public class NodeId implements Streamable
     }
 
     @Override
-    public void toStream(DataOutput out) throws IOException
+    public void toStream(DataOutputStream out) throws IOException
     {
         /* Add the NodeId to the stream */
         out.write(this.getBytes());
     }
 
     @Override
-    public void fromStream(DataInput in) throws IOException
+    public void fromStream(DataInputStream in) throws IOException
     {
         byte[] input = new byte[ID_LENGTH / 8];
         in.readFully(input);

src/kademlia/operation/StoreOperation.java

@@ -4,6 +4,8 @@ import java.io.IOException;
 import java.util.ArrayList;
 import kademlia.core.KadServer;
 import kademlia.dht.KadContent;
+import kademlia.message.Message;
+import kademlia.message.StoreContentMessage;
 import kademlia.node.Node;
 
 /**
@@ -36,11 +38,25 @@ public class StoreOperation implements Operation
     {
         /* Get the nodes on which we need to store the content */
         ArrayList<Node> nodes = new NodeLookupOperation(this.server, this.localNode, this.content.getKey()).execute();
-        
-                
-        
         System.out.println("Nodes to put content on: " + nodes);
 
+        /* Create the message */
+        Message msg = new StoreContentMessage(this.localNode, this.content);
+
+        /*Store the message on all of the K-Nodes*/
+        for (Node n : nodes)
+        {
+            if (n.equals(this.localNode))
+            {
+                /* Store the content locally */
+            }
+            else
+            {
+                this.server.sendMessage(n, msg, null);
+            }
+        }
+
+
         /* Return how many nodes the content was stored on */
         return nodes.size();
     }

src/kademlia/serializer/JsonSerializer.java

@@ -4,6 +4,7 @@ import com.google.gson.Gson;
 import com.google.gson.stream.JsonReader;
 import com.google.gson.stream.JsonWriter;
 import java.io.DataInputStream;
+import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -29,10 +30,9 @@ public class JsonSerializer implements KadContentSerializer
     }
 
     @Override
-    public void write(KadContent content, OutputStream out) throws IOException
+    public void write(KadContent content, DataOutputStream out) throws IOException
     {
-        try (DataOutputStream dout = new DataOutputStream(out);
-                JsonWriter writer = new JsonWriter(new OutputStreamWriter(out)))
+        try (JsonWriter writer = new JsonWriter(new OutputStreamWriter(out)))
         {
             writer.beginArray();
 
@@ -48,7 +48,7 @@ public class JsonSerializer implements KadContentSerializer
     }
 
     @Override
-    public KadContent read(InputStream in) throws IOException, ClassNotFoundException
+    public KadContent read(DataInputStream in) throws IOException, ClassNotFoundException
     {
         try (DataInputStream din = new DataInputStream(in);
                 JsonReader reader = new JsonReader(new InputStreamReader(in)))

src/kademlia/serializer/KadContentSerializer.java

@@ -1,5 +1,7 @@
 package kademlia.serializer;
 
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -24,7 +26,7 @@ public interface KadContentSerializer
      *
      * @throws java.io.IOException
      */
-    public void write(KadContent content, OutputStream out) throws IOException;
+    public void write(KadContent content, DataOutputStream out) throws IOException;
 
     /**
      * Read a KadContent from a DataInput Stream
@@ -36,5 +38,5 @@ public interface KadContentSerializer
      * @throws java.io.IOException
      * @throws java.lang.ClassNotFoundException
      */
-    public KadContent read(InputStream in) throws IOException, ClassNotFoundException;
+    public KadContent read(DataInputStream in) throws IOException, ClassNotFoundException;
 }

src/kademlia/tests/DHTContentImpl.java

@@ -70,4 +70,9 @@ public class DHTContentImpl implements KadContent
     {
         return this.createTs;
     }
+
+    public String toString()
+    {
+        return "DHTContentImpl[data=" + this.data + "]";
+    }
 }