New SSE implementation

rx-netty-contexts/src/main/java/io/reactivex/netty/contexts/ContextPipelineConfigurators.java

@@ -23,7 +23,7 @@
 import io.reactivex.netty.protocol.http.client.HttpClientResponse;
 import io.reactivex.netty.protocol.http.server.HttpServerRequest;
 import io.reactivex.netty.protocol.http.server.HttpServerResponse;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 
 /**
  * A factory class for different {@link PipelineConfigurator} for the context module.
@@ -61,12 +61,12 @@ private ContextPipelineConfigurators() {
     public static <I> PipelineConfigurator<HttpClientResponse<ServerSentEvent>, HttpClientRequest<I>>
     sseClientConfigurator(RequestIdProvider requestIdProvider, RequestCorrelator correlator) {
         return new HttpClientContextConfigurator<I, ServerSentEvent>(requestIdProvider, correlator,
-                                                                     PipelineConfigurators.<I>sseClientConfigurator());
+                                                                     PipelineConfigurators.<I>clientSseConfigurator());
     }
 
     public static <I> PipelineConfigurator<HttpServerRequest<I>, HttpServerResponse<ServerSentEvent>>
     sseServerConfigurator(RequestIdProvider requestIdProvider, RequestCorrelator correlator) {
         return new HttpServerContextConfigurator<I, ServerSentEvent>(requestIdProvider, correlator,
-                                                                     PipelineConfigurators.<I>sseServerConfigurator());
+                                                                     PipelineConfigurators.<I>serveSseConfigurator());
     }
 }

rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/logtail/LogAggregator.java

@@ -17,7 +17,7 @@
 package io.reactivex.netty.examples.http.logtail;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.logging.LogLevel;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.pipeline.PipelineConfigurators;
 import io.reactivex.netty.protocol.http.client.HttpClient;
@@ -27,8 +27,9 @@
 import io.reactivex.netty.protocol.http.server.HttpServerRequest;
 import io.reactivex.netty.protocol.http.server.HttpServerResponse;
 import io.reactivex.netty.protocol.http.server.RequestHandler;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import rx.Observable;
+import rx.functions.Action1;
 import rx.functions.Func1;
 
 import java.util.ArrayList;
@@ -53,7 +54,7 @@ public LogAggregator(int port, int producerPortFrom, int producerPortTo) {
     }
 
     public HttpServer<ByteBuf, ServerSentEvent> createAggregationServer() {
-        server = RxNetty.createHttpServer(port,
+        server = RxNetty.newHttpServerBuilder(port,
                 new RequestHandler<ByteBuf, ServerSentEvent>() {
                     @Override
                     public Observable<Void> handle(final HttpServerRequest<ByteBuf> request,
@@ -62,18 +63,11 @@ public Observable<Void> handle(final HttpServerRequest<ByteBuf> request,
                         return connectToLogProducers().flatMap(new Func1<ServerSentEvent, Observable<Void>>() {
                             @Override
                             public Observable<Void> call(ServerSentEvent sse) {
-                                ServerSentEvent data = new ServerSentEvent(sse.getEventId(), "data", sse.getEventData());
-                                return response.writeAndFlush(data);
-                            }
-                        }).onErrorResumeNext(new Func1<Throwable, Observable<Void>>() {
-                            @Override
-                            public Observable<Void> call(Throwable throwable) {
-                                response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
-                                return response.close();
+                                return response.writeAndFlush(sse);
                             }
                         });
                     }
-                }, PipelineConfigurators.<ByteBuf>sseServerConfigurator());
+                }).enableWireLogging(LogLevel.ERROR).pipelineConfigurator(PipelineConfigurators.<ByteBuf>serveSseConfigurator()).build();
         System.out.println("Logs aggregator server started...");
         return server;
     }
@@ -88,12 +82,18 @@ private Observable<ServerSentEvent> connectToLogProducers() {
 
     private static Observable<ServerSentEvent> connectToLogProducer(int port) {
         HttpClient<ByteBuf, ServerSentEvent> client =
-                RxNetty.createHttpClient("localhost", port, PipelineConfigurators.<ByteBuf>sseClientConfigurator());
+                RxNetty.createHttpClient("localhost", port, PipelineConfigurators.<ByteBuf>clientSseConfigurator());
 
         return client.submit(HttpClientRequest.createGet("/logstream")).flatMap(new Func1<HttpClientResponse<ServerSentEvent>, Observable<ServerSentEvent>>() {
             @Override
             public Observable<ServerSentEvent> call(HttpClientResponse<ServerSentEvent> response) {
-                return response.getContent();
+                return response.getContent()
+                               .doOnNext(new Action1<ServerSentEvent>() {
+                                   @Override
+                                   public void call(ServerSentEvent serverSentEvent) {
+                                       serverSentEvent.retain();
+                                   }
+                               });
             }
         });
     }

rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/logtail/LogProducer.java

@@ -17,13 +17,14 @@
 package io.reactivex.netty.examples.http.logtail;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.handler.logging.LogLevel;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.pipeline.PipelineConfigurators;
 import io.reactivex.netty.protocol.http.server.HttpServer;
 import io.reactivex.netty.protocol.http.server.HttpServerRequest;
 import io.reactivex.netty.protocol.http.server.HttpServerResponse;
 import io.reactivex.netty.protocol.http.server.RequestHandler;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import rx.Observable;
 import rx.functions.Func1;
 
@@ -45,14 +46,16 @@ public LogProducer(int port, int interval) {
     }
 
     public HttpServer<ByteBuf, ServerSentEvent> createServer() {
-        HttpServer<ByteBuf, ServerSentEvent> server = RxNetty.createHttpServer(port,
+        HttpServer<ByteBuf, ServerSentEvent> server = RxNetty.newHttpServerBuilder(port,
                 new RequestHandler<ByteBuf, ServerSentEvent>() {
                     @Override
                     public Observable<Void> handle(HttpServerRequest<ByteBuf> request,
                                                    HttpServerResponse<ServerSentEvent> response) {
                         return createReplyHandlerObservable(response);
                     }
-                }, PipelineConfigurators.<ByteBuf>sseServerConfigurator());
+                }).pipelineConfigurator(PipelineConfigurators.<ByteBuf>serveSseConfigurator())
+                  .enableWireLogging(LogLevel.DEBUG)
+                  .build();
         System.out.println("Started log producer on port " + port);
         return server;
     }
@@ -62,12 +65,10 @@ private Observable<Void> createReplyHandlerObservable(final HttpServerResponse<S
                 .flatMap(new Func1<Long, Observable<Void>>() {
                     @Override
                     public Observable<Void> call(Long interval) {
-                        ServerSentEvent data = new ServerSentEvent(
-                                Long.toString(interval),
-                                "data",
-                                LogEvent.randomLogEvent(source).toCSV()
-                        );
-                        return response.writeAndFlush(data);
+                        ByteBuf eventId = response.getAllocator().buffer().writeLong(interval);
+                        ByteBuf data = response.getAllocator().buffer().writeBytes(LogEvent.randomLogEvent(
+                                source).toCSV().getBytes());
+                        return response.writeAndFlush(ServerSentEvent.withEventId(eventId, data));
                     }
                 });
     }

rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/logtail/LogTailClient.java

@@ -18,12 +18,13 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.logging.LogLevel;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.pipeline.PipelineConfigurators;
 import io.reactivex.netty.protocol.http.client.HttpClient;
 import io.reactivex.netty.protocol.http.client.HttpClientRequest;
 import io.reactivex.netty.protocol.http.client.HttpClientResponse;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import rx.Observable;
 import rx.functions.Func1;
 
@@ -49,7 +50,9 @@ public LogTailClient(int port, int tailSize) {
 
     public List<LogEvent> collectEventLogs() {
         HttpClient<ByteBuf, ServerSentEvent> client =
-                RxNetty.createHttpClient("localhost", port, PipelineConfigurators.<ByteBuf>sseClientConfigurator());
+                RxNetty.<ByteBuf, ServerSentEvent>newHttpClientBuilder("localhost", port)
+                       .enableWireLogging(LogLevel.DEBUG)
+                       .pipelineConfigurator(PipelineConfigurators.<ByteBuf>clientSseConfigurator()).build();
 
         Iterable<LogEvent> eventIterable = client.submit(HttpClientRequest.createGet("/logstream"))
                 .flatMap(new Func1<HttpClientResponse<ServerSentEvent>, Observable<ServerSentEvent>>() {
@@ -63,7 +66,7 @@ public Observable<ServerSentEvent> call(HttpClientResponse<ServerSentEvent> resp
                 }).map(new Func1<ServerSentEvent, LogEvent>() {
                            @Override
                            public LogEvent call(ServerSentEvent serverSentEvent) {
-                               return LogEvent.fromCSV(serverSentEvent.getEventData());
+                               return LogEvent.fromCSV(serverSentEvent.contentAsString());
                            }
                        }
                 ).filter(new Func1<LogEvent, Boolean>() {

rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/sse/HttpSseClient.java

@@ -22,7 +22,7 @@
 import io.reactivex.netty.protocol.http.client.HttpClient;
 import io.reactivex.netty.protocol.http.client.HttpClientRequest;
 import io.reactivex.netty.protocol.http.client.HttpClientResponse;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import rx.Observable;
 import rx.functions.Func1;
 
@@ -48,7 +48,7 @@ public HttpSseClient(int port, int noOfEvents) {
 
     public List<ServerSentEvent> readServerSideEvents() {
         HttpClient<ByteBuf, ServerSentEvent> client =
-                RxNetty.createHttpClient("localhost", port, PipelineConfigurators.<ByteBuf>sseClientConfigurator());
+                RxNetty.createHttpClient("localhost", port, PipelineConfigurators.<ByteBuf>clientSseConfigurator());
 
         Iterable<ServerSentEvent> eventIterable = client.submit(HttpClientRequest.createGet("/hello")).
                 flatMap(new Func1<HttpClientResponse<ServerSentEvent>, Observable<ServerSentEvent>>() {

rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/sse/HttpSseServer.java

@@ -23,7 +23,7 @@
 import io.reactivex.netty.protocol.http.server.HttpServerRequest;
 import io.reactivex.netty.protocol.http.server.HttpServerResponse;
 import io.reactivex.netty.protocol.http.server.RequestHandler;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import rx.Notification;
 import rx.Observable;
 import rx.functions.Func1;
@@ -54,7 +54,7 @@ public Observable<Void> handle(HttpServerRequest<ByteBuf> request,
                                                    HttpServerResponse<ServerSentEvent> response) {
                         return getIntervalObservable(response);
                     }
-                }, PipelineConfigurators.<ByteBuf>sseServerConfigurator());
+                }, PipelineConfigurators.<ByteBuf>serveSseConfigurator());
         System.out.println("HTTP Server Sent Events server started...");
         return server;
     }
@@ -65,7 +65,9 @@ private Observable<Void> getIntervalObservable(final HttpServerResponse<ServerSe
                     @Override
                     public Observable<Void> call(Long interval) {
                         System.out.println("Writing SSE event for interval: " + interval);
-                        return response.writeAndFlush(new ServerSentEvent(String.valueOf(interval), "notification", "hello " + interval));
+                        ByteBuf data = response.getAllocator().buffer().writeBytes(("hello " + interval).getBytes());
+                        ServerSentEvent event = new ServerSentEvent(data);
+                        return response.writeAndFlush(event);
                     }
                 }).materialize()
                 .takeWhile(new Func1<Notification<Void>, Boolean>() {

rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/logtail/LogTailClientTest.java

@@ -19,7 +19,7 @@
 import io.netty.buffer.ByteBuf;
 import io.reactivex.netty.examples.ExamplesEnvironment;
 import io.reactivex.netty.protocol.http.server.HttpServer;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -28,8 +28,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static io.reactivex.netty.examples.http.logtail.LogTailClient.DEFAULT_TAIL_SIZE;
 import static io.reactivex.netty.examples.http.logtail.LogAggregator.DEFAULT_AG_PORT;
+import static io.reactivex.netty.examples.http.logtail.LogTailClient.DEFAULT_TAIL_SIZE;
 
 /**
  * @author Tomasz Bak
@@ -41,7 +41,7 @@ public class LogTailClientTest extends ExamplesEnvironment {
     private static final int PR_INTERVAL = 50;
 
     private HttpServer<ByteBuf, ServerSentEvent> aggregationServer;
-    private List<HttpServer<ByteBuf, ServerSentEvent>> producerServers = new ArrayList<HttpServer<ByteBuf, ServerSentEvent>>();
+    private final List<HttpServer<ByteBuf, ServerSentEvent>> producerServers = new ArrayList<HttpServer<ByteBuf, ServerSentEvent>>();
 
     @Before
     public void setupServers() {

rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/sse/HttpSseTest.java

@@ -19,7 +19,7 @@
 import io.netty.buffer.ByteBuf;
 import io.reactivex.netty.examples.ExamplesEnvironment;
 import io.reactivex.netty.protocol.http.server.HttpServer;
-import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
+import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;

rx-netty/src/main/java/io/reactivex/netty/channel/ChannelWriter.java

@@ -16,6 +16,7 @@
 
 package io.reactivex.netty.channel;
 
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.FileRegion;
 import rx.Observable;
@@ -33,6 +34,8 @@ public interface ChannelWriter<O> {
 
     <R> void write(R msg, ContentTransformer<R> transformer);
 
+    void writeBytes(ByteBuf msg);
+
     void writeBytes(byte[] msg);
 
     void writeString(String msg);
@@ -47,6 +50,8 @@ public interface ChannelWriter<O> {
 
     <R> Observable<Void> writeAndFlush(R msg, ContentTransformer<R> transformer);
 
+    Observable<Void> writeBytesAndFlush(ByteBuf msg);
+
     Observable<Void> writeBytesAndFlush(byte[] msg);
 
     Observable<Void> writeStringAndFlush(String msg);

rx-netty/src/main/java/io/reactivex/netty/channel/DefaultChannelWriter.java

@@ -72,6 +72,12 @@ public <R> Observable<Void> writeAndFlush(final R msg, final ContentTransformer<
         return flush();
     }
 
+    @Override
+    public Observable<Void> writeBytesAndFlush(ByteBuf msg) {
+        writeBytes(msg);
+        return flush();
+    }
+
     @Override
     public void write(O msg) {
         writeOnChannel(msg);
@@ -83,9 +89,14 @@ public <R> void write(R msg, ContentTransformer<R> transformer) {
         writeOnChannel(contentBytes);
     }
 
+    @Override
+    public void writeBytes(ByteBuf msg) {
+        write(msg, IdentityTransformer.DEFAULT_INSTANCE);
+    }
+
     @Override
     public void writeBytes(byte[] msg) {
-        write(msg, new ByteTransformer());
+        write(msg, ByteTransformer.DEFAULT_INSTANCE);
     }
 
     @Override
@@ -95,7 +106,7 @@ public void writeString(String msg) {
 
     @Override
     public Observable<Void> writeBytesAndFlush(byte[] msg) {
-        write(msg, new ByteTransformer());
+        write(msg, ByteTransformer.DEFAULT_INSTANCE);
         return flush();
     }
 

rx-netty/src/main/java/io/reactivex/netty/channel/IdentityTransformer.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.reactivex.netty.channel;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+
+/**
+ * An identity transformer that returns the passed {@link ByteBuf}
+ *
+ * @author Nitesh Kant
+ */
+public class IdentityTransformer implements ContentTransformer<ByteBuf> {
+
+    public static final IdentityTransformer DEFAULT_INSTANCE = new IdentityTransformer();
+
+    @Override
+    public ByteBuf call(ByteBuf toTransform, ByteBufAllocator allocator) {
+        return toTransform;
+    }
+}

