Function TestFlushReleaseOnClose

test/conn_test.go:1284–1354

(t *testing.T)

}

func TestFlushReleaseOnClose(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)

	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"))

		// Hang around until asked to quit
		<-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 = false
	opts.Servers = []string{natsURL}
	nc, err := opts.Connect()
	if err != nil {
		t.Fatalf("Expected INFO message with custom max payload, got: %s", err)
	}
	defer nc.Close()

	// First try a FlushTimeout() and make sure we timeout
	if err := nc.FlushTimeout(50 * time.Millisecond); err == nil || err != nats.ErrTimeout {
		t.Fatalf("Expected a timeout error, got: %v", err)
	}