public class JSocialKademliaDHT implements SocialKademliaDHT
{

    private transient StoredContentManager contentManager;
    private transient KadSerializer<JSocialKademliaStorageEntry> serializer = null;
    private transient KadConfiguration config;

    private final String ownerId;

    public JSocialKademliaDHT(String ownerId, KadConfiguration config)
    {
        this.ownerId = ownerId;
        this.config = config;
        this.initialize();
    }

    /**
     * Initialize this JSocialKademliaDHT to it's default state
     */
    @Override
    public final void initialize()
    {
        contentManager = new StoredContentManager();
    }

    /**
     * Set a new configuration. Mainly used when we restore the JSocialKademliaDHT state from a file
     *
     * @param con The new configuration file
     */
    @Override
    public void setConfiguration(KadConfiguration con)
    {
        this.config = con;
    }

    /**
     * Creates a new Serializer or returns an existing serializer
     *
     * @return The new ContentSerializer
     */
    @Override
    public KadSerializer<JSocialKademliaStorageEntry> getSerializer()
    {
        if (null == serializer)
        {
            serializer = new JsonSerializer<>();
        }

        return serializer;
    }

    /**
     * Handle storing content locally
     *
     * @param content The JSocialKademliaDHT content to store
     *
     * @return boolean true if we stored the content, false if the content already exists and is up to date
     *
     * @throws java.io.IOException
     */
    @Override
    public boolean store(JSocialKademliaStorageEntry content) throws IOException
    {
        boolean cached = content.getContentMetadata().isCached();   // Should we cache this content
        boolean isKNode = content.getContentMetadata().isKNode();   // Is this node one of the k-node

        /* Lets check if we have this content and it's the updated version */
        if (this.contentManager.contains(content.getContentMetadata()))
        {
            SocialKademliaStorageEntryMetadata current = this.contentManager.get(content.getContentMetadata());

            /* update the last republished time */
            current.updateLastRepublished();

            /* We have the current content, no need to update it! */
            if (current.getLastUpdatedTimestamp() >= content.getContentMetadata().getLastUpdatedTimestamp())
            {
                /* Cache it if required */
                if (cached)
                {
                    current.setCached();
                }

                /* Set this is a K-Node if required */
                if (isKNode)
                {
                    current.setKNode();
                }
                return false;
            }
            else
            {
                /* We got here means we don't have the current content, lets update it */
                /* If the current version is a cached version, remember to cache it back if we need to do an update */
                if (current.isCached())
                {
                    cached = true;
                }

                /* If this is a k-node for the current version, remember to set that back after an update */
                if (current.isKNode())
                {
                    isKNode = true;
                }

                /* Since we don't have the latest version, lets delete it so the new version will be added below */
                try
                {
                    this.absoluteRemove(current);
                }
                catch (ContentNotFoundException ex)
                {
                    /* This won't ever happen at this point since we only get here if the content is found, lets ignore it  */
                }
            }
        }

        /* We got here means we need to add the content or re-add it to update it */
        try
        {
            /* Store the content to a file and then keep track of this content in the entries manager */
            content.getContentMetadata().updateLastRepublished();
            content.getContentMetadata().setCached(cached);
            content.getContentMetadata().setKNode(isKNode);
            this.contentManager.put(content.getContentMetadata());
            this.putContentToFile(content, content.getContentMetadata());
            return true;
        }
        catch (ContentExistException e)
        {
            /**
             * Content already exist on the DHT
             * This won't happen because above takes care of removing the content if it's older and needs to be updated,
             * or returning if we already have the current content version.
             */
            return false;
        }
    }

    @Override
    public boolean store(KadContent content) throws IOException
    {
        return this.store(new JSocialKademliaStorageEntry(content));
    }

    /**
     * Handle storing content locally to keep the content cached.
     *
     * We set that this content is a cached entry and that this node is not one of the k-nodes.
     *
     * @param content The JSocialKademliaDHT content to store
     *
     * @return boolean true if we stored the content, false if the content already exists and is up to date
     *
     * @throws java.io.IOException
     */
    @Override
    public boolean cache(JSocialKademliaStorageEntry content) throws IOException
    {
        content.getContentMetadata().setCached();
        content.getContentMetadata().setKNode(false);
        return this.store(content);
    }

    @Override
    public boolean cache(KadContent content) throws IOException
    {
        return this.cache(new JSocialKademliaStorageEntry(content));
    }

    /**
     * Write the given storage entry to it's file
     */
    private void putContentToFile(JSocialKademliaStorageEntry content, SocialKademliaStorageEntryMetadata entryMD) throws IOException
    {
        String contentStorageFolder = this.getContentStorageFolderName(content.getContentMetadata().getKey());

        try (FileOutputStream fout = new FileOutputStream(contentStorageFolder + File.separator + entryMD.hashCode() + ".kct");
                DataOutputStream dout = new DataOutputStream(fout))
        {
            this.getSerializer().write(content, dout);
        }
    }

    /**
     * Update a content; the operation is only done iff we already have a copy of the content here
     *
     * @param newContent The content to update.
     *
     * @throws java.io.IOException
     */
    @Override
    public void update(JSocialKademliaStorageEntry newContent) throws IOException
    {
        if (this.contentManager.contains(newContent.getContentMetadata()))
        {
            this.store(newContent);
        }
        else
        {
            throw new NoSuchElementException("This content is not on the DHT currently, cannot update it.");
        }
    }

    /**
     * Retrieves a Content from local storage
     *
     * @param key      The Key of the content to retrieve
     * @param hashCode The hash code of the content to retrieve
     *
     * @return A KadContent object
     *
     * @throws java.io.IOException
     */
    @Override
    public JSocialKademliaStorageEntry retrieve(KademliaId key, int hashCode) throws FileNotFoundException, IOException, ClassNotFoundException
    {
        String folder = this.getContentStorageFolderName(key);
        DataInputStream din = new DataInputStream(new FileInputStream(folder + File.separator + hashCode + ".kct"));
        return this.getSerializer().read(din);
    }

    /**
     * Check if any content for the given criteria exists in this JSocialKademliaDHT
     *
     * @param param The content search criteria
     *
     * @return boolean Whether any content exist that satisfy the criteria
     */
    @Override
    public boolean contains(GetParameter param)
    {
        return this.contentManager.contains(param);
    }

    /**
     * Retrieve and create a KadContent object given the JSocialKademliaStorageEntry object
     *
     * @param entry The JSocialKademliaStorageEntry used to retrieve this content
     *
     * @return KadContent The content object
     *
     * @throws java.io.IOException
     */
    @Override
    public JSocialKademliaStorageEntry get(SocialKademliaStorageEntryMetadata entry) throws IOException, NoSuchElementException
    {
        try
        {
            return this.retrieve(entry.getKey(), entry.hashCode());
        }
        catch (FileNotFoundException e)
        {
            System.err.println("Error while loading file for content. Message: " + e.getMessage());
        }
        catch (ClassNotFoundException e)
        {
            System.err.println("The class for some content was not found. Message: " + e.getMessage());
        }

        /* If we got here, means we got no entries */
        throw new NoSuchElementException();
    }

    /**
     * Get the JSocialKademliaStorageEntry for the content if any exist,
     * retrieve the KadContent from the storage system and return it
     *
     * @param param The parameters used to filter the content needed
     *
     * @return KadContent A KadContent found on the JSocialKademliaDHT satisfying the given criteria
     *
     * @throws java.io.IOException
     */
    @Override
    public JSocialKademliaStorageEntry get(GetParameter param) throws NoSuchElementException, IOException
    {
        /* Load a KadContent if any exist for the given criteria */
        try
        {
            SocialKademliaStorageEntryMetadata e = this.contentManager.get(param);
            return this.retrieve(e.getKey(), e.hashCode());
        }
        catch (FileNotFoundException e)
        {
            System.err.println("Error while loading file for content. Message: " + e.getMessage());
        }
        catch (ClassNotFoundException e)
        {
            System.err.println("The class for some content was not found. Message: " + e.getMessage());
        }

        /* If we got here, means we got no entries */
        throw new NoSuchElementException();
    }

