Content streaming

[jeremyrsmith]

GET, POST, and PUT requests should support content streaming.

GET requests should allow content to be streamed from the resource.

POST and PUT requests should allow content to be streamed to the resource.

Finagle has various mechanisms to support this; featherbed should identify a good standard and add the support for that standard.

[herberteuler]

I am interested in working on this. Here is my proposal on interfaces:

For GET requests, we provide a method

def streamingResponse[K](implicit
  canBuild: CanBuildRequest[Self],
  decodeAll: DecodeAll[K, Accept]
): Future[(AsyncStream[Xor[NonEmptyList[Throwable], K]], Service)]

The user can then retrieve streamed values from the returned stream. If the stream is finite, it will be closed automatically on exhaustion. Otherwise, the user can close the service to indicate that no more values are interested and the connection can be closed.

For POST and PUT requests, in addition to the method above, we provide the following method:

def streamingRequest[T, Type <: String](
  offer: Offer[T], close: Offer[Unit], typ: Type)(implicit
  witness: Witness.Aux[typ.Type]
): PostRequest[T, typ.type, Accept]

The user can then stream values through offer, and when streaming is done publish a close signal through close.

[jeremyrsmith]

@herberteuler I'm not all that up-to-speed on the streaming story for HTTP with respect to AsyncStream. What does it mean for a response to be AsyncStream[K]? Does this mean that as the response is incoming, several K can be decoded?

I suspect that AsyncStream might make more sense for websockets (which I'm not sure is in scope for featherbed, as featherbed's goal is to focus pretty narrowly on REST). Can you give an example of how AsyncStream might be used for a request/response use case?

[herberteuler]

Streaming support in finagle-http is done via chunked transfer, which resembles a stream of objects, with each chunk being a serialized object.

The view of streaming by AsyncStreams has already been taken by finch. For example, in finch, asyncBody of type Endpoint[AsyncStream[Buf]] can be used to read in a stream of chunked binaries as Bufs, and a response of type AsyncStream[A] means that the result is a stream of objects of type A.

The project I am working on mimics Kubernetes' architecture. We are building an API server like RESTful service on top of ZooKeeper. There are two APIs using streaming: GET /resources?watch=true streams updates to resource list (i.e. watch on ZooKeeper's getChildren) and GET /resources/ID?watch=true streams updated values of a specific resource with ID (i.e. watch on ZooKeeper's getData). These can be viewed as a stream of resource IDs AsyncStream[List[String]] and a stream of values AsyncStream[A], respectively. I hope this makes sense.

Hence for responses, a stream of A values as AsyncStream[A] looks like a proper abstraction. For requests, the process needs to be reversed. A stream of objects can be built from an offer to which the objects are fed.

The returned Service in streamingResponse and Offer[Unit] in streamingRequest are awkward, but they are there to support infinite streams. Streaming response is nothing but a long-waiting HTTP connection, when the client is not interested in the upcoming results anymore, it simply closes the connection, so a handle to Service is necessary in this case. The same logic can be applied to the request side, hence the Offer[Unit] parameter.