A ByteStream abstraction to represent streaming content. #298

mikkokar · 2018-10-03T09:34:20Z

Raising this just to get some preliminary comments for this work.

The PR adds a ByteStream class that abstracts a live asynchronous content stream. The ByteStream implements a reactive-streams Publisher interface and is implemented on top of reactor-core Flux.
Netty ByteBuf has been encapsulated inside a new Styx Buffer class.

kvosper · 2018-10-03T10:25:07Z

components/api/src/main/java/com/hotels/styx/api/AggregateOperator.java

+            this.upstream.subscribe(this);
+            return future;
+        } else {
+            throw new IllegalStateException("Secondary subscription!");


Perhaps a message like "This operator may only be subscribed to once" would be clearer.

I will change to:

throw new IllegalStateException("ByteStreamAggregator may only be started once.");

And for what it is worth I have changed the class name to ByteStreamAggregator. As this is not really an operator. It is a consumer of a byte stream, not an operator that modifies the stream.

kvosper · 2018-10-03T10:31:09Z

components/api/src/main/java/com/hotels/styx/api/AggregateOperator.java

+            this.subscription.request(Long.MAX_VALUE);
+        } else {
+            subscription.cancel();
+            throw new IllegalStateException("Second onSubscribe event to AggregateOperator");


Looks like we should use the same exception + message as the other method (create a throwSubscriptionException method or something?)

Changed to throw new IllegalStateException("ByteStreamAggregator supports only one Producer instance.");

kvosper · 2018-10-03T10:53:03Z

components/api/src/main/java/com/hotels/styx/api/Buffer.java

+public final class Buffer {
+    private final ByteBuf delegate;
+
+    public Buffer(ByteBuf byteBuf) {


By having this constructor, netty ByteBuf is still visible in our API. I'm not sure what the solution is.

Yeah. I still need to think about this.

kvosper · 2018-10-03T13:05:40Z

components/api/src/main/java/com/hotels/styx/api/HttpResponse.java

@@ -59,27 +50,27 @@
 * chunk must be processed as they arrive. It is also useful for dealing with
 * very large content sizes, and in situations where content size is not known
 * upfront.
- * <p>
+ * <p>z


kvosper · 2018-10-03T13:10:20Z

components/api/src/test/java/com/hotels/styx/api/ByteStreamTest.java

+    }
+
+    @Test
+    public void publisherBackpressure() {


Not a proper test name.

supportsReactiveBackpressure?

supportsBackpressure?

components/api/src/test/java/com/hotels/styx/api/ByteStreamTest.java

kvosper · 2018-10-03T13:43:07Z

components/client/src/main/java/com/hotels/styx/client/Transport.java

@@ -107,7 +108,7 @@ private synchronized void returnIfConnected(Optional<ConnectionPool> connectionP
        return origin
                .map(ConnectionPool::borrowConnection)
                .orElseGet(() -> {
-                    request.releaseContentBuffers();
+                    toObservable(request.body()).forEach(it -> it.delegate().release());


Should this use request.body().drop() now?

The way to do this now is:

// Guaranteed to aggregate an empty body: request.body().drop().aggregate(1);

kvosper · 2018-10-09T09:53:38Z

components/api/src/main/java/com/hotels/styx/api/Buffer.java

+import static io.netty.buffer.Unpooled.copiedBuffer;
+import static java.util.Objects.requireNonNull;
+
+/**


Comment is mostly telling us things we already know - that we're in Styx, that classes are a form of abstraction, etc.

I think we should avoid saying "A Styx byte buffer" and just say "A byte buffer". Otherwise it seems like we're implying that we have some kind of unique (unspecified) byte buffer behaviour. To us, it seems important because we're differentiating it from Netty ByteBuf, but someone looking at the docs for the first time wouldn't be thinking about that (and can see that it is a Styx class from the package name).

👍 Makes sense.

components/api/src/main/java/com/hotels/styx/api/ByteStream.java

components/api/src/main/java/com/hotels/styx/api/ByteStreamAggregator.java

kvosper · 2018-10-09T10:23:37Z

components/api/src/main/java/com/hotels/styx/api/HttpResponse.java

            this.status = statusWithCode(response.status().code());
            this.version = httpVersion(response.version().toString());
            this.headers = response.headers().newBuilder();
-            this.body = contentStream;
+            this.body = byteStream;


Is byteStream allowed to be null?

No. Added the null checks in Builder. Do you think it still warrants one here?

components/api/src/test/java/com/hotels/styx/api/ByteStreamTest.java

components/server/src/main/java/com/hotels/styx/server/netty/connectors/HttpResponseWriter.java

All tests pass.

Enable previously commented out tests.

Add ContentStream Events.

…ream classes.

…tor. Add a file for ByteStream class.

Tidy up.

…Java objects. Use the new ByteStream class in HTTP message objects.

RxReactiveStreams bridges.

implementations.

ByteStream: Rename discard() to drop(). Fix HttpResponseTest: canRemoveResponseBody.

Change AggregateOperator name to ByteStreamAggregator.

…fter aggregation has finished. Remove HttpMessageBodies class.

components/api/src/main/java/com/hotels/styx/api/HttpResponse.java

...onents/server/src/main/java/com/hotels/styx/server/netty/connectors/HttpPipelineHandler.java

components/server/src/main/java/com/hotels/styx/server/netty/connectors/HttpResponseWriter.java

dvlato · 2018-10-10T11:09:46Z

components/proxy/src/main/java/com/hotels/styx/routing/handlers/StaticResponseHandler.java

@@ -46,7 +51,7 @@ private StaticResponseHandler(int status, String text) {

    @Override
    public StyxObservable<HttpResponse> handle(HttpRequest request, HttpInterceptor.Context context) {
-        return StyxObservable.of(HttpResponse.response(statusWithCode(status)).body(StyxObservable.of(text), UTF_8).build());
+        return StyxObservable.of(response(statusWithCode(status)).body(new ByteStream(toPublisher(Observable.just(new Buffer(text, UTF_8))))).build());


Can't we use Flux.just here?

Looks like Flux can be used here.

dvlato · 2018-10-10T11:10:30Z

support/origins-starter-app/src/main/java/com/hotels/styx/support/origins/AppHandler.java

@@ -57,11 +61,11 @@ public AppHandler(Origin origin) {
                    request.queryParam("status").ifPresent(status ->
                            responseBuilder
                                    .status(httpResponseStatus(status))
-                                    .body(StyxObservable.of("Returning requested status (" + status + ")"), UTF_8)
+                                    .body(new ByteStream(toPublisher(Observable.just(new Buffer("Returning requested status (" + status + ")", UTF_8)))))


As in other parts of coude, Is there any reason (visibility?) why we cannot use Flux.just() instead of creating an rxJava Observable with .just() and transforming it to a publisher ?

Yeah it looks it is OK to use Flux.

I didn't want to add the comment in every occurrence but you can search for Observable.just()...

dvlato · 2018-10-10T11:11:05Z

system-tests/e2e-suite/src/test/scala/com/hotels/styx/plugins/AggregatingPluginSpec.scala

@@ -65,7 +65,7 @@ class AggregatingPluginSpec extends FunSpec

    it("Gets response from aggregating plugin (with body)") {
      mockServer.stub("/body", responseSupplier(
-        () => response(OK).body(StyxObservable.from(Seq(chunk("a"), chunk("b"), chunk("c"), chunk("d"), chunk("e")).asJava)).build()
+        () => response(OK).body(new ByteStream(RxReactiveStreams.toPublisher(Observable.from(Seq(chunk("a"), chunk("b"), chunk("c"), chunk("d"), chunk("e")).asJava)))).build()


static import?

dvlato · 2018-10-10T11:11:28Z

system-tests/e2e-suite/src/test/scala/com/hotels/styx/client/OriginClosesConnectionSpec.scala

@@ -127,7 +126,7 @@ class OriginClosesConnectionSpec extends FunSuite
        .toStreamingRequest)

    responseObservable
-      .doOnNext((t: HttpResponse) => toRxObservable(t.body()).subscribe(contentSubscriber))
+      .doOnNext((t: HttpResponse) => RxReactiveStreams.toObservable(t.body()).subscribe(contentSubscriber))


static import?

dvlato · 2018-10-10T11:11:37Z

system-tests/e2e-suite/src/test/scala/com/hotels/styx/plugins/AsyncResponseContentSpec.scala

@@ -88,14 +85,14 @@ class AsyncDelayPlugin extends PluginAdapter {
    chain.proceed(request)
      .flatMap(asJavaFunction((response: HttpResponse) => {

-        val transformedContent: Observable[ByteBuf] = toRxObservable(response.body())
+        val transformedContent: Observable[Buffer] = RxReactiveStreams.toObservable(response.body())


static import?

kvosper · 2018-10-10T12:40:07Z

components/api/src/main/java/com/hotels/styx/api/ByteStream.java

+     * <ul>
+     * <li> When the mapping function returns a new {@link Buffer} instance, the reference count for
+     * the old one is automatically decremented.</li>
+     * <li> When the mapping function modifies the {@link Buffer} in place, returning the same instance


Our Buffer class is currently immutable, so this description may be confusing.

I would leave this open for now. Performance-wise it would be sensible to allow in-place modifications for the buffer. That would require further modifications to the Buffer class. WDYT?

mikkokar requested review from kvosper, dvlato and VivianLopes October 3, 2018 09:34

kvosper reviewed Oct 3, 2018

View reviewed changes

components/api/src/test/java/com/hotels/styx/api/ByteStreamTest.java Show resolved Hide resolved

kvosper reviewed Oct 3, 2018

View reviewed changes

mikkokar force-pushed the content-stream branch 2 times, most recently from bb2efca to 61a88e1 Compare October 5, 2018 13:58

kvosper suggested changes Oct 9, 2018

View reviewed changes

mikkokar added 17 commits October 10, 2018 07:36

Add ContentStream interface and an RxContentStream implementation.

c7558e9

All tests pass.

Add ContentStream.map method.

cf88b4a

Enable previously commented out tests.

Add consume method to ContentStream.

9b2a52c

Add ContentStream Events.

Implement Reactive Streams RxContentPublisher.

40a909f

Implement ContentAggregator, RxContentConsumer for the new content st…

e848df4

…ream classes.

Tidy up. Add copyright notices to new files. Etc.

e9c2d79

Add DiscardOperator. Rename ContentAggregator class to AggregateOpera…

f53ed22

…tor. Add a file for ByteStream class.

Add ByteStream and MappingOperator classes.

7586320

Refactor DiscardOperator (to resemble MappingOperator).

198bb1b

Tidy up.

Disable checkstyle for RxContentPublisher and RxContentStream FSMs.

72fab60

Use rxjava-reactive-streams to bridge between reactive streams and Rx…

4cd25eb

…Java objects. Use the new ByteStream class in HTTP message objects.

Deprecate RxContentPublisher and RxContentConsumer classes in favour of

1648e0e

RxReactiveStreams bridges.

Use Reactor-Core in Styx-API HTTP message objects and API tests.

de4e99e

Replace DiscardOperator and MappingOperator with corresponding reactor

594e109

implementations.

Make ByteStream to implement Publisher<Buffer> API.

1fde3b0

Builder setter methods for HTTP message body.

2b2687b

ByteStream: Add end-of-stream future.

f11a4b9

ByteStream: Rename discard() to drop(). Fix HttpResponseTest: canRemoveResponseBody.

mikkokar added 7 commits October 10, 2018 07:41

ByteStream: doOnEnd action.

4fe01a0

Move ByteSream into api package.

4b676e7

Tidy up pom files.

5ae13c9

Address code review comments.

65824b6

Change AggregateOperator name to ByteStreamAggregator.

Hide Netty ByteBuf from the Buffer class.

e7dbe9e

HttpRequest/Response .toFull method: Release aggregated byte buffer a…

85f2133

…fter aggregation has finished. Remove HttpMessageBodies class.

Resolved some low-hanging code review comments.

5c54c66

mikkokar force-pushed the content-stream branch from 61a88e1 to 5c54c66 Compare October 10, 2018 07:18

Add Javadocs to Buffer and ByteStream classes.

d589190

dvlato reviewed Oct 10, 2018

View reviewed changes

components/api/src/main/java/com/hotels/styx/api/HttpResponse.java Show resolved Hide resolved

dvlato reviewed Oct 10, 2018

View reviewed changes

...onents/server/src/main/java/com/hotels/styx/server/netty/connectors/HttpPipelineHandler.java Outdated Show resolved Hide resolved

Add static import.

d5bd3b8

dvlato reviewed Oct 10, 2018

View reviewed changes

components/server/src/main/java/com/hotels/styx/server/netty/connectors/HttpResponseWriter.java Show resolved Hide resolved

dvlato reviewed Oct 10, 2018

View reviewed changes

Address code review comments.

1515026

kvosper suggested changes Oct 10, 2018

View reviewed changes

dvlato approved these changes Oct 10, 2018

View reviewed changes

Improve JavaDoc for Buffer.java.

0eb4ddc

kvosper approved these changes Oct 10, 2018

View reviewed changes

mikkokar merged commit 5d9b79c into ExpediaGroup:master Oct 11, 2018

mikkokar deleted the content-stream branch October 11, 2018 08:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

A ByteStream abstraction to represent streaming content. #298

A ByteStream abstraction to represent streaming content. #298

mikkokar commented Oct 3, 2018 •

edited

Loading

kvosper Oct 3, 2018

mikkokar Oct 3, 2018

kvosper Oct 3, 2018

mikkokar Oct 3, 2018

kvosper Oct 3, 2018

mikkokar Oct 3, 2018

kvosper Oct 3, 2018

kvosper Oct 3, 2018

mikkokar Oct 3, 2018

mikkokar Oct 3, 2018

kvosper Oct 3, 2018

mikkokar Oct 3, 2018

kvosper Oct 9, 2018

kvosper Oct 10, 2018

mikkokar Oct 10, 2018

kvosper Oct 9, 2018

mikkokar Oct 10, 2018

dvlato Oct 10, 2018

mikkokar Oct 10, 2018

dvlato Oct 10, 2018

mikkokar Oct 10, 2018

dvlato Oct 10, 2018

dvlato Oct 10, 2018

dvlato Oct 10, 2018

dvlato Oct 10, 2018

kvosper Oct 10, 2018

mikkokar Oct 10, 2018 •

edited

Loading

A ByteStream abstraction to represent streaming content. #298

A ByteStream abstraction to represent streaming content. #298

Conversation

mikkokar commented Oct 3, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mikkokar Oct 10, 2018 • edited Loading

Choose a reason for hiding this comment

mikkokar commented Oct 3, 2018 •

edited

Loading

mikkokar Oct 10, 2018 •

edited

Loading