    /**
     * Delete a content from local storage
     *
     * @param content The Content to Remove
     *
     *
     * @throws kademlia.exceptions.ContentNotFoundException
     */
    @Override
    public void remove(KadContent content) throws ContentNotFoundException
    {
        this.remove(new JSocialKademliaStorageEntryMetadata(content));
    }

    /**
     * Similar to the remove method, however, in this case, we remove the content even if it's cached
     */
    private void absoluteRemove(SocialKademliaStorageEntryMetadata entry) throws ContentNotFoundException
    {
        contentManager.remove(entry);

        String folder = this.getContentStorageFolderName(entry.getKey());
        File file = new File(folder + File.separator + entry.hashCode() + ".kct");
        if (file.exists())
        {
            file.delete();
        }
        else
        {
            throw new ContentNotFoundException();
        }
    }

    @Override
    public void remove(SocialKademliaStorageEntryMetadata entry) throws ContentNotFoundException
    {
        /* If it's cached data, we don't remove it, just set that we are no longer one of the k-closest */
        if (this.contentManager.get(entry).isCached())
        {
            this.contentManager.get(entry).setKNode(false);
            return;
        }

        this.absoluteRemove(entry);
    }

    /**
     * Get the name of the folder for which a content should be stored
     *
     * @param key The key of the content
     *
     * @return String The name of the folder
     */
    private String getContentStorageFolderName(KademliaId key)
    {
        /**
         * Each content is stored in a folder named after the first 2 characters of the NodeId
         *
         * The name of the file containing the content is the hash of this content
         */
        String folderName = key.hexRepresentation().substring(0, 2);
        File contentStorageFolder = new File(this.config.getNodeDataFolder(ownerId) + File.separator + folderName);

        /* Create the content folder if it doesn't exist */
        if (!contentStorageFolder.isDirectory())
        {
            contentStorageFolder.mkdir();
        }

        return contentStorageFolder.toString();
    }

    /**
     * @return A List of all StorageEntries for this node
     */
    @Override
    public List<SocialKademliaStorageEntryMetadata> getStorageEntries()
    {
        return contentManager.getAllEntries();
    }

    /**
     * @return A List of all StorageEntries of cached content for this node
     */
    public List<SocialKademliaStorageEntryMetadata> getCachedStorageEntries()
    {
        return contentManager.getAllCachedEntries();
    }

    /**
     * Used to add a list of storage entries for existing content to the JSocialKademliaDHT.
     * Mainly used when retrieving StorageEntries from a saved state file.
     *
     * @param ientries The entries to add
     */
    @Override
    public void putStorageEntries(List<SocialKademliaStorageEntryMetadata> ientries)
    {
        for (KademliaStorageEntryMetadata e : ientries)
        {
            SocialKademliaStorageEntryMetadata se = (SocialKademliaStorageEntryMetadata) e;
            try
            {
                this.contentManager.put(se);
            }
            catch (ContentExistException ex)
            {
                /* Entry already exist, no need to store it again */
            }
        }
    }

