Function TestServerErrorClosesConnection

test/conn_test.go:1620–1715

(t *testing.T)

}

func TestServerErrorClosesConnection(t *testing.T) {
	serverInfo := "INFO {\"server_id\":\"foobar\",\"host\":\"%s\",\"port\":%d,\"auth_required\":false,\"tls_required\":false,\"max_payload\":1048576}\r\n"

	l, e := net.Listen("tcp", "127.0.0.1:0")
	if e != nil {
		t.Fatal("Could not listen on an ephemeral port")
	}
	tl := l.(*net.TCPListener)
	defer tl.Close()

	addr := tl.Addr().(*net.TCPAddr)
	done := make(chan bool)
	dch := make(chan bool)
	cch := make(chan bool)

	serverSentError := "Any Error"
	reconnected := int64(0)

	errCh := make(chan error, 1)
	go func() {
		conn, err := l.Accept()
		if err != nil {
			errCh <- fmt.Errorf("error accepting client connection: %v", err)
			return
		}
		defer conn.Close()
		info := fmt.Sprintf(serverInfo, addr.IP, addr.Port)
		conn.Write([]byte(info))

		// Read connect and ping commands sent from the client
		br := bufio.NewReaderSize(conn, 1024)
		if _, err := br.ReadString('\n'); err != nil {
			errCh <- fmt.Errorf("expected CONNECT from client, got: %s", err)
			return
		}
		if _, err := br.ReadString('\n'); err != nil {
			errCh <- fmt.Errorf("expected PING from client, got: %s", err)
			return
		}
		conn.Write([]byte("PONG\r\n"))

		// Wait a tiny, and simulate a Stale Connection
		time.Sleep(50 * time.Millisecond)
		conn.Write([]byte("-ERR '" + serverSentError + "'\r\n"))

		// Hang around a bit
		<-done
	}()

	// Wait for server mock to start
	time.Sleep(100 * time.Millisecond)

	natsURL := fmt.Sprintf("nats://%s:%d", addr.IP, addr.Port)
	opts := nats.GetDefaultOptions()
	opts.AllowReconnect = true
	opts.DisconnectedErrCB = func(_ *nats.Conn, _ error) { dch <- true }
	opts.ReconnectedCB = func(_ *nats.Conn) { atomic.AddInt64(&reconnected, 1) }
	opts.ClosedCB = func(_ *nats.Conn) { cch <- true }