syncer.go

package mirror

import (
    "github.com/coreos/etcd/clientv3"
    "golang.org/x/net/context"
)

const (
    batchLimit = 1000
)

// Syncer syncs with the key-value state of an etcd cluster.
type Syncer interface {
    // SyncBase syncs the base state of the key-value state.
    // The key-value state are sent through the returned chan.
    SyncBase(ctx context.Context) (<-chan clientv3.GetResponse, chan error)
    // SyncUpdates syncs the updates of the key-value state.
    // The update events are sent through the returned chan.
    SyncUpdates(ctx context.Context) clientv3.WatchChan
}

// NewSyncer creates a Syncer.
func NewSyncer(c *clientv3.Client, prefix string, rev int64) Syncer {
    return &syncer{c: c, prefix: prefix, rev: rev}
}

type syncer struct {
    c      *clientv3.Client
    rev    int64
    prefix string
}

func (s *syncer) SyncBase(ctx context.Context) (<-chan clientv3.GetResponse, chan error) {
    respchan := make(chan clientv3.GetResponse, 1024)
    errchan := make(chan error, 1)

    // if rev is not specified, we will choose the most recent revision.
    if s.rev == 0 {
        resp, err := s.c.Get(ctx, "foo")
        if err != nil {
            errchan <- err
            close(respchan)
            close(errchan)
            return respchan, errchan
        }
        s.rev = resp.Header.Revision
    }

    go func() {
        defer close(respchan)
        defer close(errchan)

        var key string

        opts := []clientv3.OpOption{clientv3.WithLimit(batchLimit), clientv3.WithRev(s.rev)}

        if len(s.prefix) == 0 {
            // If len(s.prefix) == 0, we will sync the entire key-value space.
            // We then range from the smallest key (0x00) to the end.
            opts = append(opts, clientv3.WithFromKey())
            key = "\x00"
        } else {
            // If len(s.prefix) != 0, we will sync key-value space with given prefix.
            // We then range from the prefix to the next prefix if exists. Or we will
            // range from the prefix to the end if the next prefix does not exists.
            opts = append(opts, clientv3.WithRange(clientv3.GetPrefixRangeEnd(s.prefix)))
            key = s.prefix
        }

        for {
            resp, err := s.c.Get(ctx, key, opts...)
            if err != nil {
                errchan <- err
                return
            }

            respchan <- (clientv3.GetResponse)(*resp)

            if !resp.More {
                return
            }
            // move to next key
            key = string(append(resp.Kvs[len(resp.Kvs)-1].Key, 0))
        }
    }()

    return respchan, errchan
}

func (s *syncer) SyncUpdates(ctx context.Context) clientv3.WatchChan {
    if s.rev == 0 {
        panic("unexpected revision = 0. Calling SyncUpdates before SyncBase finishes?")
    }
    return s.c.Watch(ctx, s.prefix, clientv3.WithPrefix(), clientv3.WithRev(s.rev+1))
}

上一篇:Dictionary 的使用


下一篇:20165230 预备作业3 Linux安装及学习