    @Override
    public synchronized String toString()
    {
        return this.contentManager.toString();
    }
} public class GetParameterFUC extends GetParameter
{

    private final long lastUpdatedTS;

    /**
     * Construct a GetParameter to search for data by NodeId and owner
     *
     * @param key
     * @param type
     * @param updatedTs The last updated Timestamp
     */
    public GetParameterFUC(KademliaId key, String type, long updatedTs)
    {
        super(key, type);
        this.lastUpdatedTS = updatedTs;
    }

    /**
     * Construct a GetParameter to search for data by NodeId, owner, type
     *
     * @param key
     * @param type
     * @param updatedTS
     * @param owner
     */
    public GetParameterFUC(KademliaId key, String type, String owner, long updatedTS)
    {
        super(key, type, owner);
        this.lastUpdatedTS = updatedTS;
    }

    /**
     * Construct our get parameter from a Content
     *
     * @param c
     */
    public GetParameterFUC(KadContent c)
    {
        super(c);
        this.lastUpdatedTS = c.getLastUpdatedTimestamp();
    }

    /**
     * Construct our get parameter from a StorageEntryMeta data
     *
     * @param md
     */
    public GetParameterFUC(SocialKademliaStorageEntryMetadata md)
    {
        super(md);
        this.lastUpdatedTS = md.getLastUpdatedTimestamp();
    }

    public long getLastUpdatedTimestamp()
    {
        return this.lastUpdatedTS;
    }

    @Override
    public String toString()
    {
        return "GetParameterFUC - [UpdatedTs: " + lastUpdatedTS + "] " + super.toString();
    }
} public class JSocialKademliaStorageEntry implements SocialKademliaStorageEntry
{

    private String content;
    private final JSocialKademliaStorageEntryMetadata metadata;
    private boolean isCompressed;       // Is this entry in compressed form

    
    {
        this.isCompressed = false;
    }

    public JSocialKademliaStorageEntry(final KadContent content)
    {
        this(content, new JSocialKademliaStorageEntryMetadata(content));
    }

    public JSocialKademliaStorageEntry(final KadContent content, final JSocialKademliaStorageEntryMetadata metadata)
    {
        this.setContent(content.toSerializedForm());
        this.metadata = metadata;
    }

    @Override
    public final void setContent(final byte[] data)
    {
        this.content = new String(data);
    }

    @Override
    public final void setContent(String data)
    {
        this.content = data;
    }

    @Override
    public final byte[] getContent()
    {
        return this.content.getBytes();
    }

    @Override
    public final String getContentString()
    {
        return this.content;
    }

    @Override
    public final SocialKademliaStorageEntryMetadata getContentMetadata()
    {
        return this.metadata;
    }

    @Override
    public boolean isCompressed()
    {
        return this.isCompressed;
    }

    @Override
    public void setCompressed()
    {
        this.isCompressed = true;
    }

    @Override
    public void setDecompressed()
    {
        this.isCompressed = false;
    }

    @Override
    public String toString()
    {
        StringBuilder sb = new StringBuilder("[StorageEntry: ");

        sb.append("[Content: ");
        sb.append(this.getContent());
        sb.append("]");

        sb.append(this.getContentMetadata());

        sb.append("]");

        return sb.toString();
    }
} public class StorageEntryCompressor
{

    /**
     * Compress a given storage entry.
     *
     * We only compress the content part of the storage entry.
     *
     * @param entry The entry whose content is to be compressed
     *
     * @return The entry with compressed content
     *
     * @throws java.io.IOException
     */
    public static JSocialKademliaStorageEntry compress(final JSocialKademliaStorageEntry entry) throws IOException
    {
        if (entry.isCompressed())
        {
            return entry;
        }

        try
        {
            entry.setContent(StringCompressor.compress(entry.getContent()));
            entry.setCompressed();
        }
        catch (IOException ex)
        {
            System.err.println("Error whiles decompressing entry.");
            System.err.println("Entry: " + entry);
            System.err.println("Error: " + ex.getMessage());
            throw new IOException(ex.getMessage());
        }
        return entry;
    }

    /**
     * Decompress a given storage entry.
     *
     * We only decompress the content part of the storage entry.
     *
     * @param entry
     *
     * @return
     *
     * @throws java.io.IOException
     */
    public static JSocialKademliaStorageEntry decompress(final JSocialKademliaStorageEntry entry) throws IOException
    {
        if (!entry.isCompressed())
        {
            return entry;
        }
        try
        {
            entry.setContent(StringCompressor.decompress(entry.getContentString()));
            entry.setDecompressed();
        }
        catch (IOException ex)
        {
            System.err.println("Error whiles decompressing entry.");
            System.err.println("Entry: " + entry);
            System.err.println("Error: " + ex.getMessage());
            throw new IOException(ex.getMessage());
        }
        return entry;
    }
} public class UpToDateContentException extends Exception
{

    public UpToDateContentException()
    {
        super();
    }

    public UpToDateContentException(String message)
    {
        super(message);
    }
} public class MessageFactory implements KademliaMessageFactory
{

    private final JSocialKademliaNode localNode;
    private final SocialKademliaDHT dht;
    private final KadConfiguration config;

    public MessageFactory(JSocialKademliaNode local, SocialKademliaDHT dht, KadConfiguration config)
    {
        this.localNode = local;
        this.dht = dht;
        this.config = config;
    }

    @Override
    public Message createMessage(byte code, DataInputStream in) throws IOException
    {
        switch (code)
        {
            case AcknowledgeMessage.CODE:
                return new AcknowledgeMessage(in);
            case ConnectMessage.CODE:
                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);
            case ContentLookupMessageFUC.CODE:
                return new ContentLookupMessageFUC(in);
            case UpToDateContentMessage.CODE:
                return new UpToDateContentMessage(in);
            default:
                //System.out.println(this.localNode + " - No Message handler found for message. Code: " + code);
                return new SimpleMessage(in);

        }
    }

    @Override
    public Receiver createReceiver(byte code, KadServer server)
    {
        switch (code)
        {
            case ConnectMessage.CODE:
                return new ConnectReceiver(server, this.localNode);
            case ContentLookupMessage.CODE:
                return new ContentLookupReceiver(server, this.localNode, this.dht, this.config);
            case NodeLookupMessage.CODE:
                return new NodeLookupReceiver(server, this.localNode, this.config);
            case StoreContentMessage.CODE:
                return new StoreContentReceiver(server, this.localNode, this.dht);
            case ContentLookupMessageFUC.CODE:
                return new ContentLookupReceiverFUC(server, this.localNode, this.dht, this.config);
            default:
                //System.out.println("No receiver found for message. Code: " + code);
                return new SimpleReceiver();
        }
    }
} public class ConnectOperation implements Operation, Receiver
{

    public static final int MAX_CONNECT_ATTEMPTS = 5;       // Try 5 times to connect to a node

    private final KadServer server;
    private final SocialKademliaNode localNode;
    private final Node bootstrapNode;
    private final KadConfiguration config;

    private boolean error;
    private int attempts;

    /**
     * @param server    The message server used to send/receive messages
     * @param local     The local node
     * @param bootstrap Node to use to bootstrap the local node onto the network
     * @param config
     */
    public ConnectOperation(KadServer server, SocialKademliaNode local, Node bootstrap, KadConfiguration config)
    {
        this.server = server;
        this.localNode = local;
        this.bootstrapNode = bootstrap;
        this.config = config;
    }

    @Override
    public synchronized void execute() throws IOException
    {
        try
        {
            /* Contact the bootstrap node */
            this.error = true;
            this.attempts = 0;
            Message m = new ConnectMessage(this.localNode.getNode());

            /* Send a connect message to the bootstrap node */
            server.sendMessage(this.bootstrapNode, m, this);

            /* If we haven't finished as yet, wait for a maximum of config.operationTimeout() time */
            int totalTimeWaited = 0;
            int timeInterval = 50;     // We re-check every 300 milliseconds
            while (totalTimeWaited < this.config.operationTimeout())
            {
                if (error)
                {
                    wait(timeInterval);
                    totalTimeWaited += timeInterval;
                }
                else
                {
                    break;
                }
            }
            if (error)
            {
                /* If we still haven't received any responses by then, do a routing timeout */
                throw new RoutingException("ConnectOperation: Bootstrap node did not respond: " + bootstrapNode);
            }

            /* Perform lookup for our own ID to get nodes close to us */
            Operation lookup = new NodeLookupOperation(this.server, this.localNode, this.localNode.getNode().getNodeId(), this.config);
            lookup.execute();

            /**
             * Refresh buckets to get a good routing table
             * After the above lookup operation, K nodes will be in our routing table,
             * Now we try to populate all of our buckets.
             */
            new BucketRefreshOperation(this.server, this.localNode, this.config).execute();
        }
        catch (InterruptedException e)
        {
            System.err.println("Connect operation was interrupted. ");
        }
    }

    /**
     * Receives an AcknowledgeMessage from the bootstrap node.
     *
     * @param comm
     */
    @Override
    public synchronized void receive(Message incoming, int comm)
    {
        /* The incoming message will be an acknowledgement message */
        AcknowledgeMessage msg = (AcknowledgeMessage) incoming;

        /* The bootstrap node has responded, insert it into our space */
        this.localNode.getRoutingTable().insert(this.bootstrapNode);

        /* We got a response, so the error is false */
        error = false;

        /* Wake up any waiting thread */
        notify();
    }

    /**
     * Resends a ConnectMessage to the boot strap node a maximum of MAX_ATTEMPTS
     * times.
     *
     * @param comm
     *
     * @throws java.io.IOException
     */
    @Override
    public synchronized void timeout(int comm) throws IOException
    {
        if (++this.attempts < MAX_CONNECT_ATTEMPTS)
        {
            this.server.sendMessage(this.bootstrapNode, new ConnectMessage(this.localNode.getNode()), this);
        }
        else
        {
            /* We just exit, so notify all other threads that are possibly waiting */
            notify();
        }
    }
} public class ContentLookupOperation implements Operation, Receiver
{

    /* Constants */
    private static final Byte UNASKED = (byte) 0x00;
    private static final Byte AWAITING = (byte) 0x01;
    private static final Byte ASKED = (byte) 0x02;
    private static final Byte FAILED = (byte) 0x03;

    private final KadServer server;
    private final SocialKademliaNode localNode;
    private JSocialKademliaStorageEntry contentFound = null;
    private final KadConfiguration config;

    private final ContentLookupMessage lookupMessage;

    private boolean isContentFound;
    private final SortedMap<Node, Byte> nodes;

    /* Tracks messages in transit and awaiting reply */
    private final Map<Integer, Node> messagesTransiting;

    /* Used to sort nodes */
    private final Comparator comparator;

    /* Statistical information */
    private final RouteLengthChecker routeLengthChecker;

    private final GetParameter params;

    
    {
        messagesTransiting = new HashMap<>();
        isContentFound = false;
        routeLengthChecker = new RouteLengthChecker();
    }

    /**
     * @param server
     * @param localNode
     * @param params    The parameters to search for the content which we need to find
     * @param config
     */
    public ContentLookupOperation(KadServer server, SocialKademliaNode localNode, GetParameter params, KadConfiguration config)
    {
        /* Construct our lookup message */
        this.lookupMessage = new ContentLookupMessage(localNode.getNode(), params);

        this.server = server;
        this.localNode = localNode;
        this.config = config;

        this.params = params;

        /**
         * We initialize a TreeMap to store nodes.
         * This map will be sorted by which nodes are closest to the lookupId
         */
        this.comparator = new KeyComparator(params.getKey());
        this.nodes = new TreeMap(this.comparator);
    }

    /**
     * @throws java.io.IOException
     * @throws kademlia.exceptions.RoutingException
     */
    @Override
    public synchronized void execute() throws IOException, RoutingException
    {
        try
        {
            /* Set the local node as already asked */
            nodes.put(this.localNode.getNode(), ASKED);

            /**
             * Check if we are a connection to the required content's owner and if we have it's node in our routing table
             */
            if (this.localNode.getRoutingTable().containsConnection(this.params.getOwnerId()))
            {
                Node connNode = this.localNode.getRoutingTable().getConnectionNode(this.params.getOwnerId());
                this.nodes.put(connNode, UNASKED);
            }

            /**
             * We add all nodes here instead of the K-Closest because there may be the case that the K-Closest are offline
             * - The operation takes care of looking at the K-Closest.
             */
            List<Node> allNodes = this.localNode.getRoutingTable().getAllNodes();
            this.addNodes(allNodes);

            /* Also add the initial set of nodes to the routeLengthChecker */
            this.routeLengthChecker.addInitialNodes(allNodes);

            /**
             * If we haven't found the requested amount of content as yet,
             * keey trying until config.operationTimeout() time has expired
             */
            int totalTimeWaited = 0;
            int timeInterval = 10;     // We re-check every n milliseconds
            while (totalTimeWaited < this.config.operationTimeout())
            {
                if (!this.askNodesorFinish() && !isContentFound)
                {
                    wait(timeInterval);
                    totalTimeWaited += timeInterval;
                }
                else
                {
                    break;
                }
            }
        }
        catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        }
    }

    /**
     * Add nodes from this list to the set of nodes to lookup
     *
     * @param list The list from which to add nodes
     */
    public void addNodes(List<Node> list)
    {
        for (Node o : list)
        {
            /* If this node is not in the list, add the node */
            if (!nodes.containsKey(o))
            {
                nodes.put(o, UNASKED);
            }
        }
    }

    /**
     * Asks some of the K closest nodes seen but not yet queried.
     * Assures that no more than DefaultConfiguration.CONCURRENCY messages are in transit at a time
     *
     * This method should be called every time a reply is received or a timeout occurs.
     *
     * If all K closest nodes have been asked and there are no messages in transit,
     * the algorithm is finished.
     *
     * @return <code>true</code> if finished OR <code>false</code> otherwise
     */
    private boolean askNodesorFinish() throws IOException
    {
        /* If >= CONCURRENCY nodes are in transit, don't do anything */
        if (this.config.maxConcurrentMessagesTransiting() <= this.messagesTransiting.size())
        {
            return false;
        }

        /* Get unqueried nodes among the K closest seen that have not FAILED */
        List<Node> unasked = this.closestNodesNotFailed(UNASKED);

        if (unasked.isEmpty() && this.messagesTransiting.isEmpty())
        {
            /* We have no unasked nodes nor any messages in transit, we're finished! */
            return true;
        }

        /* Sort nodes according to criteria */
        Collections.sort(unasked, this.comparator);

        /**
         * Send messages to nodes in the list;
         * making sure than no more than CONCURRENCY messsages are in transit
         */
        for (int i = 0; (this.messagesTransiting.size() < this.config.maxConcurrentMessagesTransiting()) && (i < unasked.size()); i++)
        {
            Node n = (Node) unasked.get(i);

            int comm = server.sendMessage(n, lookupMessage, this);

            this.nodes.put(n, AWAITING);
            this.messagesTransiting.put(comm, n);
        }

        /* We're not finished as yet, return false */
        return false;
    }

    /**
     * Find The K closest nodes to the target lookupId given that have not FAILED.
     * From those K, get those that have the specified status
     *
     * @param status The status of the nodes to return
     *
     * @return A List of the closest nodes
     */
    private List<Node> closestNodesNotFailed(Byte status)
    {
        List<Node> closestNodes = new ArrayList<>(this.config.k());
        int remainingSpaces = this.config.k();

        for (Map.Entry e : this.nodes.entrySet())
        {
            if (!FAILED.equals(e.getValue()))
            {
                if (status.equals(e.getValue()))
                {
                    /* We got one with the required status, now add it */
                    closestNodes.add((Node) e.getKey());
                }

                if (--remainingSpaces == 0)
                {
                    break;
                }
            }
        }

        return closestNodes;
    }

    @Override
    public synchronized void receive(Message incoming, int comm) throws IOException, RoutingException
    {
        if (this.isContentFound)
        {
            return;
        }

        if (incoming instanceof ContentMessage)
        {
            /* The reply received is a content message with the required content, take it in */
            ContentMessage msg = (ContentMessage) incoming;

            /* Add the origin node to our routing table */
            this.localNode.getRoutingTable().insert(msg.getOrigin());

            /* Get the Content and check if it satisfies the required parameters */
            JSocialKademliaStorageEntry content = msg.getContent();
            this.contentFound = content;
            this.isContentFound = true;
        }
        else
        {
            /* The reply received is a NodeReplyMessage with nodes closest to the content needed */
            NodeReplyMessage msg = (NodeReplyMessage) incoming;

            /* Add the origin node to our routing table */
            Node origin = msg.getOrigin();
            this.localNode.getRoutingTable().insert(origin);

            /* Set that we've completed ASKing the origin node */
            this.nodes.put(origin, ASKED);

            /* Remove this msg from messagesTransiting since it's completed now */
            this.messagesTransiting.remove(comm);

            /* Add the received nodes to the routeLengthChecker */
            this.routeLengthChecker.addNodes(msg.getNodes(), origin);

            /* Add the received nodes to our nodes list to query */
            this.addNodes(msg.getNodes());
            this.askNodesorFinish();
        }
    }

    /**
     * A node does not respond or a packet was lost, we set this node as failed
     *
     * @param comm
     *
     * @throws java.io.IOException
     */
    @Override
    public synchronized void timeout(int comm) throws IOException
    {
        /* Get the node associated with this communication */
        Node n = this.messagesTransiting.get(new Integer(comm));

        if (n == null)
        {
            throw new UnknownMessageException("Unknown comm: " + comm);
        }

        /* Mark this node as failed and inform the routing table that it's unresponsive */
        this.nodes.put(n, FAILED);
        this.localNode.getRoutingTable().setUnresponsiveContact(n);
        this.messagesTransiting.remove(comm);

        this.askNodesorFinish();
    }

    /**
     * @return Whether the content was found or not.
     */
    public boolean isContentFound()
    {
        return this.isContentFound;
    }

    /**
     * @return The list of all content found during the lookup operation
     *
     * @throws kademlia.exceptions.ContentNotFoundException
     */
    public JSocialKademliaStorageEntry getContentFound() throws ContentNotFoundException
    {
        if (this.isContentFound)
        {
            return this.contentFound;
        }
        else
        {
            throw new ContentNotFoundException("No Value was found for the given key.");
        }
    }

    /**
     * @return How many hops it took in order to get to the content.
     */
    public int routeLength()
    {
        return this.routeLengthChecker.getRouteLength();
    }
} public class ContentLookupOperationFUC implements Operation, Receiver
{

    /* Constants */
    private static final Byte UNASKED = (byte) 0x00;
    private static final Byte AWAITING = (byte) 0x01;
    private static final Byte ASKED = (byte) 0x02;
    private static final Byte FAILED = (byte) 0x03;

    private final KadServer server;
    private final SocialKademliaNode localNode;
    private JSocialKademliaStorageEntry contentFound = null;
    private final KadConfiguration config;

    private final Message lookupMessage;

    private boolean isContentFound;
    private boolean newerContentExist = false; // Whether the content we have is up to date
    private final SortedMap<Node, Byte> nodes;

    /* Tracks messages in transit and awaiting reply */
    private final Map<Integer, Node> messagesTransiting;

    /* Used to sort nodes */
    private final Comparator comparator;

    /* Statistical information */
    private final RouteLengthChecker routeLengthChecker;

    private final GetParameterFUC params;

    
    {
        messagesTransiting = new HashMap<>();
        isContentFound = false;
        routeLengthChecker = new RouteLengthChecker();
    }

    /**
     * @param server
     * @param localNode
     * @param params    The parameters to search for the content which we need to find
     * @param config
     */
    public ContentLookupOperationFUC(KadServer server, SocialKademliaNode localNode, GetParameterFUC params, KadConfiguration config)
    {
        /* Construct our lookup message */
        this.lookupMessage = new ContentLookupMessageFUC(localNode.getNode(), params);

        this.server = server;
        this.localNode = localNode;
        this.config = config;

        this.params = params;

        /**
         * We initialize a TreeMap to store nodes.
         * This map will be sorted by which nodes are closest to the lookupId
         */
        this.comparator = new KeyComparator(params.getKey());
        this.nodes = new TreeMap(this.comparator);
    }

    /**
     * @throws java.io.IOException
     * @throws kademlia.exceptions.RoutingException
     */
    @Override
    public synchronized void execute() throws IOException, RoutingException
    {
        try
        {
            /* Set the local node as already asked */
            nodes.put(this.localNode.getNode(), ASKED);

            /**
             * Check if we are a connection to the required content's owner and if we have it's node in our routing table
             */
            if (this.localNode.getRoutingTable().containsConnection(this.params.getOwnerId()))
            {
                Node connNode = this.localNode.getRoutingTable().getConnectionNode(this.params.getOwnerId());

                /* We only contact the owner of the contact if this is not the owner */
                if (!connNode.equals(this.localNode.getNode()))
                {
                    this.nodes.put(connNode, UNASKED);
                }
            }

            /**
             * We add all nodes here instead of the K-Closest because there may be the case that the K-Closest are offline
             * - The operation takes care of looking at the K-Closest.
             */
            List<Node> allNodes = this.localNode.getRoutingTable().getAllNodes();
            this.addNodes(allNodes);

            /* Also add the initial set of nodes to the routeLengthChecker */
            this.routeLengthChecker.addInitialNodes(allNodes);

            /**
             * If we haven't found the requested amount of content as yet,
             * keey trying until config.operationTimeout() time has expired
             */
            int totalTimeWaited = 0;
            int timeInterval = 10;     // We re-check every n milliseconds
            while (totalTimeWaited < this.config.operationTimeout())
            {
                if (!this.askNodesorFinish() && !isContentFound)
                {
                    wait(timeInterval);
                    totalTimeWaited += timeInterval;
                }
                else
                {
                    break;
                }
            }
        }
        catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        }
    }

    /**
     * Add nodes from this list to the set of nodes to lookup
     *
     * @param list The list from which to add nodes
     */
    public void addNodes(List<Node> list)
    {
        for (Node o : list)
        {
            /* If this node is not in the list, add the node */
            if (!nodes.containsKey(o))
            {
                nodes.put(o, UNASKED);
            }
        }
    }

    /**
     * Asks some of the K closest nodes seen but not yet queried.
     * Assures that no more than DefaultConfiguration.CONCURRENCY messages are in transit at a time
     *
     * This method should be called every time a reply is received or a timeout occurs.
     *
     * If all K closest nodes have been asked and there are no messages in transit,
     * the algorithm is finished.
     *
     * @return <code>true</code> if finished OR <code>false</code> otherwise
     */
    private boolean askNodesorFinish() throws IOException
    {
        /* If >= CONCURRENCY nodes are in transit, don't do anything */
        if (this.config.maxConcurrentMessagesTransiting() <= this.messagesTransiting.size())
        {
            return false;
        }

        /* Get unqueried nodes among the K closest seen that have not FAILED */
        List<Node> unasked = this.closestNodesNotFailed(UNASKED);

        if (unasked.isEmpty() && this.messagesTransiting.isEmpty())
        {
            /* We have no unasked nodes nor any messages in transit, we're finished! */
            return true;
        }

        /* Sort nodes according to criteria */
        Collections.sort(unasked, this.comparator);

        /**
         * Send messages to nodes in the list;
         * making sure than no more than CONCURRENCY messsages are in transit
         */
        for (int i = 0; (this.messagesTransiting.size() < this.config.maxConcurrentMessagesTransiting()) && (i < unasked.size()); i++)
        {
            Node n = (Node) unasked.get(i);

            int comm = server.sendMessage(n, lookupMessage, this);

            this.nodes.put(n, AWAITING);
            this.messagesTransiting.put(comm, n);
        }

        /* We're not finished as yet, return false */
        return false;
    }

    /**
     * Find The K closest nodes to the target lookupId given that have not FAILED.
     * From those K, get those that have the specified status
     *
     * @param status The status of the nodes to return
     *
     * @return A List of the closest nodes
     */
    private List<Node> closestNodesNotFailed(Byte status)
    {
        List<Node> closestNodes = new ArrayList<>(this.config.k());
        int remainingSpaces = this.config.k();

        for (Map.Entry e : this.nodes.entrySet())
        {
            if (!FAILED.equals(e.getValue()))
            {
                if (status.equals(e.getValue()))
                {
                    /* We got one with the required status, now add it */
                    closestNodes.add((Node) e.getKey());
                }

                if (--remainingSpaces == 0)
                {
                    break;
                }
            }
        }

        return closestNodes;
    }

    @Override
    public synchronized void receive(Message incoming, int comm) throws IOException, RoutingException
    {
        if (this.isContentFound)
        {
            return;
        }

        if (incoming instanceof ContentMessage)
        {
            /* The reply received is a content message with the required content, take it in */
            ContentMessage msg = (ContentMessage) incoming;

            /* ContentMessage node to our routing table */
            this.localNode.getRoutingTable().insert(msg.getOrigin());

            /* Get the Content and check if it satisfies the required parameters */
            JSocialKademliaStorageEntry content = msg.getContent();
            this.contentFound = content;
            this.isContentFound = true;
            this.newerContentExist = true;
        }
        else if (incoming instanceof UpToDateContentMessage)
        {
            /**
             * The content we have is up to date
             * No sense in adding our own content to the resultset, so lets just exit
             */
            this.newerContentExist = false;
            this.isContentFound = true;
        }
        else if (incoming instanceof NodeReplyMessage)
        {
            /* The reply received is a NodeReplyMessage with nodes closest to the content needed */
            NodeReplyMessage msg = (NodeReplyMessage) incoming;

            /* Add the origin node to our routing table */
            Node origin = msg.getOrigin();
            this.localNode.getRoutingTable().insert(origin);

            /* Set that we've completed ASKing the origin node */
            this.nodes.put(origin, ASKED);

            /* Remove this msg from messagesTransiting since it's completed now */
            this.messagesTransiting.remove(comm);

            /* Add the received nodes to the routeLengthChecker */
            this.routeLengthChecker.addNodes(msg.getNodes(), origin);

            /* Add the received nodes to our nodes list to query */
            this.addNodes(msg.getNodes());
            this.askNodesorFinish();
        }
        else
        {
            /* @todo Something is wrong that we get messages of other types here, investigate it! */
            System.err.println(this.localNode.getNode() + " Got a response message of type " + incoming.getClass() + " from " + ((ContentLookupMessageFUC) incoming).getOrigin() + " in ContentLookupOperationFUC. ");
        }
    }

    /**
     * A node does not respond or a packet was lost, we set this node as failed
     *
     * @param comm
     *
     * @throws java.io.IOException
     */
    @Override
    public synchronized void timeout(int comm) throws IOException
    {
        /* Get the node associated with this communication */
        Node n = this.messagesTransiting.get(new Integer(comm));

        if (n == null)
        {
            throw new UnknownMessageException("Unknown comm: " + comm);
        }

        /* Mark this node as failed and inform the routing table that it's unresponsive */
        this.nodes.put(n, FAILED);
        this.localNode.getRoutingTable().setUnresponsiveContact(n);
        this.messagesTransiting.remove(comm);

        this.askNodesorFinish();
    }

    /**
     * @return Whether the content was found or not.
     */
    public boolean isContentFound()
    {
        return this.isContentFound;
    }

    /**
     * @return The list of all content found during the lookup operation
     *
     * @throws kademlia.exceptions.UpToDateContentException
     */
    public JSocialKademliaStorageEntry getContentFound() throws UpToDateContentException
    {
        /* Check if we have newer content */
        if (this.newerContentExist)
        {
            return this.contentFound;
        }
        else
        {
            throw new UpToDateContentException("The content is up to date");
        }
    }

    /**
     * @return Boolean whether newer content exist or not
     */
    public boolean newerContentExist()
    {
        return this.newerContentExist;
    }

    /**
     * @return How many hops it took in order to get to the content.
     */
    public int routeLength()
    {
        return this.routeLengthChecker.getRouteLength();
    }
} public class KadRefreshOperation implements Operation
{

    private final KadServer server;
    private final SocialKademliaNode localNode;
    private final SocialKademliaDHT dht;
    private final KadConfiguration config;

    public KadRefreshOperation(KadServer server, SocialKademliaNode localNode, SocialKademliaDHT dht, KadConfiguration config)
    {
        this.server = server;
        this.localNode = localNode;
        this.dht = dht;
        this.config = config;
    }

    @Override
    public void execute() throws IOException
    {
        /* Run our BucketRefreshOperation to refresh buckets */
        new BucketRefreshOperation(this.server, this.localNode, this.config).execute();

        /* After buckets have been refreshed, we refresh content */
        new ContentRefreshOperation(this.server, this.localNode, this.dht, this.config).execute();
    }
} public class StoreOperation implements Operation
{

    private final KadServer server;
    private final SocialKademliaNode localNode;
    private final JSocialKademliaStorageEntry storageEntry;
    private final SocialKademliaDHT localDht;
    private final KadConfiguration config;

    /**
     * @param server
     * @param localNode
     * @param storageEntry The content to be stored on the DHT
     * @param localDht     The local DHT
     * @param config
     */
    public StoreOperation(KadServer server, SocialKademliaNode localNode, JSocialKademliaStorageEntry storageEntry, SocialKademliaDHT localDht, KadConfiguration config)
    {
        this.server = server;
        this.localNode = localNode;
        this.storageEntry = storageEntry;
        this.localDht = localDht;
        this.config = config;
    }

    @Override
    public synchronized void execute() throws IOException
    {
        /* Get the nodes on which we need to store the content */
        NodeLookupOperation ndlo = new NodeLookupOperation(this.server, this.localNode, this.storageEntry.getContentMetadata().getKey(), this.config);
        ndlo.execute();
        List<Node> nodes = ndlo.getClosestNodes();

        /* Create the message */
        Message msg = new StoreContentMessage(this.localNode.getNode(), this.storageEntry);

        /*Store the message on all of the K-Nodes*/
        for (Node n : nodes)
        {
            if (n.equals(this.localNode.getNode()))
            {
                /* Store the content locally */
                this.localDht.store(this.storageEntry);
            }
            else
            {
                /**
                 * @todo Create a receiver that receives a store acknowledgement message to count how many nodes a content have been stored at
                 */
                this.server.sendMessage(n, msg, null);
            }
        }
    }

    /**
     * @return The number of nodes that have stored this content
     *
     * @todo Implement this method
     */
    public int numNodesStoredAt()
    {
        return 1;
    }
} public interface SocialKademliaRoutingTable extends KademliaRoutingTable
{

    /**
     * Adds a Connection to the routing table
     *
     * @param actorId The actor Id of this connection
     * @param node    The node related to this connection
     */
    public void insertConnection(String actorId, Node node);

    /**
     * Checks if this routing table contain a connection.
     *
     * @param actorId The id of the actor
     *
     * @return boolean
     */
    public boolean containsConnection(String actorId);

    /**
     * Remove a connection from this routing table.
     *
     * @param actorId
     *
     * @return Boolean whether the removal was successful.
     */
    public boolean removeConnection(String actorId);

    /**
     * @param actorId
     *
     * @return The node of a given connection
     */
    public Node getConnectionNode(String actorId);

    /**
     * @return An Iterable structure with all connections in this routing table
     */
    public List<Connection> getConnections();
} public class JSocialKademliaRoutingTable extends JKademliaRoutingTable implements SocialKademliaRoutingTable
{

    /* Connections of the actor of this node in the DOSN */
    private final HashMap<String, Connection> connections;

    
    {
        connections = new HashMap<>();
    }

    public JSocialKademliaRoutingTable(Node localNode, KadConfiguration config)
    {
        super(localNode, config);
    }

    @Override
    public synchronized void insertConnection(String actorId, Node node)
    {
        if (this.containsConnection(actorId))
        {
            Connection c = this.connections.get(actorId);
            c.setNode(node);
            c.setSeenNow();
        }
        else
        {
            this.connections.put(actorId, new Connection(actorId, node));
        }
    }

    @Override
    public synchronized boolean containsConnection(String actorId)
    {
        return this.connections.containsKey(actorId);
    }

    @Override
    public synchronized boolean removeConnection(String actorId)
    {
        if (this.containsConnection(actorId))
        {
            this.connections.remove(actorId);
            return true;
        }
        return false;
    }

    @Override
    public synchronized List<Connection> getConnections()
    {
        return this.connections.isEmpty() ? new ArrayList<>() : new ArrayList<>(this.connections.values());
    }

    @Override
    public synchronized Node getConnectionNode(String actorId)
    {
        if (this.containsConnection(actorId))
        {
            return this.connections.get(actorId).getNode();
        }
        else
        {
            throw new NoSuchElementException("This connection does not exist.");
        }
    }
} public interface SocialKademliaDHT
{

    /**
     * Initialize this DHT to it's default state
     */
    public void initialize();

    /**
     * Set a new configuration. Mainly used when we restore the DHT state from a file
     *
     * @param con The new configuration file
     */
    public void setConfiguration(KadConfiguration con);

    /**
     * Creates a new Serializer or returns an existing serializer
     *
     * @return The new ContentSerializer
     */
    public KadSerializer<JSocialKademliaStorageEntry> getSerializer();

    /**
     * Handle storing content locally
     *
     * @param content The DHT content to store
     *
     * @return boolean true if we stored the content, false if the content already exists and is up to date
     *
     * @throws java.io.IOException
     */
    public boolean store(JSocialKademliaStorageEntry content) throws IOException;

    public boolean store(KadContent content) throws IOException;

    /**
     * Retrieves a Content from local storage
     *
     * @param key      The Key of the content to retrieve
     * @param hashCode The hash code of the content to retrieve
     *
     * @return A KadContent object
     *
     * @throws java.io.FileNotFoundException
     * @throws java.lang.ClassNotFoundException
     */
    public JSocialKademliaStorageEntry retrieve(KademliaId key, int hashCode) throws FileNotFoundException, IOException, ClassNotFoundException;

    /**
     * Check if any content for the given criteria exists in this DHT
     *
     * @param param The content search criteria
     *
     * @return boolean Whether any content exist that satisfy the criteria
     */
    public boolean contains(GetParameter param);

    /**
     * Retrieve and create a KadContent object given the StorageEntry object
     *
     * @param entry The StorageEntry used to retrieve this content
     *
     * @return KadContent The content object
     *
     * @throws java.io.IOException
     */
    public JSocialKademliaStorageEntry get(SocialKademliaStorageEntryMetadata entry) throws IOException, NoSuchElementException;

    /**
     * Get the StorageEntry for the content if any exist.
     *
     * @param param The parameters used to filter the content needed
     *
     * @return KadContent A KadContent found on the DHT satisfying the given criteria
     *
     * @throws java.io.IOException
     */
    public JSocialKademliaStorageEntry get(GetParameter param) throws NoSuchElementException, IOException;

    /**
     * Delete a content from local storage
     *
     * @param content The Content to Remove
     *
     *
     * @throws kademlia.exceptions.ContentNotFoundException
     */
    public void remove(KadContent content) throws ContentNotFoundException;

    public void remove(SocialKademliaStorageEntryMetadata entry) throws ContentNotFoundException;

    /**
     * @return A List of all StorageEntries for this node
     */
    public List<SocialKademliaStorageEntryMetadata> getStorageEntries();

    /**
     * Used to add a list of storage entries for existing content to the DHT.
     * Mainly used when retrieving StorageEntries from a saved state file.
     *
     * @param ientries The entries to add
     */
    public void putStorageEntries(List<SocialKademliaStorageEntryMetadata> ientries);

    /**
     * Handle storing content locally to keep the content cached.
     *
     * We set that this content is a cached entry and that this node is not one of the k-nodes.
     *
     * @param content The DHT content to store
     *
     * @return boolean true if we stored the content, false if the content already exists and is up to date
     *
     * @throws java.io.IOException
     */
    public boolean cache(JSocialKademliaStorageEntry content) throws IOException;

    public boolean cache(KadContent content) throws IOException;

    /**
     * Update a content; the operation is only done iff we already have a copy of the content here
     *
     * @param newContent The content to update.
     *
     * @throws java.io.IOException
     */
    public void update(JSocialKademliaStorageEntry newContent) throws IOException;
} public interface SocialKademliaStorageEntry extends KademliaStorageEntry
{

    /**
     * Set the content value of this storage entry
     *
     * @param data
     */
    public void setContent(String data);

    /**
     * @return Whether this entry is compressed or not
     */
    public boolean isCompressed();

    /**
     * Specify that the storage entry is compressed
     */
    public void setCompressed();

    /**
     * Specify that the storage entry is decompressed
     */
    public void setDecompressed();

    /**
     * @return The content in string format
     */
    public String getContentString();

    @Override
    public SocialKademliaStorageEntryMetadata getContentMetadata();
} public class JsonSocialKadRoutingTableSerializer implements KadSerializer<SocialKademliaRoutingTable>
{

    private final Gson gson;

    Type contactCollectionType = new TypeToken<List<Contact>>()
    {
    }.getType();

    private final KadConfiguration config;
    
    
    {
        gson = new Gson();
    }

    /**
     * Initialize the class
     *
     * @param config
     */
    public JsonSocialKadRoutingTableSerializer(KadConfiguration config)
    {
        this.config = config;
    }

    @Override
    public void write(SocialKademliaRoutingTable data, DataOutputStream out) throws IOException
    {
        try (JsonWriter writer = new JsonWriter(new OutputStreamWriter(out)))
        {
            writer.beginArray();

            /* Write the basic JKademliaRoutingTable */
            gson.toJson(data, JSocialKademliaRoutingTable.class, writer);

            /* Now Store the Contacts  */
            gson.toJson(data.getAllContacts(), contactCollectionType, writer);

            writer.endArray();
        }
    }

    @Override
    public SocialKademliaRoutingTable read(DataInputStream in) throws IOException, ClassNotFoundException
    {
        try (DataInputStream din = new DataInputStream(in);
                JsonReader reader = new JsonReader(new InputStreamReader(in)))
        {
            reader.beginArray();

            /* Read the basic JKademliaRoutingTable */
            SocialKademliaRoutingTable tbl = gson.fromJson(reader, JSocialKademliaRoutingTable.class);
            tbl.setConfiguration(config);

            /* Now get the Contacts and add them back to the JKademliaRoutingTable */
            List<Contact> contacts = gson.fromJson(reader, contactCollectionType);
            tbl.initialize();

            for (Contact c : contacts)
            {
                tbl.insert(c);
            }

            reader.endArray();
            /* Read and return the Content*/
            return tbl;
        }
    }
} public class JsonSocialKademliaDHTSerializer implements KadSerializer<SocialKademliaDHT>
{

    private final Gson gson;
    private final Type storageEntriesCollectionType;

    
    {
        gson = new Gson();

        storageEntriesCollectionType = new TypeToken<List<JSocialKademliaStorageEntryMetadata>>()
        {
        }.getType();
    }

    @Override
    public void write(SocialKademliaDHT data, DataOutputStream out) throws IOException
    {
        try (JsonWriter writer = new JsonWriter(new OutputStreamWriter(out)))
        {
            writer.beginArray();

            /* Write the basic DHT */
            gson.toJson(data, JSocialKademliaDHT.class, writer);

            /* Now Store the Entries  */
            gson.toJson(data.getStorageEntries(), this.storageEntriesCollectionType, writer);

            writer.endArray();
        }

    }

    @Override
    public SocialKademliaDHT read(DataInputStream in) throws IOException, ClassNotFoundException
    {
        try (DataInputStream din = new DataInputStream(in);
                JsonReader reader = new JsonReader(new InputStreamReader(in)))
        {
            reader.beginArray();

            /* Read the basic DHT */
            SocialKademliaDHT dht = gson.fromJson(reader, JSocialKademliaDHT.class);
            dht.initialize();

            /* Now get the entries and add them back to the DHT */
            List<SocialKademliaStorageEntryMetadata> entries = gson.fromJson(reader, this.storageEntriesCollectionType);
            dht.putStorageEntries(entries);

            reader.endArray();
            return dht;
        }
    }
}
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.util.NoSuchElementException;
import java.util.Timer;
import java.util.TimerTask;
import kademlia.DefaultConfiguration;
import kademlia.dht.GetParameter;
import kademlia.KadConfiguration;
import kademlia.KadServer;
import socialkademlia.dht.JSocialKademliaDHT;
import socialkademlia.dht.GetParameterFUC;
import kademlia.dht.KadContent;
import kademlia.dht.KademliaStorageEntry;
import socialkademlia.dht.JSocialKademliaStorageEntry;
import socialkademlia.dht.util.StorageEntryCompressor;
import kademlia.exceptions.ContentNotFoundException;
import kademlia.exceptions.RoutingException;
import kademlia.message.KademliaMessageFactory;
import socialkademlia.exceptions.UpToDateContentException;
import socialkademlia.message.MessageFactory;
import kademlia.node.Node;
import kademlia.node.KademliaId;
import socialkademlia.operation.ConnectOperation;
import socialkademlia.operation.ContentLookupOperation;
import socialkademlia.operation.ContentLookupOperationFUC;
import kademlia.operation.Operation;
import socialkademlia.operation.KadRefreshOperation;
import socialkademlia.operation.StoreOperation;
import socialkademlia.routing.SocialKademliaRoutingTable;
import socialkademlia.routing.JSocialKademliaRoutingTable;
import kademlia.util.serializer.JsonSerializer;
import socialkademlia.dht.SocialKademliaDHT;
import socialkademlia.dht.SocialKademliaStorageEntry;
import socialkademlia.util.serializer.JsonSocialKadRoutingTableSerializer;
import socialkademlia.util.serializer.JsonSocialKademliaDHTSerializer;
package socialkademlia;


