one ndjson line per index

wrap returned events in a events wrapper that includes index

nomad/event_endpoint.go

@@ -117,19 +117,16 @@ func (e *Event) stream(conn io.ReadWriteCloser) {
 			}
 
 			// Continue if there are no events
-			if events == nil {
+			if len(events.Events) == 0 {
 				continue
 			}
 
-			// Send each event as its own frame
-			for _, e := range events {
-				if err := jsonStream.Send(e); err != nil {
-					select {
-					case errCh <- err:
-					case <-ctx.Done():
-					}
-					break LOOP
+			if err := jsonStream.Send(events); err != nil {
+				select {
+				case errCh <- err:
+				case <-ctx.Done():
 				}
+				break LOOP
 			}
 		}
 	}()

nomad/event_endpoint_test.go

@@ -91,7 +91,7 @@ OUTER:
 				continue
 			}
 
-			var event stream.Event
+			var event stream.Events
 			err = json.Unmarshal(msg.Event.Data, &event)
 			require.NoError(t, err)
 
@@ -102,7 +102,7 @@ OUTER:
 				Result:   &out,
 			}
 			dec, err := mapstructure.NewDecoder(cfg)
-			dec.Decode(event.Payload)
+			dec.Decode(event.Events[0].Payload)
 			require.NoError(t, err)
 			require.Equal(t, node.ID, out.ID)
 			break OUTER
@@ -123,7 +123,7 @@ func TestEventStream_StreamErr(t *testing.T) {
 	testutil.WaitForLeader(t, s1.RPC)
 
 	req := structs.EventStreamRequest{
-		Topics: map[stream.Topic][]string{"*": []string{"*"}},
+		Topics: map[stream.Topic][]string{"*": {"*"}},
 		QueryOptions: structs.QueryOptions{
 			Region: s1.Region(),
 		},
@@ -210,7 +210,7 @@ func TestEventStream_RegionForward(t *testing.T) {
 
 	// Create request targed for region foo
 	req := structs.EventStreamRequest{
-		Topics: map[stream.Topic][]string{"*": []string{"*"}},
+		Topics: map[stream.Topic][]string{"*": {"*"}},
 		QueryOptions: structs.QueryOptions{
 			Region: "foo",
 		},
@@ -272,7 +272,7 @@ OUTER:
 				continue
 			}
 
-			var event stream.Event
+			var event stream.Events
 			err = json.Unmarshal(msg.Event.Data, &event)
 			require.NoError(t, err)
 
@@ -282,7 +282,7 @@ OUTER:
 				Result:   &out,
 			}
 			dec, err := mapstructure.NewDecoder(cfg)
-			dec.Decode(event.Payload)
+			dec.Decode(event.Events[0].Payload)
 			require.NoError(t, err)
 			require.Equal(t, node.ID, out.ID)
 			break OUTER

nomad/stream/event.go

@@ -14,3 +14,8 @@ type Event struct {
 	Index      uint64
 	Payload    interface{}
 }
+
+type Events struct {
+	Index  uint64
+	Events []Event
+}

nomad/stream/subscription.go

@@ -58,26 +58,26 @@ func newSubscription(req *SubscribeRequest, item *bufferItem, unsub func()) *Sub
 	}
 }
 
-func (s *Subscription) Next(ctx context.Context) ([]Event, error) {
+func (s *Subscription) Next(ctx context.Context) (Events, error) {
 	if atomic.LoadUint32(&s.state) == subscriptionStateClosed {
-		return nil, ErrSubscriptionClosed
+		return Events{}, ErrSubscriptionClosed
 	}
 
 	for {
 		next, err := s.currentItem.Next(ctx, s.forceClosed)
 		switch {
 		case err != nil && atomic.LoadUint32(&s.state) == subscriptionStateClosed:
-			return nil, ErrSubscriptionClosed
+			return Events{}, ErrSubscriptionClosed
 		case err != nil:
-			return nil, err
+			return Events{}, err
 		}
 		s.currentItem = next
 
 		events := filter(s.req, next.Events)
 		if len(events) == 0 {
 			continue
 		}
-		return events, nil
+		return Events{Index: next.Index, Events: events}, nil
 	}
 }