Function TestSubscriptionEvents

test/sub_test.go:1658–1839

(t *testing.T)

}

func TestSubscriptionEvents(t *testing.T) {
	t.Run("default events", func(t *testing.T) {
		s := RunDefaultServer()
		defer s.Shutdown()

		nc := NewDefaultConnection(t)
		// disable slow consumer prints
		nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {})
		defer nc.Close()

		blockChan := make(chan struct{})
		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
			// block in subscription callback
			// to force slow consumer
			<-blockChan
		})
		if err != nil {
			t.Fatalf("Error subscribing: %v", err)
		}
		sub.SetPendingLimits(10, 1024)
		status := sub.StatusChanged()

		// initial status
		WaitOnChannel(t, status, nats.SubscriptionActive)

		for range 11 {
			nc.Publish("foo", []byte("Hello"))
		}
		WaitOnChannel(t, status, nats.SubscriptionSlowConsumer)
		close(blockChan)

		sub.Drain()

		WaitOnChannel(t, status, nats.SubscriptionDraining)

		WaitOnChannel(t, status, nats.SubscriptionClosed)
	})

	t.Run("slow consumer event only", func(t *testing.T) {
		s := RunDefaultServer()
		defer s.Shutdown()

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

		blockChan := make(chan struct{})
		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
			// block in subscription callback
			// to force slow consumer
			<-blockChan
		})
		// disable slow consumer prints
		nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {})
		defer sub.Unsubscribe()
		if err != nil {
			t.Fatalf("Error subscribing: %v", err)
		}
		sub.SetPendingLimits(10, 1024)