Function TestCompatibilityObjectStoreWatchUpdates

test/compat_test.go:341–392

(t *testing.T)

}

func TestCompatibilityObjectStoreWatchUpdates(t *testing.T) {
	t.Parallel()

	type config struct {
		Bucket string `json:"bucket"`
		Object string `json:"object"`
	}

	nc := connect(t)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Error connecting to NATS: %v", err)
	}
	defer nc.Close()
	ctx := context.Background()

	// setup subscription on which tester will be sending requests
	sub, err := nc.SubscribeSync("tests.object-store.watch-updates.>")
	if err != nil {
		t.Fatalf("Error subscribing to test subject: %v", err)
	}
	defer sub.Unsubscribe()

	msg, err := sub.NextMsg(1 * time.Hour)
	if err != nil {
		t.Fatalf("Error getting message: %v", err)
	}
	// Watch object
	var cfg config
	if err := json.Unmarshal(msg.Data, &cfg); err != nil {
		t.Fatalf("Error unmarshalling message: %v", err)
	}
	os, err := js.ObjectStore(ctx, cfg.Bucket)
	if err != nil {
		t.Fatalf("Error getting object store: %v", err)
	}
	watcher, err := os.Watch(ctx, jetstream.UpdatesOnly())
	if err != nil {
		t.Fatalf("Error getting watcher: %v", err)
	}
	var info *jetstream.ObjectInfo
	select {
	case info = <-watcher.Updates():
	case <-time.After(30 * time.Second):
		t.Fatalf("Timeout waiting for object update")
	}

	if err := msg.Respond([]byte(info.Digest)); err != nil {
		t.Fatalf("Error responding to message: %v", err)
	}
	validateTestResult(t, sub)
}

func TestCompatibilityObjectStoreGetLink(t *testing.T) {
	t.Parallel()