Function testJetStream_ClusterMultipleFetchPullSubscribe

test/js_test.go:7203–7334

(t *testing.T, subject string, srvs ...*jsServer)

}

func testJetStream_ClusterMultipleFetchPullSubscribe(t *testing.T, subject string, srvs ...*jsServer) {
	srv := srvs[0]
	nc, js := jsClient(t, srv.Server)
	defer nc.Close()

	var wg sync.WaitGroup
	ctx, done := context.WithTimeout(context.Background(), 5*time.Second)
	defer done()

	// Setup a number of subscriptions with different inboxes that will be
	// fetching the messages in parallel.
	nsubs := 4
	subs := make([]*nats.Subscription, nsubs)
	errCh := make(chan error, nsubs)
	var queues sync.Map
	for i := 0; i < nsubs; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			var sub *nats.Subscription
			var err error
			sub, err = js.PullSubscribe(subject, "shared")
			if err != nil {
				errCh <- err
			} else {
				subs[n] = sub
				queues.Store(sub.Subject, make([]*nats.Msg, 0))
			}
		}(i)
	}

	// Publishing of messages happens after the subscriptions are ready.
	// The subscribers will be fetching messages while these are being
	// produced so sometimes there are not going to be messages available.
	wg.Wait()
	var (
		total     uint64 = 100
		delivered uint64
		batchSize = 2
	)
	go func() {
		for i := 0; i < int(total); i++ {
			js.Publish(subject, []byte(fmt.Sprintf("n:%v", i)))
			time.Sleep(1 * time.Millisecond)
		}
	}()

	ctx2, done2 := context.WithTimeout(ctx, 3*time.Second)
	defer done2()

	for _, psub := range subs {
		if psub == nil {
			continue
		}
		sub := psub
		subject := sub.Subject
		v, _ := queues.Load(sub.Subject)
		queue := v.([]*nats.Msg)