/**
 * The main Kademlia Node on the network, this node manages everything for this local system.
 *
 * @author Joshua Kissoon
 * @since 20140215
 *
 * @todo When we receive a store message - if we have a newer version of the content, re-send this newer version to that node so as to update their version
 * @todo Handle IPv6 Addresses
 *
 */
public class JSocialKademliaNode implements SocialKademliaNode
{

    /* Kademlia Attributes */
    private final String ownerId;

    /* Objects to be used */
    private final transient Node localNode;
    private final transient KadServer server;
    private final transient SocialKademliaDHT dht;
    private transient SocialKademliaRoutingTable routingTable;
    private final int udpPort;
    private transient KadConfiguration config;

    /* Timer used to execute refresh operations */
    private transient Timer refreshOperationTimer;
    private transient TimerTask refreshOperationTTask;

    /* Factories */
    private final transient KademliaMessageFactory messageFactory;

    /* Statistics */
    private final transient SocialKadStatistician statistician;

    
    {
        statistician = new Statistician();
    }

    /**
     * Creates a Kademlia DistributedMap using the specified name as filename base.
     * If the id cannot be read from disk the specified defaultId is used.
     * The instance is bootstraped to an existing network by specifying the
     * address of a bootstrap node in the network.
     *
     * @param ownerId      The Name of this node used for storage
     * @param localNode    The Local Node for this Kad instance
     * @param udpPort      The UDP port to use for routing messages
     * @param dht          The JSocialKademliaDHT for this instance
     * @param config
     * @param routingTable
     *
     * @throws IOException If an error occurred while reading id or local map
     *                     from disk <i>or</i> a network error occurred while
     *                     attempting to bootstrap to the network
     * */
    public JSocialKademliaNode(String ownerId, Node localNode, int udpPort, SocialKademliaDHT dht, SocialKademliaRoutingTable routingTable, KadConfiguration config) throws IOException
    {
        this.ownerId = ownerId;
        this.udpPort = udpPort;
        this.localNode = localNode;
        this.dht = dht;
        this.config = config;
        this.routingTable = routingTable;
        this.messageFactory = new MessageFactory(this, this.dht, this.config);
        this.server = new KadServer(udpPort, this.messageFactory, this.localNode, this.config, this.statistician);
        this.startRefreshOperation();
    }

