[1.0] Ensure that reference counting works well with the new API.

[kvosper]

The problem

Styx uses Netty reference counted ByteBuf for content streams. Netty reference counts are exposed to Styx consumers.

Styx should protect its consumers from accidentally leaking byte buffers.

The dangerous areas are

• Removing a body content stream
• Transforming the content stream

Detailed Description

Consider this code snippet to transform an UTF8 HTTP content stream to UTF-16. It leaks both buffers buf1A and buf1B. The buffers leak because the content transformation function, implemented as lambda, never releases them.

    ByteBuf buf1A = copiedBuffer("buf-1a", UTF_8);
    ByteBuf buf1B = copiedBuffer("buf-1b", UTF_8);
    
    // The HTTP request body as UTF 8 encoded byte stream:
    StyxObservable<ByteBuf> utf8Body = StyxObservable.from(ImmutableList.of(buf1A, buf1B));
    
    HttpRequest req1 = HttpRequest.put("/")
            .body(utf8Body)
            .build();

    // The UTF 8 response body transformed to UTF-16:
    StyxObservable<ByteBuf> utf16Body = req1.body().map(buf -> copiedBuffer(buf.toString(UTF_8), UTF_16));
    
    FullHttpRequest req2 = req1.newBuilder()
            .body(utf16Body)
            .build()
            .toFullRequest(100)
            .asCompletableFuture()
            .get();
    
    System.out.println("result: " + Arrays.toString(r2.body()));
    System.out.println("buf1 refcount: " + buf1A.refCnt());
    System.out.println("buf1 refcount: " + buf1B.refCnt());

Acceptance criteria

• Ideally, the Netty ByteBuf is wrapped with Styx facade to avoid leaking 3rd party dependencies in the API.
• Styx to take care of, or provide some helper methods, to automatically release the reference counts.

[mikkokar]

One option is to implement a pair of release methods:

static Function<ByteBuf, ByteBuf> release(
        Function<ByteBuf, ByteBuf> usersMapper)
{
    return input -> {
        ByteBuf output= usersMapper.apply(input);
        if (output != input) {
            input.release();
        }
        return output;
    };
}

static Function<ByteBuf, StyxObservable<ByteBuf>> releaseAsync(
        Function<ByteBuf, StyxObservable<ByteBuf>> usersMapper)
{
    return input ->
            usersMapper.apply(input)
                    .map(output -> {
                        if (output != input) {
                            input.release();
                        }
                        return output;
                    });
}

They could be used in the content transformation like this:

    // The UTF 8 response body transformed to UTF-16:
    StyxObservable<ByteBuf> utf16Body = r1.body()
            .map(release(buf -> copiedBuffer(buf.toString(UTF_8), UTF_16)));

[kvosper]

Closing all issues over 3 years old. New issues can be created if problems are still occurring.