integration: test with multi events in unsynced

Related etcd-io#4216.

integration/v3_grpc_test.go

@@ -360,8 +360,6 @@ func TestV3WatchFromCurrentRevision(t *testing.T) {
 				},
 			},
 		},
-
-		// TODO: watch and receive multiple-events from synced (need Txn)
 	}
 
 	for i, tt := range tests {
@@ -636,6 +634,68 @@ func TestV3WatchMultipleEventsFromCurrentRevision(t *testing.T) {
 	clus.Terminate(t)
 }
 
+// TestV3WatchMultipleEventsFromUnsynced tests Watch APIs from unsynced watchers.
+func TestV3WatchMultipleEventsFromUnsynced(t *testing.T) {
+	clus := newClusterGRPC(t, &clusterConfig{size: 3})
+
+	wAPI := pb.NewWatchClient(clus.RandConn())
+	wStream, wErr := wAPI.Watch(context.TODO())
+	if wErr != nil {
+		t.Fatalf("wAPI.Watch error: %v", wErr)
+	}
+
+	if err := wStream.Send(&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("foo"), StartRevision: 1}}); err != nil {
+		t.Fatalf("wStream.Send error: %v", err)
+	}
+
+	kvc := pb.NewKVClient(clus.RandConn())
+	for i := 0; i < 3; i++ {
+		if _, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte(fmt.Sprintf("foo%d", i)), Value: []byte("bar")}); err != nil {
+			t.Fatalf("couldn't put key (%v)", err)
+		}
+	}
+
+	events := []*storagepb.Event{}
+	for len(events) < 3 {
+		resp, err := wStream.Recv()
+		if err != nil {
+			t.Errorf("wStream.Recv error: %v", err)
+		}
+		if resp.Created {
+			continue
+		}
+		events = append(events, resp.Events...)
+	}
+	sort.Sort(eventsSortByKey(events))
+
+	wevents := []*storagepb.Event{
+		{
+			Type: storagepb.PUT,
+			Kv:   &storagepb.KeyValue{Key: []byte("foo0"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
+		},
+		{
+			Type: storagepb.PUT,
+			Kv:   &storagepb.KeyValue{Key: []byte("foo1"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
+		},
+		{
+			Type: storagepb.PUT,
+			Kv:   &storagepb.KeyValue{Key: []byte("foo2"), Value: []byte("bar"), CreateRevision: 4, ModRevision: 4, Version: 1},
+		},
+	}
+
+	if !reflect.DeepEqual(events, wevents) {
+		t.Errorf("events got = %+v, want = %+v", events, wevents)
+	}
+
+	rok, nr := WaitResponse(wStream, 1*time.Second)
+	if !rok {
+		t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
+	}
+
+	// can't defer because tcp ports will be in use
+	clus.Terminate(t)
+}
+
 type eventsSortByKey []*storagepb.Event
 
 func (evs eventsSortByKey) Len() int           { return len(evs) }