Function TestChanSubscriberPendingLimits

test/sub_test.go:934–1013

(t *testing.T)

}

func TestChanSubscriberPendingLimits(t *testing.T) {
	s := RunDefaultServer()
	defer s.Shutdown()

	nc := NewDefaultConnection(t)
	defer nc.Close()
	ncp := NewDefaultConnection(t)
	defer ncp.Close()

	// There was a defect that prevented to receive more than
	// the default pending message limit. Trying to send more
	// than this limit.
	pending := 1000
	total := pending + 100

	for typeSubs := 0; typeSubs < 3; typeSubs++ {

		func() {
			// Create our own channel.
			ch := make(chan *nats.Msg, total)

			var err error
			var sub *nats.Subscription
			switch typeSubs {
			case 0:
				sub, err = nc.ChanSubscribe("foo", ch)
				if err := sub.SetPendingLimits(pending, -1); err == nil {
					t.Fatalf("Expected an error setting pending limits")
				}
			case 1:
				sub, err = nc.ChanQueueSubscribe("foo", "bar", ch)
				if err := sub.SetPendingLimits(pending, -1); err == nil {
					t.Fatalf("Expected an error setting pending limits")
				}
			case 2:
				sub, err = nc.QueueSubscribeSyncWithChan("foo", "bar", ch)
				if err := sub.SetPendingLimits(pending, -1); err == nil {
					t.Fatalf("Expected an error setting pending limits")
				}
			}
			if err != nil {
				t.Fatalf("Unexpected error on subscribe: %v", err)
			}
			defer sub.Unsubscribe()
			nc.Flush()

			// Send some messages
			for range total {
				if err := ncp.Publish("foo", []byte("Hello")); err != nil {
					t.Fatalf("Unexpected error on publish: %v", err)
				}
			}

			received := 0
			tm := time.NewTimer(10 * time.Second)
			defer tm.Stop()

			chk := func(ok bool) {