Function TestObjectWatch

jetstream/test/object_test.go:566–729

(t *testing.T)

}

func TestObjectWatch(t *testing.T) {
	expectUpdateF := func(t *testing.T, watcher jetstream.ObjectWatcher) func(name string) {
		return func(name string) {
			t.Helper()
			select {
			case info := <-watcher.Updates():
				if false && info.Name != name { // TODO what is supposed to happen here?
					t.Fatalf("Expected update for %q, but got %+v", name, info)
				}
			case <-time.After(time.Second):
				t.Fatalf("Did not receive an update like expected")
			}
		}
	}

	expectNoMoreUpdatesF := func(t *testing.T, watcher jetstream.ObjectWatcher) func() {
		return func() {
			t.Helper()
			select {
			case info := <-watcher.Updates():
				t.Fatalf("Got an unexpected update: %+v", info)
			case <-time.After(100 * time.Millisecond):
			}
		}
	}

	expectInitDoneF := func(t *testing.T, watcher jetstream.ObjectWatcher) func() {
		return func() {
			t.Helper()
			select {
			case info := <-watcher.Updates():
				if info != nil {
					t.Fatalf("Did not get expected: %+v", info)
				}
			case <-time.After(time.Second):
				t.Fatalf("Did not receive a init done like expected")
			}
		}
	}

	t.Run("default watcher", func(t *testing.T) {
		s := RunBasicJetStreamServer()
		defer shutdownJSServerAndRemoveStorage(t, s)

		nc, js := jsClient(t, s)
		defer nc.Close()
		ctx := context.Background()

		obs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "WATCH-TEST"})
		expectOk(t, err)

		watcher, err := obs.Watch(ctx)
		expectOk(t, err)
		defer watcher.Stop()

		expectUpdate := expectUpdateF(t, watcher)
		expectNoMoreUpdates := expectNoMoreUpdatesF(t, watcher)
		expectInitDone := expectInitDoneF(t, watcher)