// Bamboo<http://bamboo-dht.org/> sample code
//
// $ export CLASSPATH=$BAMBOO_DIR/src/:`echo $BAMBOO_DIR/lib/*.jar|sed -e 's/ /:/g'`
// $ javac Dht.java
//
// append the following line to node.cfg:
//  <My>
//    class my.Dht
//    <initargs>
//        client_stage_name GatewayClient
//    </initargs>
//  </My>
//  <GatewayClient>
//    class bamboo.dht.GatewayClient
//    <initargs>
//        gateway                 localhost:3632
//    </initargs>
//  </GatewayClient>
//
// $ ${BAMBOO_DIR}/bin/run-java bamboo.lss.DustDevil node.cfg

package my;

import bamboo.util.StandardStage;
import bamboo.lss.ASyncCore;
import bamboo.router.Router;
import seda.sandStorm.api.ConfigDataIF;
import bamboo.dht.GatewayClient;
import bamboo.dht.bamboo_put_args;
import bamboo.dht.bamboo_get_args;
import bamboo.dht.bamboo_key;
import bamboo.dht.bamboo_value;
import bamboo.dht.bamboo_stat;
import bamboo.dht.bamboo_get_res;
import bamboo.dht.bamboo_placemark;
import java.security.MessageDigest;
import static bamboo.util.Curry.*;

public class Dht extends StandardStage {

    protected MessageDigest digest;
    protected GatewayClient client;
    
    public void init(ConfigDataIF config) throws Exception {
	super.init(config);

        try {
            digest = MessageDigest.getInstance("SHA");
        } catch (Exception e) {
            assert false;
        }
        String client_stg_name = config_get_string(config, "client_stage_name");
        client = (GatewayClient)lookup_stage(config, client_stg_name);

        acore.register_timer(1000, do_put_cb, null);
    }
        
    public ASyncCore.TimerCB do_put_cb = new ASyncCore.TimerCB() {
            public void timer_cb(Object user_data) {
                String key = "p2p";
                String val = "/ascii/unixmagazine/p2p.pdf";
                
                bamboo_put_args put_args = new bamboo_put_args();
                put_args.application = "my-test";
                put_args.key = new bamboo_key();
                put_args.key.value = digest.digest(key.getBytes());
                put_args.value = new bamboo_value();
                put_args.value.value = val.getBytes();
                put_args.ttl_sec = 300;// [sec]

                client.put(put_args, curry(put_done_cb, put_args, null));
            }
        };

    public Thunk3<bamboo_put_args, Object, Integer> put_done_cb = 
        new Thunk3<bamboo_put_args, Object, Integer>() {
            public void run(bamboo_put_args put_args, Object user_ctx, Integer put_res) {
                if (put_res.intValue() == bamboo_stat.BAMBOO_OK) {
                    System.out.println("DHT put success");
                    
                    acore.register_timer(1000, do_get_cb, null);
                } else {
                    System.out.println("DHT put failure. reason=" + put_res);
                }
            }
        };

    
    public ASyncCore.TimerCB do_get_cb = new ASyncCore.TimerCB() {
            public void timer_cb(Object user_data) {
                String key = "p2p";
                
                bamboo_get_args get_args = new bamboo_get_args();
                get_args.application = "my-test";
                get_args.key = new bamboo_key();
                get_args.key.value = digest.digest(key.getBytes());
                get_args.maxvals = 1;
                get_args.placemark = new bamboo_placemark();
                get_args.placemark.value = new byte[] {};

                client.get(get_args, curry(get_done_cb, null));
            }
        };

    public Thunk2<Object, bamboo_get_res> get_done_cb = 
        new Thunk2<Object, bamboo_get_res>() {
            public void run(Object user_ctx, bamboo_get_res get_res) {
                if (get_res.values.length > 0) {
                    System.out.println("DHT get value is " + new String(get_res.values[0].value));
                } else {
                    System.out.println("DHT not found");
                }
            }
        };
}