Function TestQueueSubscribeIterator

test/nats_iter_test.go:227–336

(t *testing.T)

}

func TestQueueSubscribeIterator(t *testing.T) {
	t.Run("basic", func(t *testing.T) {
		s := RunServerOnPort(-1)
		defer s.Shutdown()

		nc, err := nats.Connect(s.ClientURL())
		if err != nil {
			t.Fatalf("Error on connect: %v", err)
		}
		defer nc.Close()

		subs := make([]*nats.Subscription, 4)
		for i := 0; i < 4; i++ {
			sub, err := nc.QueueSubscribeSync("foo", "q")
			if err != nil {
				t.Fatal("Failed to subscribe: ", err)
			}
			subs[i] = sub
			defer sub.Unsubscribe()
		}

		// Send some messages to ourselves.
		total := 100
		for i := 0; i < total; i++ {
			if err := nc.Publish("foo", []byte(fmt.Sprintf("%d", i))); err != nil {
				t.Fatalf("Error on publish: %v", err)
			}
		}

		wg := sync.WaitGroup{}
		wg.Add(100)
		startWg := sync.WaitGroup{}
		startWg.Add(4)

		for i := range subs {
			go func(i int) {
				startWg.Done()
				for _, err := range subs[i].MsgsTimeout(100 * time.Millisecond) {
					if err != nil {
						break
					}
					wg.Done()
				}
			}(i)
		}

		startWg.Wait()

		wg.Wait()

		for _, sub := range subs {
			if _, err = sub.NextMsg(100 * time.Millisecond); !errors.Is(err, nats.ErrTimeout) {
				t.Fatalf("Expected timeout waiting for next message, got %v", err)
			}
		}
	})

	t.Run("permissions violation", func(t *testing.T) {