feat: optional SSE in REST protocol#176
Conversation
Signed-off-by: Jordan Oroshiba <jordan@astria.org>
joroshiba
force-pushed
the
sse-support
branch
from
1d4ad09 to
b9a1ff0
Compare
…support via protobuf annotation directives Signed-off-by: Jordan Oroshiba <jordan@astria.org>
Signed-off-by: Jordan Oroshiba <jordan@astria.org>
emcfarlane
left a comment
emcfarlane
left a comment
There was a problem hiding this comment.
Thanks for taking the time to put this PR together. I can see the need for having a custom encoding to support streaming use cases but have concerns about the approach. For extending the definition of google.api.http it would be better suited as a custom plugin (see this issue: #159), where you would be able to configure the streaming behaviour to match exactly what your clients expect. To add this behaviour to vanguard we would require the behaviour to be as generally applicable as possible. We don't however currently support pluggable interfaces.
| rpc WatchBooks(WatchBooksRequest) returns (stream BookUpdate) { | ||
| option (google.api.http) = { | ||
| get: "/v1/{parent=shelves/*}/books:watch" | ||
| response_body: "SSE_EVENT=type,SSE_ID=sequence,SSE_OMIT" |
There was a problem hiding this comment.
This response_body is not a valid field path. It breaks the semantic meaning describe in google.api.http. Any other tool that interprets HttpRule annotations will misinterpret this.
| **Available directives:** | ||
| - `SSE_EVENT=field_name` - Extract `field_name` from each message to use as the SSE event type | ||
| - `SSE_ID=field_name` - Extract `field_name` from each message to use as the SSE event ID | ||
| - `SSE_OMIT` - Remove the extracted fields from the JSON data payload |
There was a problem hiding this comment.
A custom annotation would be more idiomatic and avoid overloading the response_body.
| var jsonData map[string]interface{} | ||
| needsFiltering := s.omitExtractedFields && (s.eventField != "" || s.eventIDField != "" || s.retryField != "") | ||
| err := json.Unmarshal(data, &jsonData) |
There was a problem hiding this comment.
This unmarshals the already marshalled data to extract and filter and then must re-encode. This has overhead but also correctness issues, the map[string]any will re-order the fields.
| if r.useSSE && !isErr { | ||
| headers["Content-Type"] = []string{contentTypeSSE} | ||
| headers["Cache-Control"] = []string{"no-cache"} | ||
| headers["Connection"] = []string{"keep-alive"} |
There was a problem hiding this comment.
keep-alive should not be set for HTTP/2.
| buf.WriteString(",") | ||
| } | ||
| first = false | ||
| buf.WriteString(`"` + k + `":"` + v + `"`) |
There was a problem hiding this comment.
This must escape the key and values for headers to avoid breaking encoding.
| eventIDField string // Field name to extract from message for event ID | ||
| retryField string // Field name to extract from message for retry interval | ||
| omitExtractedFields bool // Whether to remove extracted fields from data payload | ||
| responseType protoreflect.MessageType |
There was a problem hiding this comment.
responseType is never used.
| buf.WriteString(sseNewline) | ||
|
|
||
| // Write to delegate | ||
| _, err := s.delegate.Write(buf.Bytes()) |
There was a problem hiding this comment.
Does this need to call Flush()?
2 participants
This adds support for SSE streaming on REST Protocol via an option, but is not enabled by default. When using
WithRESTServerSentEvents(), only streaming rpcs using REST protocol when request headerAcceptis set totext/event-streamare affected.When SSE is being used:
openis fireddatafield of the event.errorevents are generated when error encounteredcompleteevent is firedThere are SSE specific field configurations available for
eventidandretry. To support these fields, custom directives are parsed out of thegoogle.api.httpresponse_body annotation field, to configure these fields to be set as specifed for each data sent.SSE_EVENT=<field_name>will change the event name on the SSE to be derived from the field specifiedSSE_ID=<field_name>, will change the event id to be derived from the field specified. If this is not set no ID is specified, when it is the id field will be blank unless the attribute has a value on the rendered data.SSE_RETRY=<field_name>, will set the retry field based on the value, assuming it is all ascii numbers.SSE_OMITwill, if any of the above are set, omit the fields specified from the data json body.Related issue:
#161