Support for request/response streaming?

[yfinkelstein]

I may have misunderstood the scope of the framework. I see that HttpObjectAggregator is unconditionally inserted into the channel pipeline, and http service invocation handler delivers FullHttpRequest all the time. So, what about large requests with multipart encoding or simply chunked encoding that would benefit from stream processing?
Also, gRPC, for example, supports steaming requests and responses as a part of their IDL definition. This makes it possible to extend the basic "message in/message out" RPC with "message in/many messages out" etc model.
Is this completely out of scope for the Armeria, on you can see this being added some time soon?

[trustin]

You are correct. Armeria does not support streaming request/response at the moment, and we hope to change this before 1.0.

[trustin]

Thinking of using reactive streams API to implement this.

[yfinkelstein]

Makes sense. Other options are Observable Api in both rxJava and grpc. And streaming should be 2 way

On Apr 21, 2016, at 02:45, Trustin Lee notifications@github.com wrote:

Thinking of using reactive streams API to implement this.

—
You are receiving this because you authored the thread.
Reply to this email directly or view it on GitHub

[trustin]

@yfinkelstein You mean streaming both requests and responses, right?

[yfinkelstein]

Yes. One request one response, one req N responses, N requests one response, and N requests N responses - all combinations should be supported.

On Apr 22, 2016, at 4:25 AM, Trustin Lee notifications@github.com wrote:

@yfinkelstein You mean streaming both requests and responses, right?

—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub

[trustin]

@yfinkelstein Could you explain the case of 1:N, N:1, N:M?

[yfinkelstein]

Consider the use case of client interacting with a server that's fronting a portioned data storage cluster. The client sends a query "select ..".  The server forwards this request to N partitions and merges the results (for simplicity let's say the order does not matter). To avoid buffering the entire response from all partitions the server might prefer to send response to the client in chunks where a chunk can be the response from a single partition or a part of it. This gives is single request from client followed by N chunks (responses) from the server.
Reverse that and imagine that the client is uploading a large data set which it acquires from some source. The client might want to to divide the request in chunks and we get N requests followed by 1 response when the server is done.
NxM is a natural combination of both patterns.
gRPC really nailed this problem. Their API is really slick with Observable used both on the client and server.

On Saturday, April 23, 2016 5:53 PM, Trustin Lee <notifications@github.com> wrote:

@yfinkelstein Could you explain the case of 1:N, N:1, N:M?—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub

[trustin]

I see. Aren't they still one request - one response? It's just their contents that are being streamed.

[yfinkelstein]

Yes, content is split into multiple parts that are sent and received independently

Sent from my iPhone

On Apr 23, 2016, at 21:17, Trustin Lee notifications@github.com wrote:

I see. Aren't they still one request - one response? It's just their contents that are being streamed.

—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub

[yfinkelstein]

Finite streams

Sent from my iPhone

On Apr 23, 2016, at 21:21, Yuri Finkelstein yurif2003@yahoo.com wrote:

Yes, content is split into multiple parts that are sent and received independently

Sent from my iPhone

On Apr 23, 2016, at 21:17, Trustin Lee notifications@github.com wrote:

I see. Aren't they still one request - one response? It's just their contents that are being streamed.

—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub

[yfinkelstein]

Streaming is not the only important feature here. Regular rpc messages can be streamed too if the transport supports it. The main feature here is that stream parts are delivered separately at the api level.

Sent from my iPhone

On Apr 23, 2016, at 21:34, Yuri Finkelstein yurif2003@yahoo.com wrote:

Finite streams

Sent from my iPhone

On Apr 23, 2016, at 21:21, Yuri Finkelstein yurif2003@yahoo.com wrote:

Yes, content is split into multiple parts that are sent and received independently

Sent from my iPhone

On Apr 23, 2016, at 21:17, Trustin Lee notifications@github.com wrote:

I see. Aren't they still one request - one response? It's just their contents that are being streamed.

—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub

[trustin]

Spent some time this week about this, and here's the idea so far:

@FunctionalInterface
public interface Service<I, O> {
    Publisher<O> apply(RequestPublisher<I> request) throws Exception;
}

public interface RequestPublisher<T> extends Publisher<T> {
    // getters for the information available from the beginning of the stream
    // e.g. initial headers, stream ID, remote/local address
}

public class MyHttp2Service<Http2Object, Http2Object> {
    public Publisher<Http2Object> apply(RequestPublisher<Http2Object> request) {
        Publisher<Http2Object> response = new MyPublisher<Http2Object>();
        request.subscribe(new MySubscriber(response);
        return response;
    }

    private class MySubscriber implements Subscriber<Http2Object> {
        private Subscription subscription;
        public void onSubscribe(Subscription subscription) {
            this.subscription = subscription;
            subscription.request(1);
        }
        public void onNext(Http2Object o) {
            ...
            subscription.request(1);
        }
        public void onComplete() {
            response.publish(someHttp2ResponseObject);
        }
    }
}

public class LoggingService<IN, OUT> extends Service<IN, OUT> {
    private Service delegate;
    public Publisher<OUT> apply(Publisher<IN> request) {
        // *Processor are both a publisher and a subscriber.
        request.subscribe(requestLoggingProcessor);
        Publisher<Object> response = delegate.apply(requestLoggingProcessor);
        response.subscribe(responseLoggingProcessor);
        return responseLoggingProcessor;
    }
}

As you see, ServiceCodec and ServiceInvocationHandler are gone. It is likely that ServiceInvocationContext will require some changes as well. It's a lot of changes, but it seems rational to me. WDYT?

I'm currently focusing on the server-side but the client-side could change similarly. For example, ClientCodec.encodeRequest() would produce a Publisher and decodeResponse() would require a Publisher.

[yfinkelstein]

I'd like to understand the overall approach first. In gRPC, "stream" is a keyword in their IDL (protobuf based). The author of the RPC interface signals the nature of the message as being being either a single part or a stream of parts. The generated code will then reflect this. Is this your intention as well (with thrift of course)?

The code snippet above looks quite readable and conveys the idea well. What is your approach to back pressure management? It's one of the key issues that always comes with with streaming. The best idea known to me today is that back pressure must be managed explicitly and cooperatively by both sides rather than just rely on TCP back pressure on each side.
Am I reading it right that .request(1) is the back pressure control mechanism and '1' is the number of parts subscriber is asking producer to produce?

[trustin]

Am I reading it right that .request(1) is the back pressure control mechanism and '1' is the number of parts subscriber is asking producer to produce?

Yes. Back pressure control considered. :-)

[anuraaga]

As you say, this is just a step to be able to read HTTP streams partially rather than all up front. That would be enough for a custom HttpService to do streaming if it has a use case, but more interesting will be at that point we can implement a wire protocol to frame arbitrary RPC messages (we'll probably implement the grpc wire protocol initially with other options to increase support to browsers, etc), and then generated code will need to be able to work with the wire protocol. Since thrift would require significant work in the IDL compiler itself, I guess it's more likely the first shining example of streaming in armeria will be protobuf-based.

[trustin]

This feature has been implemented long time ago (0.21.0)