Merge pull request #227 from NiteshKant/0.x

Fixes #223 #225 #226 #214 #228 #230
ReactiveX · Sep 7, 2014 · 136af0d · 136af0d
2 parents 4744a2d + 24be8a7
commit 136af0d
Show file tree

Hide file tree

Showing 35 changed files with 563 additions and 158 deletions.
diff --git a/...texts/src/main/java/io/reactivex/netty/contexts/http/HttpContextClientChannelFactory.java b/...texts/src/main/java/io/reactivex/netty/contexts/http/HttpContextClientChannelFactory.java
@@ -100,7 +100,7 @@ public void onError(Throwable e) {
         @Override
         public void onNext(ObservableConnection<HttpClientResponse<O>, HttpClientRequest<I>> connection) {
             if (null != requestId && null != container) {
-                connection.getChannelHandlerContext().pipeline()
+                connection.getChannel().pipeline()
                           .fireUserEventTriggered(new NewContextEvent(requestId, container));
             }
             original.onNext(connection);

diff --git a/rx-netty-contexts/src/test/java/io/reactivex/netty/contexts/http/ContextPropagationTest.java b/rx-netty-contexts/src/test/java/io/reactivex/netty/contexts/http/ContextPropagationTest.java
@@ -28,9 +28,11 @@
 import io.reactivex.netty.contexts.RxContexts;
 import io.reactivex.netty.contexts.TestContext;
 import io.reactivex.netty.contexts.TestContextSerializer;
+import io.reactivex.netty.protocol.http.client.FlatResponseOperator;
 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.http.client.ResponseHolder;
 import io.reactivex.netty.protocol.http.server.HttpServer;
 import io.reactivex.netty.protocol.http.server.HttpServerBuilder;
 import io.reactivex.netty.protocol.http.server.HttpServerRequest;
@@ -41,18 +43,15 @@
 import org.junit.Before;
 import org.junit.Test;
 import rx.Observable;
-import rx.functions.Action0;
-import rx.functions.Action1;
 import rx.functions.Func1;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static io.reactivex.netty.contexts.ThreadLocalRequestCorrelator.getCurrentContextContainer;
 import static io.reactivex.netty.contexts.ThreadLocalRequestCorrelator.getCurrentRequestId;
@@ -103,7 +102,7 @@ public String getContextValue(String key) {
                     return Observable.error(e);
                 }
             }
-        }).enableWireLogging(LogLevel.DEBUG).build();
+        }).enableWireLogging(LogLevel.ERROR).build();
         mockServer.start();
     }
 
@@ -121,10 +120,10 @@ public void testEndToEnd() throws Exception {
                             public Observable<HttpClientResponse<ByteBuf>> call(HttpClient<ByteBuf, ByteBuf> client) {
                                 return client.submit(HttpClientRequest.createGet("/"));
                             }
-                        }).enableWireLogging(LogLevel.ERROR).build().start();
+                        }).enableWireLogging(LogLevel.DEBUG).build().start();
 
         HttpClient<ByteBuf, ByteBuf> testClient = RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder("localhost", server.getServerPort())
-                                                         .enableWireLogging(LogLevel.DEBUG).build();
+                                                         .enableWireLogging(LogLevel.ERROR).build();
 
         String reqId = "testE2E";
         sendTestRequest(testClient, reqId);
