Basic test of Pastry

pastry-test.go

package main

import (
	"crypto/rand"
	"flag"
	"fmt"
	"secondbit.org/pastry"
	"time"
)

type Callback struct {
	Client pastry.NodeID
}

func (c *Callback) OnError(err error) {
	panic(c.Client.String() + ": " + err.Error())
}

func (c *Callback) OnDeliver(msg pastry.Message) {
	fmt.Printf("%s: Delivering %s\n", c.Client, msg.Key)
}

func (c *Callback) OnForward(msg *pastry.Message, next pastry.NodeID) bool {
	fmt.Printf("%s: Forwarding %s to %s\n", c.Client, msg, next)
	return true
}

func (c *Callback) OnNewLeaves(leaves []*pastry.Node) {
	fmt.Printf("%s: Leafset updated.\n", c.Client)
	leaves_string := ""
	for _, leaf := range leaves {
		leaves_string = leaves_string + leaf.ID.String() + ", "
	}
	fmt.Printf("%s: Leaves: \033[4;33m%s\033[0m\n", c.Client, leaves_string)
}

func (c *Callback) OnNodeJoin(node pastry.Node) {
	fmt.Printf("%s: Node joined: %s\n", c.Client, node.ID)
}

func (c *Callback) OnNodeExit(node pastry.Node) {
	fmt.Printf("%s: Node left: %s\n", c.Client, node.ID)
}

func (c *Callback) OnHeartbeat(node pastry.Node) {
	fmt.Printf("%s: Received heartbeat from %s\n", c.Client, node.ID)
}

func makeCluster(port int, credentials string) *pastry.Cluster {
	b := make([]byte, 16)
	_, err := rand.Read(b)
	if err != nil {
		panic(err.Error())
	}
	id, err := pastry.NodeIDFromBytes(b)
	if err != nil {
		panic(err.Error())
	}
	node := pastry.NewNode(id, "127.0.0.1", "127.0.0.1", "macbookpro", port)
	cluster := pastry.NewCluster(node, pastry.Passphrase(credentials))
	cluster.SetHeartbeatFrequency(10)
	cluster.SetLogLevel(pastry.LogLevelWarn)
	return cluster
}

func main() {
	var port int
	flag.IntVar(&port, "port", 8080, "The port to listen on.")
	var target, credentials string
	var targetPort int
	flag.StringVar(&target, "target", "127.0.0.1", "The target to connect to.")
	flag.IntVar(&targetPort, "targetPort", -1, "The port the target to connect to is listening on.")
	flag.StringVar(&credentials, "credentials", "test credentials", "The credentials to use when joining the cluster.")
	flag.Parse()
	c := makeCluster(port, credentials)
	c.RegisterCallback(&Callback{Client: c.ID()})
	go func(){
		defer c.Stop()
		err := c.Listen()
		if err != nil {
			panic(err.Error())
		}
	}()
	time.Sleep(1 * time.Second)
	if targetPort != -1 {
		err := c.Join(target, targetPort)
		if err != nil {
			panic(err)
		}
	}
	go func() {
		for {
			time.Sleep(20 * time.Second)
			b := make([]byte, 16)
			_, err := rand.Read(b)
			if err != nil {
				panic(err.Error())
			}
			key, err := pastry.NodeIDFromBytes(b)
			if err != nil {
				panic(err.Error())
			}
			msg := c.NewMessage(byte(16), key, []byte("Testing this out."))
			err = c.Send(msg)
			if err != nil {
				panic(err.Error())
			}
		}
	}()
	select {}
}