test-serf.go

package main

import (
    "fmt"
    "log"
    "os"
    "sync"

    "github.com/hashicorp/memberlist"
    "github.com/hashicorp/serf/serf"
)

const (
    UDPBufferSize = 2000
    baseAddress   = "127.0.0.1"
)

var (
    nodeAddress string
    NODE_ID     = os.Getenv("NODE_ID")
    knownNodes  = []string{fmt.Sprintf("%s:%s", baseAddress, NODE_ID)}
    broadcasts  *memberlist.TransmitLimitedQueue
)

type broadcast struct {
    msg    []byte
    notify chan<- struct{}
}

type delegate struct{}

type update struct {
    Action string // add, del
    Data   map[string]string
}

func (b *broadcast) Invalidates(other memberlist.Broadcast) bool {
    return false
}

func (b *broadcast) Message() []byte {
    return b.msg
}

func (b *broadcast) Finished() {
    if b.notify != nil {
        close(b.notify)
    }
}

func (d *delegate) NodeMeta(limit int) []byte {
    return []byte{}
}

func (d *delegate) NotifyMsg(b []byte) {
    if len(b) == 0 {
        return
    }
    fmt.Println("Received message")
}

func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
    return broadcasts.GetBroadcasts(overhead, limit)
}

func (d *delegate) LocalState(join bool) []byte {
    fmt.Println("push pulllll")
    return []byte{}
}

func (d *delegate) MergeRemoteState(buf []byte, join bool) {
}

func configServer() error {
    hostname, _ := os.Hostname()
    memberlistConfig := memberlist.DefaultLocalConfig()
    memberlistConfig.Name = nodeAddress
    memberlistConfig.BindAddr = baseAddress
    if _, err := fmt.Sscanf(NODE_ID, "%d", &memberlistConfig.BindPort); err != nil {
        return err
    }
    memberlistConfig.PushPullInterval = 0
    memberlistConfig.UDPBufferSize = UDPBufferSize
    memberlistConfig.Delegate = &delegate{}
    memberlistConfig.Name = nodeAddress

    serfConfig := serf.DefaultConfig()
    serfConfig.NodeName = nodeAddress
    // serfConfig.EventCh = serfEvents
    serfConfig.MemberlistConfig = memberlistConfig

    m, err := serf.Create(serfConfig)

    if err != nil {
        return err
    }
    if nodeAddress != knownNodes[0] {
        _, err := m.Join(knownNodes, false)
        if err != nil {
            return err
        }
    }
    broadcasts = &memberlist.TransmitLimitedQueue{
        NumNodes: func() int {
            return m.NumNodes()
        },
        RetransmitMult: 1,
    }
    node := m.LocalMember()
    fmt.Printf("Local member %s:%d\n", node.Addr, node.Port)
    return nil
}

func main() {
    var err error
    var wg sync.WaitGroup

    nodeAddress = fmt.Sprintf("%s:%s", baseAddress, NODE_ID)

    wg.Add(1)
    go func() {
        err = configServer()
        if err != nil {
            log.Panic(err)
        }
        wg.Done()
    }()
    wg.Wait()

}