runtime: marshal custom stream delimiters

runtime/handler.go

@@ -38,6 +38,11 @@ func ForwardResponseStream(ctx context.Context, mux *ServeMux, marshaler Marshal
 		return
 	}
 
+	var delimiter []byte
+	if d, ok := marshaler.(Delimited); ok {
+		delimiter = d.Delimiter()
+	}
+
 	var wroteHeader bool
 	for {
 		resp, err := recv()
@@ -64,6 +69,10 @@ func ForwardResponseStream(ctx context.Context, mux *ServeMux, marshaler Marshal
 			return
 		}
 		wroteHeader = true
+		if _, err = w.Write(delimiter); err != nil {
+			grpclog.Printf("Failed to send delimiter chunk: %v", err)
+			return
+		}
 		f.Flush()
 	}
 }

runtime/handler_test.go

@@ -93,6 +93,7 @@ func TestForwardResponseStream(t *testing.T) {
 					t.Errorf("marshaler.Marshal() failed %v", err)
 				}
 				want = append(want, b...)
+				want = append(want, marshaler.Delimiter()...)
 			}
 
 			if string(body) != string(want) {

runtime/marshal_json.go

@@ -35,3 +35,8 @@ func (j *JSONBuiltin) NewDecoder(r io.Reader) Decoder {
 func (j *JSONBuiltin) NewEncoder(w io.Writer) Encoder {
 	return json.NewEncoder(w)
 }
+
+// Delimiter for newline encoded JSON streams.
+func (j *JSONBuiltin) Delimiter() []byte {
+	return []byte("\n")
+}

runtime/marshal_jsonpb.go

@@ -182,3 +182,8 @@ type protoEnum interface {
 }
 
 var typeProtoMessage = reflect.TypeOf((*proto.Message)(nil)).Elem()
+
+// Delimiter for newline encoded JSON streams.
+func (j *JSONPb) Delimiter() []byte {
+	return []byte("\n")
+}

runtime/marshaler.go

@@ -40,3 +40,9 @@ type EncoderFunc func(v interface{}) error
 
 // Encode delegates invocations to the underlying function itself.
 func (f EncoderFunc) Encode(v interface{}) error { return f(v) }
+
+// Delimited defines the streaming delimiter.
+type Delimited interface {
+	// Delimiter returns the record seperator for the stream.
+	Delimiter() []byte
+}