Function TestStreamSourceWithConsumer

jetstream/test/jetstream_test.go:2534–2596

(t *testing.T)

}

func TestStreamSourceWithConsumer(t *testing.T) {
	srv := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, srv)

	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	upstream, err := js.CreateStream(ctx, jetstream.StreamConfig{
		Name:     "UP",
		Subjects: []string{"up"},
	})
	if err != nil {
		t.Fatalf("CreateStream upstream: %v", err)
	}

	if _, err := upstream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Durable:        "C",
		DeliverSubject: "deliver",
		AckPolicy:      jetstream.AckFlowControlPolicy,
		IdleHeartbeat:  time.Second,
	}); err != nil {
		t.Fatalf("CreateOrUpdateConsumer: %v", err)
	}

	down, err := js.CreateStream(ctx, jetstream.StreamConfig{
		Name: "DOWN",
		Sources: []*jetstream.StreamSource{{
			Name: "UP",
			Consumer: &jetstream.StreamConsumerSource{
				Name:           "C",
				DeliverSubject: "deliver",
			},
		}},
	})
	if err != nil {
		t.Fatalf("CreateStream down: %v", err)
	}

	if _, err := js.Publish(ctx, "up", []byte("hello")); err != nil {
		t.Fatalf("Publish: %v", err)
	}

	checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
		info, err := down.Info(ctx)
		if err != nil {
			return err
		}
		if info.State.Msgs != 1 {