mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
client: WaitIndex -> AfterIndex
This commit is contained in:
parent
a834f297f9
commit
1c03df62a5
@ -82,14 +82,14 @@ type KeysAPI interface {
|
||||
}
|
||||
|
||||
type WatcherOptions struct {
|
||||
// WaitIndex defines the index after-which the Watcher should
|
||||
// AfterIndex defines the index after-which the Watcher should
|
||||
// start emitting events. For example, if a value of 5 is
|
||||
// provided, the first event will have an index >= 6.
|
||||
//
|
||||
// Setting WaitIndex to 0 (default) means that the Watcher
|
||||
// Setting AfterIndex to 0 (default) means that the Watcher
|
||||
// should start watching for events starting at the current
|
||||
// index, whatever that may be.
|
||||
WaitIndex uint64
|
||||
AfterIndex uint64
|
||||
|
||||
// Recursive specifices whether or not the Watcher should emit
|
||||
// events that occur in children of the given keyspace. If set
|
||||
@ -276,7 +276,7 @@ func (k *httpKeysAPI) Watcher(key string, opts *WatcherOptions) Watcher {
|
||||
}
|
||||
|
||||
if opts != nil {
|
||||
act.WaitIndex = opts.WaitIndex
|
||||
act.AfterIndex = opts.AfterIndex
|
||||
act.Recursive = opts.Recursive
|
||||
}
|
||||
|
||||
@ -302,7 +302,7 @@ func (hw *httpWatcher) Next(ctx context.Context) (*Response, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
hw.nextWait.WaitIndex = resp.Node.ModifiedIndex + 1
|
||||
hw.nextWait.AfterIndex = resp.Node.ModifiedIndex + 1
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
@ -336,10 +336,10 @@ func (g *getAction) HTTPRequest(ep url.URL) *http.Request {
|
||||
}
|
||||
|
||||
type waitAction struct {
|
||||
Prefix string
|
||||
Key string
|
||||
WaitIndex uint64
|
||||
Recursive bool
|
||||
Prefix string
|
||||
Key string
|
||||
AfterIndex uint64
|
||||
Recursive bool
|
||||
}
|
||||
|
||||
func (w *waitAction) HTTPRequest(ep url.URL) *http.Request {
|
||||
@ -347,7 +347,7 @@ func (w *waitAction) HTTPRequest(ep url.URL) *http.Request {
|
||||
|
||||
params := u.Query()
|
||||
params.Set("wait", "true")
|
||||
params.Set("waitIndex", strconv.FormatUint(w.WaitIndex, 10))
|
||||
params.Set("waitIndex", strconv.FormatUint(w.AfterIndex, 10))
|
||||
params.Set("recursive", strconv.FormatBool(w.Recursive))
|
||||
u.RawQuery = params.Encode()
|
||||
|
||||
|
@ -152,32 +152,32 @@ func TestWaitAction(t *testing.T) {
|
||||
wantHeader := http.Header{}
|
||||
|
||||
tests := []struct {
|
||||
waitIndex uint64
|
||||
recursive bool
|
||||
wantQuery string
|
||||
afterIndex uint64
|
||||
recursive bool
|
||||
wantQuery string
|
||||
}{
|
||||
{
|
||||
recursive: false,
|
||||
waitIndex: uint64(0),
|
||||
wantQuery: "recursive=false&wait=true&waitIndex=0",
|
||||
recursive: false,
|
||||
afterIndex: uint64(0),
|
||||
wantQuery: "recursive=false&wait=true&waitIndex=0",
|
||||
},
|
||||
{
|
||||
recursive: false,
|
||||
waitIndex: uint64(12),
|
||||
wantQuery: "recursive=false&wait=true&waitIndex=12",
|
||||
recursive: false,
|
||||
afterIndex: uint64(12),
|
||||
wantQuery: "recursive=false&wait=true&waitIndex=12",
|
||||
},
|
||||
{
|
||||
recursive: true,
|
||||
waitIndex: uint64(12),
|
||||
wantQuery: "recursive=true&wait=true&waitIndex=12",
|
||||
recursive: true,
|
||||
afterIndex: uint64(12),
|
||||
wantQuery: "recursive=true&wait=true&waitIndex=12",
|
||||
},
|
||||
}
|
||||
|
||||
for i, tt := range tests {
|
||||
f := waitAction{
|
||||
Key: "/foo/bar",
|
||||
WaitIndex: tt.waitIndex,
|
||||
Recursive: tt.recursive,
|
||||
Key: "/foo/bar",
|
||||
AfterIndex: tt.afterIndex,
|
||||
Recursive: tt.recursive,
|
||||
}
|
||||
got := *f.HTTPRequest(ep)
|
||||
|
||||
|
@ -190,7 +190,7 @@ func (d *discovery) createSelf(contents string) error {
|
||||
}
|
||||
|
||||
// ensure self appears on the server we connected to
|
||||
w := d.c.Watcher(d.selfKey(), &client.WatcherOptions{WaitIndex: resp.Node.CreatedIndex})
|
||||
w := d.c.Watcher(d.selfKey(), &client.WatcherOptions{AfterIndex: resp.Node.CreatedIndex})
|
||||
_, err = w.Next(context.Background())
|
||||
return err
|
||||
}
|
||||
@ -279,7 +279,7 @@ func (d *discovery) waitNodes(nodes []*client.Node, size int, index uint64) ([]*
|
||||
nodes = nodes[:size]
|
||||
}
|
||||
// watch from the next index
|
||||
w := d.c.Watcher(d.cluster, &client.WatcherOptions{WaitIndex: index + 1, Recursive: true})
|
||||
w := d.c.Watcher(d.cluster, &client.WatcherOptions{AfterIndex: index + 1, Recursive: true})
|
||||
all := make([]*client.Node, len(nodes))
|
||||
copy(all, nodes)
|
||||
for _, n := range all {
|
||||
|
@ -141,7 +141,7 @@ func TestForceNewCluster(t *testing.T) {
|
||||
cancel()
|
||||
// ensure create has been applied in this machine
|
||||
ctx, cancel = context.WithTimeout(context.Background(), requestTimeout)
|
||||
if _, err := kapi.Watcher("/foo", &client.WatcherOptions{WaitIndex: resp.Node.ModifiedIndex}).Next(ctx); err != nil {
|
||||
if _, err := kapi.Watcher("/foo", &client.WatcherOptions{AfterIndex: resp.Node.ModifiedIndex}).Next(ctx); err != nil {
|
||||
t.Fatalf("unexpected watch error: %v", err)
|
||||
}
|
||||
cancel()
|
||||
@ -162,7 +162,7 @@ func TestForceNewCluster(t *testing.T) {
|
||||
kapi = client.NewKeysAPI(cc)
|
||||
// ensure force restart keep the old data, and new cluster can make progress
|
||||
ctx, cancel = context.WithTimeout(context.Background(), requestTimeout)
|
||||
if _, err := kapi.Watcher("/foo", &client.WatcherOptions{WaitIndex: resp.Node.ModifiedIndex}).Next(ctx); err != nil {
|
||||
if _, err := kapi.Watcher("/foo", &client.WatcherOptions{AfterIndex: resp.Node.ModifiedIndex}).Next(ctx); err != nil {
|
||||
t.Fatalf("unexpected watch error: %v", err)
|
||||
}
|
||||
cancel()
|
||||
@ -188,7 +188,7 @@ func clusterMustProgress(t *testing.T, membs []*member) {
|
||||
mcc := mustNewHTTPClient(t, []string{u})
|
||||
mkapi := client.NewKeysAPI(mcc)
|
||||
mctx, mcancel := context.WithTimeout(context.Background(), requestTimeout)
|
||||
if _, err := mkapi.Watcher(key, &client.WatcherOptions{WaitIndex: resp.Node.ModifiedIndex}).Next(mctx); err != nil {
|
||||
if _, err := mkapi.Watcher(key, &client.WatcherOptions{AfterIndex: resp.Node.ModifiedIndex}).Next(mctx); err != nil {
|
||||
t.Fatalf("#%d: watch on %s error: %v", i, u, err)
|
||||
}
|
||||
mcancel()
|
||||
@ -551,7 +551,7 @@ func (m *member) WaitOK(t *testing.T) {
|
||||
kapi := client.NewKeysAPI(cc)
|
||||
for {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
|
||||
_, err := kapi.Get(ctx, "/")
|
||||
_, err := kapi.Get(ctx, "/", nil)
|
||||
if err != nil {
|
||||
time.Sleep(tickDuration)
|
||||
continue
|
||||
|
@ -109,7 +109,7 @@ func TestSnapshotAndRestartMember(t *testing.T) {
|
||||
kapi := client.NewKeysAPI(cc)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
|
||||
key := fmt.Sprintf("foo%d", i)
|
||||
resp, err := kapi.Get(ctx, "/"+key)
|
||||
resp, err := kapi.Get(ctx, "/"+key, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("#%d: get on %s error: %v", i, m.URL(), err)
|
||||
}
|
||||
|
@ -269,14 +269,18 @@ func getPeersFromDiscoveryURL(discoverURL string) ([]string, error) {
|
||||
}
|
||||
token := u.Path
|
||||
u.Path = ""
|
||||
c, err := client.NewHTTPClient(&http.Transport{}, []string{u.String()})
|
||||
cfg := client.Config{
|
||||
Transport: &http.Transport{},
|
||||
Endpoints: []string{u.String()},
|
||||
}
|
||||
c, err := client.New(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dc := client.NewDiscoveryKeysAPI(c)
|
||||
dc := client.NewKeysAPIWithPrefix(c, "")
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), client.DefaultRequestTimeout)
|
||||
resp, err := dc.Get(ctx, token)
|
||||
resp, err := dc.Get(ctx, token, nil)
|
||||
cancel()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
Loading…
x
Reference in New Issue
Block a user