func (srv *clonesrv_t) was_pending(kvmsg *kvmsg.Kvmsg) bool {
	uuid1, _ := kvmsg.GetUuid()
	for i, msg := range srv.pending {
		if uuid2, _ := msg.GetUuid(); uuid1 == uuid2 {
			srv.pending = append(srv.pending[:i], srv.pending[i+1:]...)
			return true
		}
	}
	return false
}
The collector is more complex than in the clonesrv5 example since how
process updates depends on whether we're active or passive. The active
applies them immediately to its kvmap, whereas the passive queues them
as pending:
If message was already on pending list, remove it and return TRUE,
else return FALSE.
package main
import (
	zmq "github.com/pebbe/zmq4"
	"github.com/pebbe/zmq4/examples/bstar"
	"github.com/pebbe/zmq4/examples/kvmsg"
	"errors"
	"fmt"
	"log"
	"os"
	"strconv"
	"strings"
	"time"
)
type clonesrv_t struct {
	kvmap      map[string]*kvmsg.Kvmsg  
	kvmap_init bool
	bstar      *bstar.Bstar    
	sequence   int64           
	port       int             
	peer       int             
	publisher  *zmq.Socket     
	collector  *zmq.Socket     
	subscriber *zmq.Socket     
	pending    []*kvmsg.Kvmsg  
	primary    bool            
	active     bool            
	passive    bool            
}
func main() {
	var err error
	srv := &clonesrv_t{}
	if len(os.Args) == 2 && os.Args[1] == "-p" {
		log.Println("I: primary active, waiting for backup (passive)")
		srv.bstar, err = bstar.New(bstar.PRIMARY, "tcp://*:5003", "tcp://localhost:5004")
		if err != nil {
			log.Println(err)
			return
		}
		srv.bstar.Voter("tcp://*:5556", zmq.ROUTER, func(soc *zmq.Socket) error { return snapshots(soc, srv) })
		srv.port = 5556
		srv.peer = 5566
		srv.primary = true
	} else if len(os.Args) == 2 && os.Args[1] == "-b" {
		log.Println("I: backup passive, waiting for primary (active)")
		srv.bstar, err = bstar.New(bstar.BACKUP, "tcp://*:5004", "tcp://localhost:5003")
		srv.bstar.Voter("tcp://*:5566", zmq.ROUTER, func(soc *zmq.Socket) error { return snapshots(soc, srv) })
		srv.port = 5566
		srv.peer = 5556
		srv.primary = false
	} else {
		fmt.Println("Usage: clonesrv6 { -p | -b }")
		return
	}
	if srv.primary {
		srv.kvmap = make(map[string]*kvmsg.Kvmsg, 0)
		srv.kvmap_init = true
	}
	srv.pending = make([]*kvmsg.Kvmsg, 0)
	srv.bstar.SetVerbose(true)
	srv.publisher, _ = zmq.NewSocket(zmq.PUB)
	srv.collector, _ = zmq.NewSocket(zmq.SUB)
	srv.collector.SetSubscribe("")
	srv.publisher.Bind(fmt.Sprint("tcp://*:", srv.port+1))
	srv.collector.Bind(fmt.Sprint("tcp://*:", srv.port+2))
	srv.subscriber, _ = zmq.NewSocket(zmq.SUB)
	srv.subscriber.SetSubscribe("")
	srv.subscriber.Connect(fmt.Sprint("tcp://localhost:", srv.peer+1))
	srv.bstar.NewActive(func() error { return new_active(srv) })
	srv.bstar.NewPassive(func() error { return new_passive(srv) })
	srv.bstar.Reactor.AddSocket(srv.collector, zmq.POLLIN,
		func(e zmq.State) error { return collector(srv) })
	srv.bstar.Reactor.AddChannelTime(time.Tick(1000*time.Millisecond), 1,
		func(i interface{}) error {
			if e := flush_ttl(srv); e != nil {
				return e
			}
			return send_hugz(srv)
		})
	err = srv.bstar.Start()
	log.Println(err)
}
func snapshots(socket *zmq.Socket, srv *clonesrv_t) (err error) {
	msg, err := socket.RecvMessage(0)
	if err != nil {
		return
	}
	identity := msg[0]
	request := msg[1]
	if request != "ICANHAZ?" {
		err = errors.New("E: bad request, aborting")
		return
	}
	subtree := msg[2]
	for _, kvmsg := range srv.kvmap {
		if key, _ := kvmsg.GetKey(); strings.HasPrefix(key, subtree) {
			socket.Send(identity, zmq.SNDMORE)
			kvmsg.Send(socket)
		}
	}
	log.Println("I: sending shapshot =", srv.sequence)
	socket.Send(identity, zmq.SNDMORE)
	kvmsg := kvmsg.NewKvmsg(srv.sequence)
	kvmsg.SetKey("KTHXBAI")
	kvmsg.SetBody(subtree)
	kvmsg.Send(socket)
	return
}