    /**
     * Schedule the recurring refresh operation
     */
    @Override
    public final void startRefreshOperation()
    {
        this.refreshOperationTimer = new Timer(true);
        refreshOperationTTask = new TimerTask()
        {
            @Override
            public void run()
            {
                try
                {
                    /* Runs a JSocialKademliaDHT RefreshOperation  */
                    JSocialKademliaNode.this.refresh();
                }
                catch (IOException e)
                {
                    System.err.println("KademliaNode: Refresh Operation Failed; Message: " + e.getMessage());
                }
            }
        };
        refreshOperationTimer.schedule(refreshOperationTTask, this.config.restoreInterval(), this.config.restoreInterval());
    }

    @Override
    public final void stopRefreshOperation()
    {
        /* Close off the timer tasks */
        this.refreshOperationTTask.cancel();
        this.refreshOperationTimer.cancel();
        this.refreshOperationTimer.purge();
    }

    public JSocialKademliaNode(String ownerId, Node node, int udpPort, SocialKademliaRoutingTable routingTable, KadConfiguration config) throws IOException
    {
        this(
                ownerId,
                node,
                udpPort,
                new JSocialKademliaDHT(ownerId, config),
                routingTable,
                config
        );
    }

    public JSocialKademliaNode(String ownerId, Node node, int udpPort, KadConfiguration config) throws IOException
    {
        this(
                ownerId,
                node,
                udpPort,
                new JSocialKademliaRoutingTable(node, config),
                config
        );
    }

