events: fix slow client connection to empty event stream

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## 1.1.1 (Unreleased)
 
 BUG FIXES:
+* api: Fixed event stream connection initialization when there are no events to send [[GH-10637](https://github.com/hashicorp/nomad/issues/10637)]
 * cli: Fixed a bug where `quota status` and `namespace status` commands may panic if the CLI targets a pre-1.1.0 cluster
 
 ## 1.1.0 (May 18, 2021)

nomad/stream/ndjson.go

@@ -33,14 +33,16 @@ type JsonStream struct {
 
 // NewJsonStream creates a new json stream that will output Json structs
 // to the passed output channel. The constructor starts a goroutine
-// to begin heartbeating on its set interval.
+// to begin heartbeating on its set interval and also sends an initial heartbeat
+// to notify the client about the successful connection initialization.
 func NewJsonStream(ctx context.Context, heartbeat time.Duration) *JsonStream {
 	s := &JsonStream{
 		ctx:           ctx,
 		outCh:         make(chan *structs.EventJson, 10),
 		heartbeatTick: time.NewTicker(heartbeat),
 	}
 
+	s.outCh <- JsonHeartbeat
 	go s.heartbeat()
 
 	return s

nomad/stream/ndjson_test.go

@@ -1,7 +1,6 @@
 package stream
 
 import (
-	"bytes"
 	"context"
 	"testing"
 	"time"
@@ -24,12 +23,12 @@ func TestJsonStream(t *testing.T) {
 
 	require.NoError(t, s.Send(testObj{Name: "test"}))
 
-	out1 := <-out
+	initialHeartbeat := <-out
+	require.Equal(t, []byte(`{}`), initialHeartbeat.Data)
 
-	var expected bytes.Buffer
-	expected.Write([]byte(`{"name":"test"}`))
+	testMessage1 := <-out
+	require.Equal(t, []byte(`{"name":"test"}`), testMessage1.Data)
 
-	require.Equal(t, expected.Bytes(), out1.Data)
 	select {
 	case msg := <-out:
 		require.Failf(t, "Did not expect another message", "%#v", msg)
@@ -38,12 +37,8 @@ func TestJsonStream(t *testing.T) {
 
 	require.NoError(t, s.Send(testObj{Name: "test2"}))
 
-	out2 := <-out
-	expected.Reset()
-
-	expected.Write([]byte(`{"name":"test2"}`))
-	require.Equal(t, expected.Bytes(), out2.Data)
-
+	testMessage2 := <-out
+	require.Equal(t, []byte(`{"name":"test2"}`), testMessage2.Data)
 }
 
 func TestJson_Send_After_Stop(t *testing.T) {