Content Store, Retrieve and DHT File and Table management working well

Got a few Todos now to finish off

src/kademlia/core/Configuration.java

@@ -21,7 +21,7 @@ public class Configuration
     /**
      * Maximum number of milliseconds for performing an operation.
      * */
-    public static long OPERATION_TIMEOUT = 10000;
+    public static long OPERATION_TIMEOUT = 3000;
 
     /**
      * Maximum number of concurrent messages in transit.

src/kademlia/core/GetParameter.java

@@ -41,13 +41,29 @@ public class GetParameter
         return this.key;
     }
 
+    public void setOwnerId(String ownerId)
+    {
+        this.ownerId = ownerId;
+    }
+
     public String getOwnerId()
     {
         return this.ownerId;
     }
 
+    public void setType(String type)
+    {
+        this.type = type;
+    }
+
     public String getType()
     {
         return this.type;
     }
+
+    @Override
+    public String toString()
+    {
+        return "GetParameter - [Key: " + key + "][Owner: " + this.ownerId + "][Type: " + this.type + "]";
+    }
 }

src/kademlia/core/KadServer.java

@@ -170,12 +170,13 @@ public class KadServer
 
                     Message msg = messageFactory.createMessage(messCode, din);
                     din.close();
-                    System.out.println(this.localNode.getNodeId() + " Message Received: " + msg);
+                    System.out.println(this.localNode.getNodeId() + " Message Received: [Comm: " + comm + "] " + msg);
 
                     /* Get a receiver for this message */
                     Receiver receiver;
                     if (this.receivers.containsKey(comm))
                     {
+                        System.out.println("Receiver found");
                         /* If there is a reciever in the receivers to handle this */
                         synchronized (this)
                         {

src/kademlia/core/Kademlia.java

@@ -2,6 +2,7 @@ package kademlia.core;
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Timer;
@@ -153,18 +154,24 @@ public class Kademlia
      */
     public List<KadContent> get(GetParameter param, int numResultsReq) throws NoSuchElementException, IOException
     {
+        List contentFound;
         if (this.dht.contains(param))
         {
             /* If the content exist in our own DHT, then return it. */
-            return this.dht.get(param);
+            System.out.println("Found content locally");
+            contentFound = new ArrayList<>();
+            contentFound.add(this.dht.get(param));
         }
         else
         {
             /* Seems like it doesn't exist in our DHT, get it from other Nodes */
+            System.out.println("Looking for content on foreign nodes");
             ContentLookupOperation clo = new ContentLookupOperation(server, localNode, param, numResultsReq);
             clo.execute();
-            return clo.getContentFound();
+            contentFound = clo.getContentFound();
         }
+
+        return contentFound;
     }
 
     /**

src/kademlia/dht/DHT.java

@@ -7,8 +7,6 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.NoSuchElementException;
 import kademlia.core.Configuration;
 import kademlia.core.GetParameter;
@@ -25,10 +23,12 @@ public class DHT
 {
 
     private final StorageEntryManager entriesManager;
+    private final JsonSerializer<KadContent> contentSerializer;
 
     
     {
         entriesManager = new StorageEntryManager();
+        contentSerializer = new JsonSerializer<>();
     }
 
     /**
@@ -46,7 +46,7 @@ public class DHT
         /* Now we store the content locally in a file */
         String contentStorageFolder = this.getContentStorageFolderName(content.getKey());
         DataOutputStream dout = new DataOutputStream(new FileOutputStream(contentStorageFolder + File.separator + content.hashCode() + ".kct"));
-        new JsonSerializer().write(content, dout);
+        contentSerializer.write(content, dout);
     }
 
     /**
@@ -61,7 +61,7 @@ public class DHT
     {
         String folder = this.getContentStorageFolderName(key);
         DataInputStream in = new DataInputStream(new FileInputStream(folder + File.separator + hashCode + ".kct"));
-        return new JsonSerializer().read(in);
+        return contentSerializer.read(in);
     }
 
     /**
@@ -86,17 +86,14 @@ public class DHT
      *
      * @throws java.io.IOException
      */
-    public List<KadContent> get(GetParameter param) throws NoSuchElementException, IOException
+    public KadContent get(GetParameter param) throws NoSuchElementException, IOException
     {
-        /* Load all KadContent for the entries if any exist */
+        /* Load a KadContent if any exist for the given criteria */
-        List<KadContent> values = new ArrayList<>();
-
         try
         {
-            for (StorageEntry e : this.entriesManager.get(param))
+            StorageEntry e = this.entriesManager.get(param);
-            {
+            return this.retrieve(e.getKey(), e.getContentHash());
-                values.add(this.retrieve(e.getKey(), e.getContentHash()));
+
-            }
         }
         catch (FileNotFoundException e)
         {
@@ -107,12 +104,8 @@ public class DHT
             System.err.println("The class for some content was not found.");
         }
 
-        if (values.isEmpty())
+        /* If we got here, means we got no entries */
-        {
+        throw new NoSuchElementException();
-            throw new NoSuchElementException();
-        }
-
-        return values;
     }
 
     /**

src/kademlia/dht/StorageEntryManager.java

@@ -38,7 +38,6 @@ public class StorageEntryManager
         {
             this.entries.put(entry.getKey(), new ArrayList<StorageEntry>());
         }
-
         this.entries.get(entry.getKey()).add(entry);
     }
 
@@ -77,30 +76,22 @@ public class StorageEntryManager
      *
      * @return List of content for the specific search parameters
      */
-    public List<StorageEntry> get(GetParameter param) throws NoSuchElementException
+    public StorageEntry get(GetParameter param) throws NoSuchElementException
     {
         if (this.entries.containsKey(param.getKey()))
         {
             /* Content with this key exist, check if any match the rest of the search criteria */
-            List<StorageEntry> results = new ArrayList<>();
-
             for (StorageEntry e : this.entries.get(param.getKey()))
             {
                 /* If any entry satisfies the given parameters, return true */
                 if (e.satisfiesParameters(param))
                 {
-                    results.add(e);
+                    return e;
                 }
             }
 
-            if (results.size() > 0)
+            /* If we got here, means we didn't find any entry */
-            {
+            throw new NoSuchElementException();
-                return results;
-            }
-            else
-            {
-                throw new NoSuchElementException();
-            }
         }
         else
         {

src/kademlia/message/AcknowledgeMessage.java

@@ -5,9 +5,7 @@
  */
 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;

src/kademlia/message/ConnectMessage.java

@@ -5,9 +5,7 @@
  */
 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;

src/kademlia/message/ContentLookupMessage.java

@@ -1,10 +1,13 @@
 package kademlia.message;
 
+import com.google.gson.JsonSerializationContext;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import kademlia.core.GetParameter;
+import kademlia.dht.KadContent;
 import kademlia.node.Node;
+import kademlia.serializer.JsonSerializer;
 
 /**
  * Messages used to send to another node requesting content
@@ -15,7 +18,7 @@ import kademlia.node.Node;
 public class ContentLookupMessage implements Message
 {
 
-    public static final byte CODE = 0x47;
+    public static final byte CODE = 0x03;
 
     private Node origin;
     private GetParameter params;
@@ -49,12 +52,25 @@ public class ContentLookupMessage implements Message
     public void toStream(DataOutputStream out) throws IOException
     {
         this.origin.toStream(out);
+
+        /* Write the params to the stream */
+        new JsonSerializer<GetParameter>().write(this.params, out);
     }
 
     @Override
     public final void fromStream(DataInputStream in) throws IOException
     {
         this.origin = new Node(in);
+
+        /* Read the params from the stream */
+        try
+        {
+            this.params = new JsonSerializer<GetParameter>().read(in);
+        }
+        catch (ClassNotFoundException e)
+        {
+            e.printStackTrace();
+        }
     }
 
     @Override

src/kademlia/message/ContentLookupReceiver.java

@@ -31,12 +31,14 @@ public class ContentLookupReceiver implements Receiver
     public void receive(Message incoming, int comm) throws IOException
     {
         ContentLookupMessage msg = (ContentLookupMessage) incoming;
+        this.localNode.getRoutingTable().insert(msg.getOrigin());
 
         /* Check if we can have this data */
         if (this.dht.contains(msg.getParameters()))
         {
             /* Return a ContentMessage with the required data */
             ContentMessage cMsg = new ContentMessage(localNode, this.dht.get(msg.getParameters()));
+            server.reply(msg.getOrigin(), cMsg, comm);
         }
         else
         {

src/kademlia/message/ContentMessage.java

@@ -16,7 +16,7 @@ import kademlia.serializer.JsonSerializer;
 public class ContentMessage implements Message
 {
 
-    public static final byte CODE = 0x56;
+    public static final byte CODE = 0x04;
 
     private KadContent content;
     private Node origin;
@@ -43,17 +43,17 @@ public class ContentMessage implements Message
         this.origin.toStream(out);
 
         /* Serialize the KadContent, then send it to the stream */
-        JsonSerializer serializer = new JsonSerializer();
+        new JsonSerializer<KadContent>().write(content, out);
-        serializer.write(content, out);
     }
 
     @Override
     public final void fromStream(DataInputStream in) throws IOException
     {
         this.origin = new Node(in);
+        
         try
         {
-            this.content = new JsonSerializer().read(in);
+            this.content = new JsonSerializer<KadContent>().read(in);
         }
         catch (ClassNotFoundException e)
         {
@@ -80,6 +80,6 @@ public class ContentMessage implements Message
     @Override
     public String toString()
     {
-        return "StoreMessage[origin=" + origin + ",content=" + content + "]";
+        return "ContentMessage[origin=" + origin + ",content=" + content + "]";
     }
 }

src/kademlia/message/MessageFactory.java

@@ -29,20 +29,24 @@ public class MessageFactory
     {
         switch (code)
         {
-            case SimpleMessage.CODE:
-                return new SimpleMessage(in);
-            case ConnectMessage.CODE:
-                return new ConnectMessage(in);
             case AcknowledgeMessage.CODE:
                 return new AcknowledgeMessage(in);
-            case NodeReplyMessage.CODE:
+            case ConnectMessage.CODE:
-                return new NodeReplyMessage(in);
+                return new ConnectMessage(in);
+            case ContentMessage.CODE:
+                return new ContentMessage(in);
+            case ContentLookupMessage.CODE:
+                return new ContentLookupMessage(in);
             case NodeLookupMessage.CODE:
                 return new NodeLookupMessage(in);
+            case NodeReplyMessage.CODE:
+                return new NodeReplyMessage(in);
+            case SimpleMessage.CODE:
+                return new SimpleMessage(in);
             case StoreContentMessage.CODE:
                 return new StoreContentMessage(in);
             default:
-                System.out.println("No Message handler found for message. Code: " + code);
+                System.out.println(this.localNode + " - No Message handler found for message. Code: " + code);
                 return new SimpleMessage(in);
 
         }
@@ -52,17 +56,19 @@ public class MessageFactory
     {
         switch (code)
         {
-            default:
-            case SimpleMessage.CODE:
-                return new SimpleReceiver();
             case ConnectMessage.CODE:
                 return new ConnectReceiver(server, this.localNode);
-            case NodeLookupMessage.CODE:
-                return new NodeLookupReceiver(server, this.localNode);
-            case StoreContentMessage.CODE:
-                return new StoreContentReceiver(server, this.localNode, this.dht);
             case ContentLookupMessage.CODE:
                 return new ContentLookupReceiver(server, localNode, dht);
+            case NodeLookupMessage.CODE:
+                return new NodeLookupReceiver(server, this.localNode);
+            case SimpleMessage.CODE:
+                return new SimpleReceiver();
+            case StoreContentMessage.CODE:
+                return new StoreContentReceiver(server, this.localNode, this.dht);
+            default:
+                System.out.println("No reveiver found for message. Code: " + code);
+                return new SimpleReceiver();
         }
     }
 }

src/kademlia/message/NodeLookupMessage.java

@@ -5,9 +5,7 @@
  */
 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;
@@ -19,7 +17,7 @@ public class NodeLookupMessage implements Message
     private Node origin;
     private NodeId lookupId;
 
-    public static final byte CODE = 0x03;
+    public static final byte CODE = 0x05;
 
     /**
      * A new NodeLookupMessage to find nodes

src/kademlia/message/NodeReplyMessage.java

@@ -16,7 +16,7 @@ public class NodeReplyMessage implements Message
 {
 
     private Node origin;
-    public static final byte CODE = 0x04;
+    public static final byte CODE = 0x06;
     private List<Node> nodes;
 
     public NodeReplyMessage(Node origin, List<Node> nodes)

src/kademlia/message/SimpleMessage.java

@@ -5,9 +5,7 @@
  */
 package kademlia.message;
 
-import java.io.DataInput;
 import java.io.DataInputStream;
-import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
 
@@ -15,7 +13,7 @@ public class SimpleMessage implements Message
 {
 
     /* Message constants */
-    public static final byte CODE = 0x05;
+    public static final byte CODE = 0x07;
 
     private String content;
 

src/kademlia/message/StoreContentMessage.java

@@ -16,7 +16,7 @@ import kademlia.serializer.JsonSerializer;
 public class StoreContentMessage implements Message
 {
 
-    public static final byte CODE = 0x55;
+    public static final byte CODE = 0x08;
 
     private KadContent content;
     private Node origin;
@@ -43,8 +43,7 @@ public class StoreContentMessage implements Message
         this.origin.toStream(out);
 
         /* Serialize the KadContent, then send it to the stream */
-        JsonSerializer serializer = new JsonSerializer();
+        new JsonSerializer<KadContent>().write(content, out);
-        serializer.write(content, out);
     }
 
     @Override
@@ -53,7 +52,7 @@ public class StoreContentMessage implements Message
         this.origin = new Node(in);
         try
         {
-            this.content = new JsonSerializer().read(in);
+            this.content = new JsonSerializer<KadContent>().read(in);
         }
         catch (ClassNotFoundException e)
         {
@@ -80,6 +79,6 @@ public class StoreContentMessage implements Message
     @Override
     public String toString()
     {
-        return "StoreMessage[origin=" + origin + ",content=" + content + "]";
+        return "StoreContentMessage[origin=" + origin + ",content=" + content + "]";
     }
 }

src/kademlia/operation/ContentLookupOperation.java

@@ -105,12 +105,21 @@ public class ContentLookupOperation implements Operation, Receiver
             if (!this.askNodesorFinish())
             {
                 /* If we haven't finished as yet, wait a while */
+                /**
+                 * @todo Get rid of this wait here! 
+                 * We should run this until there are no nodes left to ask from the K closest nodes
+                 * and only pause for short intervals in between
+                 * 
+                 * @todo Do the same for the NodeLookupOperation
+                 */
                 wait(Configuration.OPERATION_TIMEOUT);
 
                 /* If we still haven't received any responses by then, do a routing timeout */
                 if (error)
                 {
-                    throw new RoutingException("Lookup Timeout.");
+                    /* Lets not throw any exception */
+                    
+                    //throw new RoutingException("Content Lookup Operation Timeout.");
                 }
             }
         }
@@ -132,16 +141,9 @@ public class ContentLookupOperation implements Operation, Receiver
             /* If this node is not in the list, add the node */
             if (!nodes.containsKey(o))
             {
-                System.out.println("Adding node " + o.getNodeId());
                 nodes.put(o, UNASKED);
             }
         }
-
-        System.out.println(this.localNode.getNodeId() + " Nodes List: ");
-        for (Node o : this.nodes.keySet())
-        {
-            System.out.println(o.getNodeId() + " hash: " + o.hashCode());
-        }
     }
 
     /**
@@ -230,6 +232,11 @@ public class ContentLookupOperation implements Operation, Receiver
     @Override
     public synchronized void receive(Message incoming, int comm) throws IOException, RoutingException
     {
+        if (!this.isRunning)
+        {
+            return;
+        }
+        
         if (incoming instanceof ContentMessage)
         {
             /* The reply received is a content message with the required content, take it in */
@@ -240,6 +247,7 @@ public class ContentLookupOperation implements Operation, Receiver
 
             /* Get the Content and check if it satisfies the required parameters */
             KadContent content = msg.getContent();
+            System.out.println("Content Received: " + content);
 
             /*@todo Check if the content matches the given criteria */
             this.contentFound.add(content);
@@ -247,6 +255,8 @@ public class ContentLookupOperation implements Operation, Receiver
             if (this.contentFound.size() == this.numResultsReq)
             {
                 /* We've got all the content required, let's stop the loopup operation */
+                System.out.println("We good");
+                this.isRunning = false;
             }
         }
         else

src/kademlia/operation/NodeLookupOperation.java

@@ -128,16 +128,16 @@ public class NodeLookupOperation implements Operation, Receiver
             /* If this node is not in the list, add the node */
             if (!nodes.containsKey(o))
             {
-                System.out.println("Adding node " + o.getNodeId());
+                //System.out.println("Adding node " + o.getNodeId());
                 nodes.put(o, UNASKED);
             }
         }
 
-        System.out.println(this.localNode.getNodeId() + " Nodes List: ");
+//        System.out.println(this.localNode.getNodeId() + " Nodes List: ");
-        for (Node o : this.nodes.keySet())
+//        for (Node o : this.nodes.keySet())
-        {
+//        {
-            System.out.println(o.getNodeId() + " hash: " + o.hashCode());
+//            System.out.println(o.getNodeId() + " hash: " + o.hashCode());
-        }
+//        }
     }
 
     /**
@@ -161,10 +161,10 @@ public class NodeLookupOperation implements Operation, Receiver
 
         /* Get unqueried nodes among the K closest seen that have not FAILED */
         List<Node> unasked = this.closestNodesNotFailed(UNASKED);