    public JSocialKademliaNode(String ownerId, KademliaId defaultId, int udpPort) throws IOException
    {
        this(
                ownerId,
                new Node(defaultId, InetAddress.getLocalHost(), udpPort),
                udpPort,
                new DefaultConfiguration()
        );
    }

    /**
     * Load Stored state using default configuration
     *
     * @param ownerId The ID of the owner for the stored state
     *
     * @return A Kademlia instance loaded from a stored state in a file
     *
     * @throws java.io.FileNotFoundException
     * @throws java.lang.ClassNotFoundException
     */
    public static JSocialKademliaNode loadFromFile(String ownerId) throws FileNotFoundException, IOException, ClassNotFoundException
    {
        return JSocialKademliaNode.loadFromFile(ownerId, new DefaultConfiguration());
    }

    /**
     * Load Stored state
     *
     * @param ownerId The ID of the owner for the stored state
     * @param iconfig Configuration information to work with
     *
     * @return A Kademlia instance loaded from a stored state in a file
     *
     * @throws java.io.FileNotFoundException
     * @throws java.lang.ClassNotFoundException
     */
    public static JSocialKademliaNode loadFromFile(String ownerId, KadConfiguration iconfig) throws FileNotFoundException, IOException, ClassNotFoundException
    {
        DataInputStream din;

        /**
         * @section Read Basic Kad data
         */
        din = new DataInputStream(new FileInputStream(getStateStorageFolderName(ownerId, iconfig) + File.separator + "kad.kns"));
        JSocialKademliaNode ikad = new JsonSerializer<JSocialKademliaNode>().read(din);

        /**
         * @section Read the routing table
         */
        din = new DataInputStream(new FileInputStream(getStateStorageFolderName(ownerId, iconfig) + File.separator + "routingtable.kns"));
        SocialKademliaRoutingTable irtbl = new JsonSocialKadRoutingTableSerializer(iconfig).read(din);

        /**
         * @section Read the node state
         */
        din = new DataInputStream(new FileInputStream(getStateStorageFolderName(ownerId, iconfig) + File.separator + "node.kns"));
        Node inode = new JsonSerializer<Node>().read(din);

        /**
         * @section Read the DHT
         */
        din = new DataInputStream(new FileInputStream(getStateStorageFolderName(ownerId, iconfig) + File.separator + "dht.kns"));
        SocialKademliaDHT idht = new JsonSocialKademliaDHTSerializer().read(din);
        idht.setConfiguration(iconfig);

        return new JSocialKademliaNode(ownerId, inode, ikad.getPort(), idht, irtbl, iconfig);
    }

