iwanbk/bcache

package bcache

import (
    "log"
    "sync"

    "github.com/weaveworks/mesh"
)

const (
    defaultNumEntries = 5
)

// message defines gossip message used for communication between peers
// TODO: use protobuf for encoding
type message struct {
    mux     sync.RWMutex
    PeerID  mesh.PeerName
    Entries map[string]entry
}

// entry is a single key value entry
type entry struct {
    Val     string
    Expired int64
    Deleted int64
}

func newMessage(peerID mesh.PeerName, numEntries int) *message {
    if numEntries == 0 {
        numEntries = defaultNumEntries
    }
    return newMessageFromEntries(peerID, make(map[string]entry, numEntries))
}

func newMessageFromEntries(peerID mesh.PeerName, entries map[string]entry) *message {
    newEntries := make(map[string]entry, len(entries))
    for k, v := range entries {
        newEntries[k] = v
    }
    return &message{
        PeerID:  peerID,
        Entries: newEntries,
    }
}

func newMessageFromBuf(b []byte) (*message, error) {
    var m message
    err := unmarshal(b, &m)
    return &m, err
}

func (m *message) add(key, val string, expired, deleted int64) {
    m.mux.Lock()
    m.Entries[key] = entry{
        Val:     val,
        Expired: expired,
        Deleted: deleted,
    }
    m.mux.Unlock()
}

// Encode implements mesh.GossipData.Encode
// TODO: split the encoding by X number of keys
func (m *message) Encode() [][]byte {
    m.mux.RLock()
    defer m.mux.RUnlock()

    b, err := marshal(m)
    if err != nil {
        log.Printf("failed to encode message: %v", err)
    }
    return [][]byte{b}
}

// Merge implements mesh.GossipData.Merge
func (m *message) Merge(other mesh.GossipData) (complete mesh.GossipData) {
    return m.mergeComplete(other.(*message))
}

func (m *message) mergeComplete(other *message) mesh.GossipData {
    m.mux.Lock()
    defer m.mux.Unlock()

    for k, v := range other.Entries {
        existing, ok := m.Entries[k]

        // merge
        // - the key not exists in
        // - has less expiration time
        if !ok || existing.Expired < v.Expired {
            m.Entries[k] = v
        }
    }
    return newMessageFromEntries(m.PeerID, m.Entries)
}