path: root/src/main/java/org/gnunet/dht/DistributedHashTable.java

/*
 This file is part of GNUnet.
 Copyright (C) 2011, 2012 Christian Grothoff (and other contributing authors)

 GNUnet is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published
 by the Free Software Foundation; either version 3, or (at your
 option) any later version.

 GNUnet is distributed in the hope that it will be useful, but
 WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with GNUnet; see the file COPYING.  If not, write to the
 Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
 Boston, MA 02110-1301, USA.
 */

package org.gnunet.dht;

import com.google.common.base.Charsets;
import org.gnunet.dht.messages.*;
import org.gnunet.mq.Envelope;
import org.gnunet.requests.MatchingRequestContainer;
import org.gnunet.requests.Request;
import org.gnunet.requests.RequestIdentifier;
import org.gnunet.requests.SequentialRequestContainer;
import org.gnunet.util.*;
import org.gnunet.util.getopt.Argument;
import org.gnunet.util.getopt.ArgumentAction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.EnumSet;
import java.util.List;
import java.util.Set;

/**
 * API for the GNUnet dht service.
 * <p/>
 * Stores data under a key, distributed across the network.
 * <p/>
 */
public class DistributedHashTable {
    private static final Logger logger = LoggerFactory
            .getLogger(DistributedHashTable.class);

    private Client client;

    /**
     * next UID used on get/monitor requests, incremented after each use.
     */
    private long nextUID = 1;

    private MatchingRequestContainer<Long, PutRequest> putRequests;
    private MatchingRequestContainer<Long, GetRequest> getRequests;
    private SequentialRequestContainer<MonitorRequest> monitorRequests;

    private class PutRequest extends Request {
        public byte[] data;
        public HashCode key;
        public int replicationLevel;
        public AbsoluteTime expiration;
        public int type;
        public Continuation cont;
        public long uid;
        private int options;

        public PutRequest() {
            this.uid = nextUID++;
        }

        @Override
        public Envelope assembleRequest() {
            final ClientPutMessage cpm = new ClientPutMessage();
            cpm.data = data;
            cpm.hash = key;
            cpm.desiredReplicationLevel = replicationLevel;
            cpm.expiration = expiration.asMessage();
            cpm.type = type;
            cpm.uid = uid;
            cpm.options = options;
            return new Envelope(cpm);
        }

        public void onCancel() {
        }
    }


    private class GetRequest extends Request {
        public long uid;
        public HashCode key;
        public ResultCallback cb;
        public int type;
        public int replication;
        public byte[] xquery;
        public int options;

        public GetRequest() {
            uid = DistributedHashTable.this.nextUID++;
        }

        @Override
        public Envelope assembleRequest() {
            ClientGetMessage gm = new ClientGetMessage();
            gm.desiredReplicationLevel = replication;
            gm.type = type;
            gm.xquery = xquery == null ? new byte[0] : xquery;
            gm.key = key;
            gm.uniqueId = uid;
            gm.options = options;
            return new Envelope(gm);
        }
        public void onCancel() {

        }
    }

    private class MonitorRequest extends Request {
        public int blockType;
        public HashCode key;
        public MonitorGetHandler getHandler;
        public MonitorGetResponseHandler getResponseHandler;
        public MonitorPutHandler putHandler;

        @Override
        public Envelope assembleRequest() {
            MonitorStartStop mss = new MonitorStartStop();
            if (key != null) {
                mss.filterKey = 1;
                mss.key = key;
            } else {
                mss.key = new HashCode();
            }
            if (getHandler != null) {
                mss.get = 1;
            }
            if (getResponseHandler != null) {
                mss.getResp = 1;
            }
            if (putHandler != null) {
                mss.put = 1;
            }
            mss.type = blockType;
            return new Envelope(mss);
        }

        public void onCancel() {
            // todo: use priority requests
            MonitorRequest cancelRequest = new MonitorRequest();
            cancelRequest.getHandler = null;
            cancelRequest.getResponseHandler = null;
            cancelRequest.putHandler = null;
            monitorRequests.addRequest(cancelRequest);

            monitorRequests.addRequest(cancelRequest);
        }
    }

    private class DHTMessageReceiver extends RunaboutMessageReceiver {
        public void visit(ClientPutConfirmationMessage pcm) {
            PutRequest thePutRequest = putRequests.getAndRetireRequest(pcm.uid);
            if (thePutRequest == null) {
                logger.warn("getRequestIdentifier UID not found");
                return;
            }
            if (thePutRequest.cont != null) {
                thePutRequest.cont.cont(true);
            }
        }

        public void visit(ClientResultMessage rm) {
            GetRequest theGetRequest = getRequests.getAndRetireRequest(rm.uid);
            if (theGetRequest == null) {
                logger.warn("getRequestIdentifier UID not found");
                return;
            }
            theGetRequest.cb.handleResult(AbsoluteTime.fromNetwork(rm.expiration), rm.key, null, null,
                    BlockType.TEST,
                    rm.data);
        }

