iwanbk/bcache

package bcache

import (
    "github.com/weaveworks/mesh"
)

type peer struct {
    cc       *cache
    name     mesh.PeerName
    send     mesh.Gossip
    actionCh chan func()
    quitCh   chan struct{}
    logger   Logger
}

func newPeer(name mesh.PeerName, maxKeys int, logger Logger) (*peer, error) {
    cc, err := newCache(maxKeys)
    if err != nil {
        return nil, err
    }

    p := &peer{
        cc:       cc,
        name:     name,
        send:     nil, // must be registered
        actionCh: make(chan func()),
        quitCh:   make(chan struct{}),
        logger:   logger,
    }
    go p.loop()
    return p, nil
}

// register the result of a mesh.Router.NewGossip.
func (p *peer) register(send mesh.Gossip) {
    p.actionCh <- func() {
        p.send = send
    }
}

// Gossip implements mesh.Gossiper.Gossip
func (p *peer) Gossip() mesh.GossipData {
    return p.cc.Messages()
}

// OnGossip merges received data into state and returns "everything new
// I've just learnt", or nil if nothing in the received data was new.
//
// It implements mesh.Gossiper.OnGossip
func (p *peer) OnGossip(buf []byte) (delta mesh.GossipData, err error) {
    msg, err := newMessageFromBuf(buf)
    if err != nil {
        return
    }

    var deltaMsg *message

    delta = p.cc.mergeNew(msg)
    if delta != nil {
        deltaMsg = delta.(*message)
    }

    p.logger.Debugf("[%d]OnGossip %v => delta %v", p.name, msg, deltaMsg)
    return
}

// OnGossipBroadcast merges received data into state and returns a
// representation of the received data (typically a delta) for further
// propagation.
//
// It implements mesh.Gossiper.OnGossipBroadcast
func (p *peer) OnGossipBroadcast(src mesh.PeerName, update []byte) (received mesh.GossipData, err error) {
    if src == p.name { // message from ourself, is it possible?
        return
    }
    msg, err := newMessageFromBuf(update)
    if err != nil {
        return
    }

    var recvMsg *message

    received = p.cc.mergeDelta(msg)
    if received != nil {
        recvMsg = received.(*message)
    }
    p.logger.Debugf("[%d]OnGossipBroadcast %v => delta %v", p.name, msg, recvMsg)
    return

}

func (p *peer) OnGossipUnicast(src mesh.PeerName, update []byte) error {
    msg, err := newMessageFromBuf(update)
    if err != nil {
        return err
    }
    p.cc.mergeComplete(msg)
    return nil
}

func (p *peer) Set(key, val string, expiredTimestamp int64) {
    c := make(chan struct{})

    p.actionCh <- func() {
        defer close(c)

        // set our cache
        p.cc.Set(key, val, expiredTimestamp, 0)

        // construct & send the message
        m := newMessage(p.name, 1)
        m.add(key, val, expiredTimestamp, 0)

        p.broadcast(m)
    }

    <-c // wait for it to be finished
}

func (p *peer) Delete(key string, deleteTimestamp int64) bool {
    var (
        c     = make(chan struct{})
        exist bool
    )

    p.actionCh <- func() {
        defer close(c)

        // delete from our cache
        val, expired, exist := p.cc.Delete(key, deleteTimestamp)
        if !exist {
            return
        }

        // construct & send the message
        m := newMessage(p.name, 1)
        m.add(key, val, expired, deleteTimestamp)

        p.broadcast(m)
    }

    <-c // wait for it to be finished
    return exist
}

func (p *peer) Get(key string) (string, bool) {
    return p.cc.Get(key)
}

func (p *peer) loop() {
    for {
        select {
        case f := <-p.actionCh:
            f()
        case <-p.quitCh:
            return
        }
    }
}

func (p *peer) broadcast(msg *message) {
    if p.send == nil {
        return
    }
    p.send.GossipBroadcast(msg)

}