-        for (Node nn : unasked)
+//        for (Node nn : unasked)
-        {
+//        {
-            System.out.println(nn.getNodeId());
+//            System.out.println(nn.getNodeId());
-        }
+//        }
 
         if (unasked.isEmpty() && this.messagesTransiting.isEmpty())
         {

src/kademlia/serializer/JsonSerializer.java

@@ -4,22 +4,21 @@ 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;
 import java.io.InputStreamReader;
-import java.io.OutputStream;
 import java.io.OutputStreamWriter;
-import kademlia.dht.KadContent;
 
 /**
  * A KadContentSerializer that serializes content to JSON format
  *
+ * @param <T> The type of content to serialize
+ *
  * @author Joshua Kissoon
+ *
  * @since 20140225
  */
-public class JsonSerializer implements KadContentSerializer
+public class JsonSerializer<T> implements KadContentSerializer<T>
 {
 
     private final Gson gson;
@@ -30,17 +29,17 @@ public class JsonSerializer implements KadContentSerializer
     }
 
     @Override
-    public void write(KadContent content, DataOutputStream out) throws IOException
+    public void write(T data, DataOutputStream out) throws IOException
     {
         try (JsonWriter writer = new JsonWriter(new OutputStreamWriter(out)))
         {
             writer.beginArray();
 
             /* Store the content type */
-            gson.toJson(content.getClass().getName(), String.class, writer);
+            gson.toJson(data.getClass().getName(), String.class, writer);
 
             /* Now Store the content */
-            gson.toJson(content, content.getClass(), writer);
+            gson.toJson(data, data.getClass(), writer);
 
             writer.endArray();
         }
@@ -48,7 +47,7 @@ public class JsonSerializer implements KadContentSerializer
     }
 
     @Override
-    public KadContent read(DataInputStream in) throws IOException, ClassNotFoundException
+    public T read(DataInputStream in) throws IOException, ClassNotFoundException
     {
         try (DataInputStream din = new DataInputStream(in);
                 JsonReader reader = new JsonReader(new InputStreamReader(in)))

src/kademlia/serializer/KadContentSerializer.java

@@ -3,9 +3,6 @@ package kademlia.serializer;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import kademlia.dht.KadContent;
 
 /**
  * A Serializer is used to transform data to and from a specified form.
@@ -13,30 +10,32 @@ import kademlia.dht.KadContent;
  * Here we define the structure of any Serializer used in Kademlia
  *
  * @author Joshua Kissoon
+ * @param <T> The type of content being serialized
+ *
  * @since 20140225
  */
-public interface KadContentSerializer
+public interface KadContentSerializer<T>
 {
 
     /**
      * Write a KadContent to a DataOutput stream
      *
-     * @param content The content to write
+     * @param data The data to write
-     * @param out     The output Stream to write to
+     * @param out  The output Stream to write to
      *
      * @throws java.io.IOException
      */
-    public void write(KadContent content, DataOutputStream out) throws IOException;
+    public void write(T data, DataOutputStream out) throws IOException;
 
     /**
-     * Read a KadContent from a DataInput Stream
+     * Read data of type T from a DataInput Stream
      *
      * @param in The InputStream to read the data from
      *
-     * @return KadContent
+     * @return T Data of type T
      *
      * @throws java.io.IOException
      * @throws java.lang.ClassNotFoundException
      */
-    public KadContent read(DataInputStream in) throws IOException, ClassNotFoundException;
+    public T read(DataInputStream in) throws IOException, ClassNotFoundException;
 }

src/kademlia/tests/ContentSendingTest.java

@@ -1,7 +1,10 @@
 package kademlia.tests;
 
 import java.io.IOException;
+import java.util.List;
+import kademlia.core.GetParameter;
 import kademlia.core.Kademlia;
+import kademlia.dht.KadContent;
 import kademlia.node.NodeId;
 
 /**
@@ -30,6 +33,20 @@ public class ContentSendingTest
             DHTContentImpl c = new DHTContentImpl(kad2.getOwnerId(), "Some Data");
             kad2.put(c);
             
+            /**
+             * Lets retrieve the content
+             */
+            System.out.println("Retrieving Content");
+            GetParameter gp = new GetParameter(c.getKey());
+            gp.setType(DHTContentImpl.TYPE);
+            gp.setOwnerId(c.getOwnerId());
+            System.out.println("Get Parameter: " + gp);
+            List<KadContent> conte = kad2.get(gp, 1);
+            for (KadContent cc : conte)
+            {
+                System.out.println("Content Found: " + cc);
+            }
+
         }
         catch (IOException e)
         {

src/kademlia/tests/DHTContentImpl.java

@@ -17,7 +17,7 @@ public class DHTContentImpl implements KadContent
     private final String ownerId;
     private final long createTs;
 
-    private static final String type = "DHTContentImpl";
+    public static final String TYPE = "DHTContentImpl";
 
     
     {
@@ -56,7 +56,7 @@ public class DHTContentImpl implements KadContent
     @Override
     public String getType()
     {
-        return type;
+        return TYPE;
     }
 
     @Override