        public void visit(MonitorGetMessage monitorGetMessage) {
            for (RequestIdentifier<MonitorRequest> monitorRequest : monitorRequests.iter()) {
                boolean typeOk = (monitorGetMessage.type == BlockType.ANY.val)
                        || (monitorGetMessage.type == monitorRequest.getRequest().blockType);
                boolean keyOk = monitorGetMessage.key.isAllZero()
                        || monitorGetMessage.key.equals(monitorRequest.getRequest().key);

                if (keyOk && typeOk && monitorRequest.getRequest().getHandler != null) {
                    monitorRequest.getRequest().getHandler.onGet(monitorGetMessage.options, monitorGetMessage.type,
                            monitorGetMessage.hopCount, monitorGetMessage.desiredReplicationLevel, monitorGetMessage.getPath,
                            monitorGetMessage.key);
                }
            }
        }

        public void visit(MonitorGetRespMessage monitorGetRespMessage) {
            for (RequestIdentifier<MonitorRequest> monitorRequest : monitorRequests.iter()) {
                boolean typeOk = (monitorGetRespMessage.type == BlockType.ANY.val)
                        || (monitorGetRespMessage.type == monitorRequest.getRequest().blockType);
                boolean keyOk = monitorGetRespMessage.key.isAllZero()
                        || monitorGetRespMessage.key.equals(monitorRequest.getRequest().key);

                if (keyOk && typeOk && monitorRequest.getRequest().getResponseHandler != null) {
                    monitorRequest.getRequest().getResponseHandler.onGetResponse(
                            monitorGetRespMessage.type,
                            monitorGetRespMessage.getPath,
                            monitorGetRespMessage.putPath,
                            monitorGetRespMessage.expiration,
                            monitorGetRespMessage.key,
                            monitorGetRespMessage.data);
                }
            }

        }

        public void visit(MonitorPutMessage monitorPutMessage) {
            for (RequestIdentifier<MonitorRequest> monitorRequest : monitorRequests.iter()) {
                boolean typeOk = (monitorPutMessage.type == BlockType.ANY.val)
                        || (monitorPutMessage.type == monitorRequest.getRequest().blockType);
                boolean keyOk = monitorPutMessage.key.isAllZero()
                        || monitorPutMessage.key.equals(monitorRequest.getRequest().key);

                if (keyOk && typeOk && monitorRequest.getRequest().putHandler != null) {
                    monitorRequest.getRequest().putHandler.onPut(monitorPutMessage.options, monitorPutMessage.type,
                            monitorPutMessage.hopCount, monitorPutMessage.expirationTime,
                            monitorPutMessage.putPath, monitorPutMessage.key, monitorPutMessage.data);
                }
            }
        }

        @Override
        public void handleError() {
            // FIXME
        }
    }


    /**
     * Create a connection with the DHT service.
     *
     * @param cfg the configuration to use
     */
    public DistributedHashTable(Configuration cfg) {
        client = new Client("dht", cfg);
        client.installReceiver(new DHTMessageReceiver());
        putRequests = new MatchingRequestContainer<Long, PutRequest>(client);
        getRequests = new MatchingRequestContainer<Long, GetRequest>(client);
        monitorRequests = new SequentialRequestContainer<MonitorRequest>(client);
    }

    /**
     * Put data into the dht.
     *
     * @param key              key key to store the data under
     * @param data             data data to store
     * @param replicationLevel how many peers should store this value
     * @param routeOptions     additional options
     * @param type             type of the data to store
     * @param expiration       how long should the value be stored? TODO: what is the maximum?
     * @param cont             called after the put operation failed or succeeded
     */
    public void put(HashCode key, byte[] data, int replicationLevel, Set<RouteOption> routeOptions,
                    int type, AbsoluteTime expiration, final Continuation cont) {
        PutRequest pr = new PutRequest();
        pr.key = key;
        pr.data = data;
        pr.replicationLevel = replicationLevel;
        pr.expiration = expiration;
        pr.type = type;
        pr.cont = cont;
        pr.options = 0;
        for (RouteOption routeOption : routeOptions) {
            pr.options |= routeOption.val;
        }

        putRequests.addRequest(pr.uid, pr);
    }


    /**
     * Request results from the DHT.
     *
     * @param timeout      timeout for the getRequestIdentifier
     * @param type         which type of data do we want to query for? (the DHT does not support TYPE_ANY)
     * @param key          the key we want to query
     * @param replication  how many peers do we want to ask?
     * @param routeOptions extra routing options, null for default
     * @param xquery       extra query parameters, defaults to null
     * @param cb           the callback object for results or failure indication
     * @return a handle to onCancel the getRequestIdentifier
     */
    public Cancelable startGet(RelativeTime timeout, int type, HashCode key,
                               int replication, EnumSet<RouteOption> routeOptions,
                               byte[] xquery, ResultCallback cb) {

        final GetRequest getRequest = new GetRequest();
        getRequest.key = key;
        getRequest.cb = cb;
        getRequest.type = type;
        getRequest.replication = type;
        getRequest.xquery = xquery;
        getRequest.replication = replication;
        getRequest.options = 0;
        for (RouteOption routeOption : routeOptions) {
            getRequest.options |= routeOption.val;
        }

        return getRequests.addRequest(getRequest.uid, getRequest);
    }

