Function TestAsyncErrHandler

test/sub_test.go:572–649

(t *testing.T)

}

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

	opts := nats.GetDefaultOptions()

	nc, err := opts.Connect()
	if err != nil {
		t.Fatalf("Could not connect to server: %v\n", err)
	}
	defer nc.Close()

	subj := "async_test"
	bch := make(chan bool)

	sub, err := nc.Subscribe(subj, func(_ *nats.Msg) {
		// block to back us up..
		<-bch
	})
	if err != nil {
		t.Fatalf("Could not subscribe: %v\n", err)
	}

	limit := 10
	toSend := 100

	// Limit internal subchan length to trip condition easier.
	sub.SetPendingLimits(limit, 1024)

	ch := make(chan bool)

	aeCalled := int64(0)

	nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {
		atomic.AddInt64(&aeCalled, 1)

		if s != sub {
			t.Fatal("Did not receive proper subscription")
		}
		if !errors.Is(e, nats.ErrSlowConsumer) {
			t.Fatalf("Did not receive proper error: %v vs %v", e, nats.ErrSlowConsumer)
		}
		// Suppress additional calls
		if atomic.LoadInt64(&aeCalled) == 1 {
			// release the sub
			defer close(bch)
			// release the test
			ch <- true
		}
	})

	b := []byte("Hello World!")
	// First one trips the ch wait in subscription callback.
	nc.Publish(subj, b)
	nc.Flush()
	for range toSend {
		nc.Publish(subj, b)
	}