Function TestPullConsumerConsume_WithCluster

jetstream/test/pull_test.go:3319–3536

(t *testing.T)

}

func TestPullConsumerConsume_WithCluster(t *testing.T) {
	testSubject := "FOO.123"
	testMsgs := []string{"m1", "m2", "m3", "m4", "m5"}
	publishTestMsgs := func(t *testing.T, js jetstream.JetStream) {
		for _, msg := range testMsgs {
			if _, err := js.Publish(context.Background(), testSubject, []byte(msg)); err != nil {
				t.Fatalf("Unexpected error during publish: %s", err)
			}
		}
	}

	name := "cluster"
	singleStream := jetstream.StreamConfig{
		Name:     name,
		Replicas: 1,
		Subjects: []string{"FOO.*"},
	}

	streamWithReplicas := jetstream.StreamConfig{
		Name:     name,
		Replicas: 3,
		Subjects: []string{"FOO.*"},
	}

	for _, stream := range []jetstream.StreamConfig{singleStream, streamWithReplicas} {
		t.Run(fmt.Sprintf("num replicas: %d, no options", stream.Replicas), func(t *testing.T) {
			withJSClusterAndStream(t, name, 3, stream, func(t *testing.T, subject string, srvs ...*jsServer) {
				nc, err := nats.Connect(srvs[0].ClientURL())
				if err != nil {
					t.Fatalf("Unexpected error: %v", err)
				}

				js, err := jetstream.New(nc)
				if err != nil {
					t.Fatalf("Unexpected error: %v", err)
				}
				defer nc.Close()

				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
				defer cancel()
				s, err := js.Stream(ctx, stream.Name)
				if err != nil {
					t.Fatalf("Unexpected error: %v", err)
				}
				c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
				if err != nil {
					t.Fatalf("Unexpected error: %v", err)
				}

				msgs := make([]jetstream.Msg, 0)
				wg := &sync.WaitGroup{}
				wg.Add(len(testMsgs))
				l, err := c.Consume(func(msg jetstream.Msg) {
					msgs = append(msgs, msg)
					wg.Done()
				})
				if err != nil {
					t.Fatalf("Unexpected error: %v", err)