    /**
     * Start monitoring certain types of requests.
     *
     * @param blockType block type of requests we're interested in
     * @param key key we're interested in
     * @param getHandler listener for get requests
     * @param getResponseHandler listener for get responses
     * @param putHandler listener for put requests
     * @return a handle to cancel the monitoring
     */
    public Cancelable startMonitor(int blockType, HashCode key, MonitorGetHandler getHandler,
                                   MonitorGetResponseHandler getResponseHandler,
                                   MonitorPutHandler putHandler) {
        MonitorRequest monitorRequest = new MonitorRequest();
        monitorRequest.blockType = blockType;
        monitorRequest.key = key;
        monitorRequest.getHandler = getHandler;
        monitorRequest.getResponseHandler = getResponseHandler;
        monitorRequest.putHandler = putHandler;

        return monitorRequests.addRequest(monitorRequest);
    }


    /**
     * Destroy the connection to the service.
     */
    public void destroy() {
        // there's nothing to sync, just destroy!
        client.disconnect();
    }

    public static void main(String[] args) {
        new Program() {
            @Argument(action = ArgumentAction.SET,
                    shortname = "p",
                    longname = "put",
                    description = "set a value in the DHT; default is get")
            boolean modePut = false;

            @Argument(action = ArgumentAction.SET,
                    shortname = "m",
                    longname = "monitor",
                    description = "monitor requests going to the local DHT")
            boolean monitor = false;


            @Argument(action = ArgumentAction.STORE_STRING,
                    shortname = "d",
                    longname = "data",
                    description = "data (only used with --put)")
            String data = null;

            @Argument(action = ArgumentAction.STORE_STRING,
                    shortname = "k",
                    longname = "key",
                    description = "key used for the operation")
            String key = null;


            // todo: implement the following options
            /*
            @Argument(action = ArgumentAction.STORE_STRING,
                    shortname = "t",
                    longname = "type",
                    description = "type of data used in this operation")
            String type = null;

            @Argument(action = ArgumentAction.STORE_STRING,
                    shortname = "e",
                    longname = "expire",
                    description = "expiration (ony use with --put)")
            String expiration = null;
            */


            @Argument(action = ArgumentAction.STORE_NUMBER,
                    shortname = "r",
                    longname = "replication",
                    description = "desired replication (only used with --put)")
            int replication = 5;


            public void run() {
                if (modePut) {

                    if (key == null) {
                        System.out.println("key required");
                        return;
                    }

                    if (data == null) {
                        System.out.println("data required on put");
                        return;
                    }
                    final DistributedHashTable dht = new DistributedHashTable(cfg);

                    dht.put(new HashCode(key), data.getBytes(), replication, EnumSet.of(RouteOption.NONE),
                            BlockType.TEST.val, AbsoluteTime.now().add(RelativeTime.HOUR),
                            new Continuation() {
                                @Override
                                public void cont(boolean success) {
                                    if (success) {
                                        System.out.println("put getRequestIdentifier sent");
                                    } else {
                                        System.out.println("error");
                                    }
                                    dht.destroy();
                                }
                            });
                } else if (monitor) {
                    final DistributedHashTable dht = new DistributedHashTable(cfg);
                    dht.startMonitor(BlockType.TEST.val, null,
                            new MonitorGetHandler() {
                                @Override
                                public void onGet(int options, int type, int hopCount,
                                                  int desiredReplicationLevel, PeerIdentity[] getPath, HashCode key) {
                                    System.out.println("get monitored");
                                }
                            },
                            new MonitorGetResponseHandler() {
                                @Override
                                public void onGetResponse(int type, PeerIdentity[] getPath, PeerIdentity[] putPath,
                                                          AbsoluteTimeMessage expiration, HashCode key, byte[] data) {
                                    System.out.println("get response monitored");
                                }
                            },
                            new MonitorPutHandler() {
                                @Override
                                public void onPut(int options, int type, int hop_count, AbsoluteTimeMessage
                                        expirationTime, PeerIdentity[] putPath, HashCode key, byte[] data) {
                                    System.out.println("put monitored");
                                }
                            });
                } else { // get
                    if (key == null) {
                        System.out.println("key required");
                        return;
                    }
                    if (data != null) {
                        System.out.println("get does not take data as an option");
                        return;
                    }

                    final DistributedHashTable dht = new DistributedHashTable(cfg);

                    dht.startGet(RelativeTime.SECOND, BlockType.TEST.val, new HashCode(key), replication, null,
                            new byte[0], new ResultCallback() {
                        @Override
                        public void handleResult(AbsoluteTime expiration, HashCode key, List<PeerIdentity>
                                getPath, List<PeerIdentity> putPath, BlockType type, byte[] data) {
                            System.out.println("got result:");
                            System.out.println(new String(data, Charsets.UTF_8));
                        }
                    });
                }
            }
        }.start(args);
    }
}