Function TestReconnectOnFlusherError

test/conn_test.go:2084–2207

(t *testing.T)

}

func TestReconnectOnFlusherError(t *testing.T) {
	if runtime.GOOS == "windows" {
		t.SkipNow()
	}

	for _, tc := range []struct {
		name          string
		withOption    bool
		noReconnect   bool
		wantReconnect bool
		wantClosed    bool
	}{
		{"enabled", true, false, true, false},
		{"disabled", false, false, false, false},
		{"no_reconnect", true, true, false, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			s := RunDefaultServer()
			defer s.Shutdown()

			fakeAddr := startStalledMockServer(t)

			reconnectedCh := make(chan struct{}, 1)
			closedCh := make(chan struct{}, 1)
			asyncErrCh := make(chan error, 1)

			opts := []nats.Option{
				nats.SetCustomDialer(&lowWriteBufferDialer{}),
				nats.FlusherTimeout(15 * time.Millisecond),
				nats.MaxReconnects(10),
				nats.DontRandomize(),
				nats.ReconnectHandler(func(_ *nats.Conn) {
					select {
					case reconnectedCh <- struct{}{}:
					default:
					}
				}),
				nats.ClosedHandler(func(_ *nats.Conn) {
					select {
					case closedCh <- struct{}{}:
					default:
					}
				}),
				nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
					select {
					case asyncErrCh <- err:
					default:
					}
				}),
			}
			if tc.withOption {
				opts = append(opts, nats.ReconnectOnFlusherError())
			}
			if tc.noReconnect {
				opts = append(opts, nats.NoReconnect())
			}

			nc, err := nats.Connect(