Upgrade to RxJava 1.0

- and remove Javascript as it distracts from the purpose

Author: benjchristensen

reactive-lab-edge/build.gradle

@@ -2,10 +2,14 @@ sourceCompatibility = JavaVersion.VERSION_1_8
 targetCompatibility = JavaVersion.VERSION_1_8
 
 dependencies {
-	compile 'com.netflix.rxnetty:rx-netty:0.3.3'
-	compile 'com.netflix.rxjava:rxjava-core:0.18.2'
-	compile 'com.netflix.hystrix:hystrix-core:1.4.0-RC4'
-	compile 'com.netflix.hystrix:hystrix-metrics-event-stream:1.4.0-RC4'
+	compile 'com.netflix.rxnetty:rx-netty:0.3.18'
+	compile 'io.reactivex:rxjava:1.0.0-rc.10'
+	compile('com.netflix.hystrix:hystrix-core:1.4.0-RC5') {
+        exclude group: 'com.netflix.rxjava', module: 'rxjava-core' 
+    }
+	compile('com.netflix.hystrix:hystrix-metrics-event-stream:1.4.0-RC5') {
+        exclude group: 'com.netflix.rxjava', module: 'rxjava-core'  
+    }
 	compile 'org.codehaus.jackson:jackson-core-asl:1.9.13'
 	compile 'org.codehaus.jackson:jackson-mapper-asl:1.9.13'
     testCompile 'junit:junit-dep:4.10'

reactive-lab-edge/src/main/java/io/reactivex/lab/edge/APIServiceLayer.java

@@ -10,7 +10,7 @@
 public class APIServiceLayer {
 
     public Observable<String> getData() {
-        return Observable.from("one", "two", "three");
+        return Observable.just("one", "two", "three");
     }
 
     public String hello(String name) {

reactive-lab-edge/src/main/java/io/reactivex/lab/edge/EdgeServer.java

@@ -1,7 +1,6 @@
 package io.reactivex.lab.edge;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.reactivex.lab.edge.routes.RouteForDeviceHome;
 import io.reactivex.lab.edge.routes.mock.TestRouteBasic;
@@ -11,15 +10,11 @@
 import io.reactivex.netty.pipeline.PipelineConfigurators;
 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.serialization.ContentTransformer;
 
 import java.util.concurrent.TimeUnit;
 
 import rx.Observable;
 import rx.Subscriber;
-import rx.functions.Func0;
-import rx.operators.OperatorDefer;
 import rx.subscriptions.Subscriptions;
 
 import com.netflix.hystrix.HystrixRequestLog;
@@ -43,7 +38,7 @@ public static void main(String... args) {
         // start web services => http://localhost:8080
         RxNetty.createHttpServer(8080, (request, response) -> {
             System.out.println("Server => Request: " + request.getPath());
-            return defer(() -> {
+            return Observable.defer(() -> {
                 HystrixRequestContext.initializeContext();
                 try {
                     return handleRoutes(request, response);
@@ -79,9 +74,6 @@ private static Observable<Void> handleRoutes(HttpServerRequest<ByteBuf> request,
             return TestRouteWithSimpleFaultTolerance.handle(request, response);
         } else if (request.getPath().equals("/testWithHystrix")) {
             return TestRouteWithHystrix.handle(request, response);
-        } else if (request.getPath().endsWith(".js")) {
-            System.out.println("Server => Javascript Request: " + request.getPath());
-            return JavascriptRuntime.getInstance().handleRequest(request, response);
         } else {
             return writeError(request, response, "Unknown path: " + request.getPath());
         }
@@ -93,13 +85,13 @@ private static void startHystrixMetricsStream() {
             response.getHeaders().add("content-type", "text/event-stream");
             return Observable.create((Subscriber<? super Void> s) -> {
                 s.add(streamPoller.subscribe(json -> {
-                    response.writeAndFlush(new ServerSentEvent("", "data", json));
+                    response.writeStringAndFlush("data: " + json);
                 }, error -> {
                     s.onError(error);
                 }));
 
                 s.add(Observable.interval(1000, TimeUnit.MILLISECONDS).flatMap(n -> {
-                    return response.writeAndFlush(new ServerSentEvent("", "ping", ""))
+                    return response.writeStringAndFlush("ping:")
                             .onErrorReturn(e -> {
                                 System.out.println("Connection closed, unsubscribing from Hystrix Stream");
                                 s.unsubscribe();
@@ -132,25 +124,4 @@ public static Observable<Void> writeError(HttpServerRequest<?> request, HttpServ
         return response.writeStringAndFlush("Error 500: " + message);
     }
 
-    public static SSETransformer SSE_TRANSFORMER = new SSETransformer();
-
-    private static class SSETransformer implements ContentTransformer<ServerSentEvent> {
-        @Override
-        public ByteBuf transform(ServerSentEvent toTransform, ByteBufAllocator byteBufAllocator) {
-            StringBuilder eventBuilder = new StringBuilder();
-            eventBuilder.append(toTransform.getEventName());
-            eventBuilder.append(": ");
-            eventBuilder.append(toTransform.getEventData());
-            eventBuilder.append("\n\n");
-            String data = eventBuilder.toString();
-            return byteBufAllocator.buffer(data.length()).writeBytes(data.getBytes());
-        }
-    }
-
-    /*
-     * Workaround for Java 8 issue: https://github.com/Netflix/RxJava/issues/1157
-     */
-    public final static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {
-        return Observable.create(new OperatorDefer<T>(observableFactory));
-    }
 }

reactive-lab-edge/src/main/java/io/reactivex/lab/edge/JavascriptRuntime.java

@@ -7,12 +7,23 @@
 import io.reactivex.netty.pipeline.PipelineConfigurators;
 import io.reactivex.netty.protocol.http.client.HttpClient;
 import io.reactivex.netty.protocol.http.client.HttpClientBuilder;
+import io.reactivex.netty.protocol.http.server.HttpServer;
+import io.reactivex.netty.protocol.http.server.HttpServerBuilder;
+import io.reactivex.netty.protocol.http.server.RequestHandler;
 import io.reactivex.netty.protocol.text.sse.ServerSentEvent;
 
 public class RxNettySSE {
 
     private static final EventLoopGroup EVENT_LOOP = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors(), new RxNettyThreadFactory());
 
+    public static HttpServer<ByteBuf, ServerSentEvent> createHttpServer(int port,
+            RequestHandler<ByteBuf, ServerSentEvent> requestHandler) {
+        return new HttpServerBuilder<ByteBuf, ServerSentEvent>(port, requestHandler)
+                .pipelineConfigurator(PipelineConfigurators.<ByteBuf> sseServerConfigurator())
+                .eventLoop(EVENT_LOOP)
+                .build();
+    }
+
     public static HttpClient<ByteBuf, ServerSentEvent> createHttpClient(String host, int port) {
         return new HttpClientBuilder<ByteBuf, ServerSentEvent>(host, port)
                 .pipelineConfigurator(PipelineConfigurators.<ByteBuf> sseClientConfigurator())

reactive-lab-edge/src/main/java/io/reactivex/lab/edge/common/RxNettySSE.java

@@ -15,7 +15,6 @@
 import io.reactivex.lab.edge.common.SimpleJson;
 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 java.util.HashMap;
 import java.util.List;
@@ -60,7 +59,7 @@ public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerRes
 
             return Observable.merge(catalog, social);
         }).flatMap(data -> {
-            return response.writeAndFlush(new ServerSentEvent("", "data", SimpleJson.mapToJson(data)), EdgeServer.SSE_TRANSFORMER);
+            return response.writeStringAndFlush("data: " + SimpleJson.mapToJson(data));
         });
     }