    /**
     * @return Node The local node for this system
     */
    @Override
    public Node getNode()
    {
        return this.localNode;
    }

    /**
     * @return The KadServer used to send/receive messages
     */
    @Override
    public KadServer getServer()
    {
        return this.server;
    }

    /**
     * @return The JSocialKademliaDHT for this kad instance
     */
    @Override
    public SocialKademliaDHT getDHT()
    {
        return this.dht;
    }

    /**
     * @return The current KadConfiguration object being used
     */
    @Override
    public KadConfiguration getCurrentConfiguration()
    {
        return this.config;
    }

    /**
     * Connect to an existing peer-to-peer network.
     *
     * @param n The known node in the peer-to-peer network
     *
     * @throws RoutingException      If the bootstrap node could not be contacted
     * @throws IOException           If a network error occurred
     * @throws IllegalStateException If this object is closed
     * */
    @Override
    public synchronized final void bootstrap(Node n) throws IOException, RoutingException
    {
        long startTime = System.nanoTime();
        Operation op = new ConnectOperation(this.server, this, n, this.config);
        op.execute();
        long endTime = System.nanoTime();
        this.statistician.setBootstrapTime(endTime - startTime);
    }

    /**
     * Stores the specified value under the given key
     * This value is stored on K nodes on the network, or all nodes if there are > K total nodes in the network
     *
     * @param content The content to put onto the JSocialKademliaDHT
     *
     * @return Integer How many nodes the content was stored on
     *
     * @throws java.io.IOException
     *
     */
    @Override
    public int put(KadContent content) throws IOException
    {
        return this.put(new JSocialKademliaStorageEntry(content));
    }