@@ -184,7 +183,8 @@ public void testWithPooledConnections() throws Exception {
                 RxContexts.<ByteBuf, ByteBuf>newHttpClientBuilder("localhost", mockServer.getServerPort(),
                                                                   REQUEST_ID_HEADER_NAME,
                                                                   RxContexts.DEFAULT_CORRELATOR)
-                          .withMaxConnections(1).withIdleConnectionsTimeoutMillis(100000).build();
+                          .withMaxConnections(1).enableWireLogging(LogLevel.ERROR)
+                          .withIdleConnectionsTimeoutMillis(100000).build();
         ContextsContainer container = new ContextsContainerImpl(new MapBackedKeySupplier());
         container.addContext(CTX_1_NAME, CTX_1_VAL);
         container.addContext(CTX_2_NAME, CTX_2_VAL, new TestContextSerializer());
@@ -203,7 +203,8 @@ public void testNoStateLeakOnThreadReuse() throws Exception {
                 RxContexts.<ByteBuf, ByteBuf>newHttpClientBuilder("localhost", mockServer.getServerPort(),
                                                                   REQUEST_ID_HEADER_NAME,
                                                                   RxContexts.DEFAULT_CORRELATOR)
-                          .withMaxConnections(1).withIdleConnectionsTimeoutMillis(100000).build();
+                          .withMaxConnections(1).enableWireLogging(LogLevel.ERROR)
+                          .withIdleConnectionsTimeoutMillis(100000).build();
 
         ContextsContainer container = new ContextsContainerImpl(new MapBackedKeySupplier());
         container.addContext(CTX_1_NAME, CTX_1_VAL);
@@ -259,7 +260,7 @@ public Observable<Void> call(HttpClientResponse<ByteBuf> response) {
 
     private static void invokeMockServer(HttpClient<ByteBuf, ByteBuf> testClient, final String requestId,
                                          boolean finishServerProcessing)
-            throws MockBackendRequestFailedException, InterruptedException {
+            throws MockBackendRequestFailedException, InterruptedException, TimeoutException, ExecutionException {
         try {
             sendTestRequest(testClient, requestId);
         } finally {
@@ -277,40 +278,24 @@ private static void invokeMockServer(HttpClient<ByteBuf, ByteBuf> testClient, fi
     }
 
     private static void sendTestRequest(HttpClient<ByteBuf, ByteBuf> testClient, final String requestId)
-            throws MockBackendRequestFailedException, InterruptedException {
+            throws MockBackendRequestFailedException, InterruptedException, TimeoutException, ExecutionException {
         System.err.println("Sending test request to mock server, with request id: " + requestId);
         RxContexts.DEFAULT_CORRELATOR.dumpThreadState(System.err);
-        final CountDownLatch finishLatch = new CountDownLatch(1);
-        final List<HttpClientResponse<ByteBuf>> responseHolder = new ArrayList<HttpClientResponse<ByteBuf>>();
-        testClient.submit(HttpClientRequest.createGet("").withHeader(REQUEST_ID_HEADER_NAME, requestId))
-                  .finallyDo(new Action0() {
-                      @Override
-                      public void call() {
-                          finishLatch.countDown();
-                      }
-                  })
-                  .subscribe(new Action1<HttpClientResponse<ByteBuf>>() {
-                      @Override
-                      public void call(HttpClientResponse<ByteBuf> response) {
-                          responseHolder.add(response);
-                      }
-                  });
-
-        finishLatch.await(1, TimeUnit.MINUTES);
-        if (responseHolder.isEmpty()) {
-            throw new AssertionError("Response not received.");
-        }
 
-        System.err.println("Received response from mock server, with request id: " + requestId
-                           + ", status: " + responseHolder.get(0).getStatus());
+        ResponseHolder<ByteBuf> responseHolder =
+                testClient.submit(HttpClientRequest.createGet("").withHeader(REQUEST_ID_HEADER_NAME, requestId))
+                          .lift(FlatResponseOperator.<ByteBuf>flatResponse())
+                          .toBlocking().toFuture().get(1, TimeUnit.MINUTES);
 
-        HttpClientResponse<ByteBuf> response = responseHolder.get(0);
+        System.err.println("Received response from mock server, with request id: " + requestId
+                           + ", status: " + responseHolder.getResponse().getStatus());
 
-        if (response.getStatus().code() != HttpResponseStatus.OK.code()) {
-            throw new MockBackendRequestFailedException("Test request failed. Status: " + response.getStatus().code());
+        if (responseHolder.getResponse().getStatus().code() != HttpResponseStatus.OK.code()) {
+            throw new MockBackendRequestFailedException("Test request failed. Status: "
+                                                        + responseHolder.getResponse().getStatus().code());
         }
 
-        String requestIdGot = response.getHeaders().get(REQUEST_ID_HEADER_NAME);
+        String requestIdGot = responseHolder.getResponse().getHeaders().get(REQUEST_ID_HEADER_NAME);
 
         if (!requestId.equals(requestId)) {
             throw new MockBackendRequestFailedException("Request Id not sent from mock server. Expected: "

diff --git a/...mples/src/main/java/io/reactivex/netty/examples/http/cpuintensive/CPUIntensiveServer.java b/...mples/src/main/java/io/reactivex/netty/examples/http/cpuintensive/CPUIntensiveServer.java
@@ -49,7 +49,7 @@ public Observable<Void> handle(HttpServerRequest<ByteBuf> request,
                                                    final HttpServerResponse<ByteBuf> response) {
                         printRequestHeader(request);
                         response.getHeaders().set(IN_EVENT_LOOP_HEADER_NAME,
-                                                  response.getChannelHandlerContext().channel().eventLoop()
+                                                  response.getChannel().eventLoop()
                                                           .inEventLoop());
                         response.writeString("Welcome!!");
                         return response.close(false);

diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/udp/HelloUdpClient.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/udp/HelloUdpClient.java
@@ -17,6 +17,7 @@
 package io.reactivex.netty.examples.udp;
 
 import io.netty.channel.socket.DatagramPacket;
+import io.netty.handler.logging.LogLevel;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.channel.ObservableConnection;
 import rx.Observable;
@@ -38,15 +39,17 @@ public HelloUdpClient(int port) {
     }
 
     public String sendHello() {
-        String content = RxNetty.createUdpClient("localhost", port).connect()
+        String content = RxNetty.<DatagramPacket, DatagramPacket>newUdpClientBuilder("localhost", port)
+                                .enableWireLogging(LogLevel.ERROR).build().connect()
                 .flatMap(new Func1<ObservableConnection<DatagramPacket, DatagramPacket>,
                         Observable<DatagramPacket>>() {
                     @Override
                     public Observable<DatagramPacket> call(ObservableConnection<DatagramPacket, DatagramPacket> connection) {
                         connection.writeStringAndFlush("Is there anybody out there?");
                         return connection.getInput();
                     }
-                }).take(1)
+                })
+                .take(1)
                 .map(new Func1<DatagramPacket, String>() {
                     @Override
                     public String call(DatagramPacket datagramPacket) {

diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/udp/HelloUdpServer.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/udp/HelloUdpServer.java
@@ -52,7 +52,7 @@ public Observable<Void> call(DatagramPacket received) {
                         InetSocketAddress sender = received.sender();
                         System.out.println("Received datagram. Sender: " + sender + ", data: "
                                 + received.content().toString(Charset.defaultCharset()));
-                        ByteBuf data = newConnection.getChannelHandlerContext().alloc().buffer(WELCOME_MSG_BYTES.length);
+                        ByteBuf data = newConnection.getChannel().alloc().buffer(WELCOME_MSG_BYTES.length);
                         data.writeBytes(WELCOME_MSG_BYTES);
                         return newConnection.writeAndFlush(new DatagramPacket(data, sender));
                     }

diff --git a/...bservable/src/main/java/io/reactivex/netty/ingress/InetAddressWhiteListIngressPolicy.java b/...bservable/src/main/java/io/reactivex/netty/ingress/InetAddressWhiteListIngressPolicy.java
@@ -17,14 +17,13 @@
 
 import io.reactivex.netty.RemoteRxEvent;
 import io.reactivex.netty.channel.ObservableConnection;
+import rx.Observable;
+import rx.functions.Action1;
 
 import java.net.InetSocketAddress;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
-import rx.Observable;
-import rx.functions.Action1;
-
 public class InetAddressWhiteListIngressPolicy implements IngressPolicy{
 
 	private AtomicReference<Set<String>> whiteList = new AtomicReference<Set<String>>();
@@ -42,7 +41,7 @@ public void call(Set<String> newList) {
 	public boolean allowed(
 			ObservableConnection<RemoteRxEvent, RemoteRxEvent> connection) {
 		InetSocketAddress inetSocketAddress 
-			= (InetSocketAddress) connection.getChannelHandlerContext().channel().remoteAddress();
+			= (InetSocketAddress) connection.getChannel().remoteAddress();
 		return whiteList.get().contains(inetSocketAddress.getAddress().getHostAddress());
 	}
 

diff --git a/rx-netty/src/main/java/io/reactivex/netty/channel/AbstractConnectionEvent.java b/rx-netty/src/main/java/io/reactivex/netty/channel/AbstractConnectionEvent.java
@@ -0,0 +1,43 @@
+/*
+ * 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 rx.Observer;
+
+/**
+ * @author Nitesh Kant
+ */
+@SuppressWarnings("rawtypes")
+public abstract class AbstractConnectionEvent<T extends ObservableConnection> {
+
+    @SuppressWarnings("rawtypes") protected final T observableConnection;
+    @SuppressWarnings("rawtypes") protected final Observer connectedObserver;
+
+    protected AbstractConnectionEvent(@SuppressWarnings("rawtypes") Observer connectedObserver,
+                                      final T observableConnection) {
+        this.connectedObserver = connectedObserver;
+        this.observableConnection = observableConnection;
+    }
+
+    public @SuppressWarnings("rawtypes")
+    Observer getConnectedObserver() {
+        return connectedObserver;
+    }
+
+    public @SuppressWarnings("rawtypes") T getConnection() {
+        return observableConnection;
+    }
+}
diff --git a/rx-netty/src/main/java/io/reactivex/netty/channel/DefaultChannelWriter.java b/rx-netty/src/main/java/io/reactivex/netty/channel/DefaultChannelWriter.java
@@ -39,7 +39,10 @@ public class DefaultChannelWriter<O> implements ChannelWriter<O> {
     protected static final Observable<Void> CONNECTION_ALREADY_CLOSED =
             Observable.error(new IllegalStateException("Connection is already closed."));
     protected final AtomicBoolean closeIssued = new AtomicBoolean();
+
     private final ChannelHandlerContext ctx;
+    private final Channel nettyChannel;
+
     /**
      * A listener for all pending writes before a flush.
      */
@@ -55,6 +58,7 @@ protected DefaultChannelWriter(ChannelHandlerContext ctx, MetricEventsSubject<?>
             throw new NullPointerException("Channel context can not be null.");
         }
         this.ctx = ctx;
+        nettyChannel = ctx.channel();
         unflushedWritesListener = new AtomicReference<MultipleFutureListener>(new MultipleFutureListener(ctx.newPromise()));
     }
 
@@ -108,14 +112,14 @@ public Observable<Void> writeStringAndFlush(String msg) {
     public Observable<Void> flush() {
         final long startTimeMillis = Clock.newStartTimeMillis();
         eventsSubject.onEvent(metricEventProvider.getFlushStartEvent());
-        MultipleFutureListener existingListener = unflushedWritesListener.getAndSet(new MultipleFutureListener(
-                ctx.newPromise()));
+        MultipleFutureListener existingListener =
+                unflushedWritesListener.getAndSet(new MultipleFutureListener(nettyChannel.newPromise()));
         /**
          * Do flush() after getting the last listener so that we do not wait for a write which is not flushed.
          * If we do it before getting the existingListener then the write that happens after the flush() from the user
          * will be contained in the retrieved listener and hence we will wait till the next flush() finish.
          */
-        ctx.flush();
+        nettyChannel.flush();
         return existingListener.asObservable()
                                .doOnCompleted(new Action0() {
                                    @Override
@@ -140,9 +144,15 @@ public void cancelPendingWrites(boolean mayInterruptIfRunning) {
 
     @Override
     public ByteBufAllocator getAllocator() {
-        return ctx.alloc();
+        return nettyChannel.alloc();
     }
 
+    /**
+     * @deprecated It is misleading to provide {@link ChannelHandlerContext} instance as it is unclear which handler
+     * this context belongs to. So, instead one should use {@link #getChannel()} and all actions possible from the
+     * {@link ChannelHandlerContext} are possible via the {@link Channel}.
+     */
+    @Deprecated
     public ChannelHandlerContext getChannelHandlerContext() {
         return ctx;
     }
@@ -153,8 +163,8 @@ protected ChannelFuture writeOnChannel(Object msg) {
         return writeFuture;
     }
 
-    protected Channel getChannel() {
-        return ctx.channel();
+    public Channel getChannel() {
+        return nettyChannel;
     }
 
     public boolean isCloseIssued() {

diff --git a/rx-netty/src/main/java/io/reactivex/netty/channel/NewRxConnectionEvent.java b/rx-netty/src/main/java/io/reactivex/netty/channel/NewRxConnectionEvent.java
@@ -22,15 +22,12 @@
  *
  * @author Nitesh Kant
  */
-public class NewRxConnectionEvent {
+@SuppressWarnings("rawtypes")
+public class NewRxConnectionEvent extends AbstractConnectionEvent<ObservableConnection> {
 
-    @SuppressWarnings("rawtypes") private final Observer connectedObserver;
-
-    public NewRxConnectionEvent(@SuppressWarnings("rawtypes") Observer connectedObserver) {
-        this.connectedObserver = connectedObserver;
+    public NewRxConnectionEvent(final ObservableConnection<?, ?> observableConnection,
+                                final Observer connectedObserver) {
+        super(connectedObserver, observableConnection);
     }
 
-    public @SuppressWarnings("rawtypes") Observer getConnectedObserver() {
-        return connectedObserver;
-    }
 }