rx-netty/src/main/java/io/reactivex/netty/channel/SingleNioLoopProvider.java

@@ -110,7 +110,8 @@ private EpollEventLoopGroup getNativeParentEventLoop() {
 
         EpollEventLoopGroup eventLoopGroup = nativeParentEventLoop.get();
         if (null == eventLoopGroup) {
-            EpollEventLoopGroup newEventLoopGroup = new EpollEventLoopGroup(parentEventLoopCount);
+            EpollEventLoopGroup newEventLoopGroup = new EpollEventLoopGroup(parentEventLoopCount,
+                                                                            new RxDefaultThreadFactory("rx-netty-epoll-eventloop"));
             if (!nativeParentEventLoop.compareAndSet(null, newEventLoopGroup)) {
                 newEventLoopGroup.shutdownGracefully();
             }
@@ -121,7 +122,8 @@ private EpollEventLoopGroup getNativeParentEventLoop() {
     private EpollEventLoopGroup getNativeEventLoop() {
         EpollEventLoopGroup eventLoopGroup = nativeEventLoop.get();
         if (null == eventLoopGroup) {
-            EpollEventLoopGroup newEventLoopGroup = new EpollEventLoopGroup(childEventLoopCount);
+            EpollEventLoopGroup newEventLoopGroup = new EpollEventLoopGroup(childEventLoopCount,
+                                                                            new RxDefaultThreadFactory("rx-netty-epoll-eventloop"));
             if (!nativeEventLoop.compareAndSet(null, newEventLoopGroup)) {
                 newEventLoopGroup.shutdownGracefully();
             }
@@ -134,11 +136,11 @@ public static class SharedNioEventLoopGroup extends NioEventLoopGroup {
         private final AtomicInteger refCount = new AtomicInteger();
 
         public SharedNioEventLoopGroup() {
-            super(0, new RxDefaultThreadFactory("rx-selector"));
+            super(0, new RxDefaultThreadFactory("rx-netty-nio-eventloop"));
         }
 
         public SharedNioEventLoopGroup(int threadCount) {
-            super(threadCount, new RxDefaultThreadFactory("rx-selector"));
+            super(threadCount, new RxDefaultThreadFactory("rx-netty-nio-eventloop"));
         }
 
         @Override