    /**
     * Stores the specified value under the given key
     * This value is stored on K nodes on the network, or all nodes if there are > K total nodes in the network
     *
     * @param entry The JSocialKademliaStorageEntry with the content to put onto the JSocialKademliaDHT
     *
     * @return Integer How many nodes the content was stored on
     *
     * @throws java.io.IOException
     *
     */
    public int put(JSocialKademliaStorageEntry entry) throws IOException
    {
        StoreOperation sop = new StoreOperation(this.server, this, this.compressStorageEntry(entry), this.dht, this.config);
        sop.execute();

        /* Return how many nodes the content was stored on */
        return sop.numNodesStoredAt();
    }

    /**
     * Put the data on the network and also cache a copy locally
     *
     * @param content The content to store
     *
     * @return How many nodes the content has been stored at excluding the local node.
     *
     * @throws java.io.IOException
     */
    public int putAndCache(KadContent content) throws IOException
    {
        JSocialKademliaStorageEntry entry = new JSocialKademliaStorageEntry(content);

        this.cache(entry);
        return this.put(entry);
    }

    /**
     * Store a content on the local node's JSocialKademliaDHT
     *
     * @param content The content to put on the JSocialKademliaDHT
     *
     * @throws java.io.IOException
     */
    @Override
    public void putLocally(KadContent content) throws IOException
    {
        this.dht.store(this.compressStorageEntry(new JSocialKademliaStorageEntry(content)));
    }

    /**
     * Stores the specified value under the given key locally;
     * This content is permanently stored locally and will not be deleted unless the cache is cleared.
     *
     * @param content The content to put onto the local JSocialKademliaDHT
     *
     * @throws java.io.IOException
     *
     */
    public void cache(KadContent content) throws IOException
    {
        this.cache(new JSocialKademliaStorageEntry(content));
    }

    private void cache(JSocialKademliaStorageEntry entry) throws IOException
    {
        this.dht.cache(this.compressStorageEntry(entry));
    }

    /**
     * Compress the storage entry
     */
    private JSocialKademliaStorageEntry compressStorageEntry(final JSocialKademliaStorageEntry entry)
    {
        try
        {
            return StorageEntryCompressor.compress(entry);