Armeria: gRPC deframer complexity

[trustin]

Is it only myself who finds ArmeriaMessageDeframer more complex than necessary? I think we can simplify it in the long term by writing a building block like the following:

Write a Reactive Streams Processor which decodes a stream of HttpObjects into another stream. It is different from FilteredStreamMessage in that it allows decoding M HttpObjects into N objects. This will provide some extension point like Netty's ByteToMessageDecoder does.

On top of this building block, we could rewrite ArmeriaDeframer with much less state management.

It may be possible to do something similar for ArmeriaMessageFramer, but I didn't think much about it yet.

[minwoox]

Write a Reactive Streams Processor which decodes a stream of HttpObjects into another stream.

We can use that for multi-part as well. 😉

[jrhee17]

Thinking of taking a stab at this (unless someone is already working on it)

[ikhoon]

I was about to start this issue ;-)
No matter who start it, let's start to talk about API design.

[jrhee17]

I know you're already working on multipart stuff, so I'll look for another issue.
I'll try to pop in for reviews instead 😄

[ikhoon]

I think we can think about two points.

1) Reorganize the existing elements.
In MultipartEncoder case, it does not need byte-level operations. It only needs a mapping function to publish N sequence.

class StreamMessageProcessor<T, U> implements StreamMessage<U>, Processor<T, U> {
   /* Flushes all buffered data (I'm not sure thess operation are required) */
   void flush();
   void flushBefore(T element);
   void flushUntil(T element);

   List<U> process(List<T> elements);
}

2) Deframe chunked bytes to messages.
ArmeriaMessageDeframer and MultipartDecoder(#2894) decodes HTTP payloads to concretes messages.

class HttpDataDeframer<T> implements StreamMessageProcessor<HttpData, T> {
    // Needs virtual buffer to access bytes across the `HttpData`s
    Queue<HttpData> ...
    List<U> process(HttpData data) {
       virtualBuffer.add(data);
       returns process()
    }
    abstract List<T> process(); 
}

And implement subclass like the following

class ArmeriaMessageDeframer<T> extends HttpDataDeframer<T> {

    List<T> process() {
        if (readableBytes() > 5) { 
            // Read data from buffers
            byte header = readByte();
            int lenght = readInt();
            ... 
        } else {
            // Wait more data
            return List.of();
        }
    }
}

[trustin]

In MultipartEncoder case, it does not need byte-level operations.

Really? In chunked transfer encoding, wouldn't be a boundary split into multiple chunks although rare?

[trustin]

Even if it's not chunked transfer encoding, HTTP decoders do not have a notion of boundaries, so they will be easily split into many pieces, so we need to do some byte-level aggregation.

[trustin]

We also need to try to reuse the HttpData in an input stream when producing an output stream to reduce unnecessary memory copies. For example, we could forward the input HttpData instance as it is when it is a partial content without any other stuff like boundaries and length headers. Could use WITH_POOLED_OBJECTS when necessary as well.

[trustin]

That being said, simple byte-level streaming abstraction wouldn't meet all the requirements. Maybe better keeping the implementation at object-level and providing some utilities for managing substates for working at byte level?

[ikhoon]

Really? In chunked transfer encoding, wouldn't be a boundary split into multiple chunks although rare?

That's a good point. I thought only files are to be considered in FilePublisher. https://github.com/ikhoon/armeria/commit/a505474fb088c536c9885b400c3414e136c85a4d#diff-cb3400de0d04cdb4c0a6ea2eb79807c2R133-R134.
But any body parts could be split into multiple chunks.

[ikhoon]

Fixed by #2981