Fix Out Of Order handling with pool acquisition

src/main/java/reactor/ipc/netty/channel/ContextHandler.java

@@ -243,8 +243,13 @@ public final ContextHandler<CHANNEL> autoCreateOperations(boolean autoCreateOper
 					}
 				}
 
-				channel.pipeline()
-				       .get(ChannelOperationsHandler.class).lastContext = this;
+				ChannelOperationsHandler h = channel.pipeline()
+				                                    .get(ChannelOperationsHandler.class);
+
+				if (h == null) {
+					return null;
+				}
+				h.lastContext = this;
 
 				channel.eventLoop().execute(op::onHandlerStart);
 			}
@@ -300,7 +305,7 @@ protected void doStarted(Channel channel) {
 	}
 
 	@Override
-	protected void initChannel(CHANNEL ch) throws Exception {
+	protected void initChannel(CHANNEL ch) {
 		accept(ch);
 	}
 

src/main/java/reactor/ipc/netty/channel/PooledClientContextHandler.java

@@ -199,9 +199,6 @@ final void connectOrAcquire(CHANNEL c) {
 		if (log.isDebugEnabled()) {
 			log.debug(format(c, "Acquired active channel"));
 		}
-		if (createOperations(c, null) == null) {
-			setFuture(pool.acquire());
-		}
 	}
 
 

src/main/java/reactor/ipc/netty/http/client/HttpClient.java

@@ -36,7 +36,6 @@
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.logging.LoggingHandler;
-
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
@@ -392,7 +391,7 @@ protected ContextHandler<SocketChannel> doHandler(BiFunction<? super NettyInboun
 				SocketAddress providedAddress,
 				ChannelPool pool,
 				Consumer<? super Channel> onSetup) {
-			return ContextHandler.<SocketChannel>newClientContext(sink,
+			ContextHandler<SocketChannel> h = ContextHandler.newClientContext(sink,
 					options,
 					loggingHandler,
 					secure,
@@ -403,7 +402,15 @@ protected ContextHandler<SocketChannel> doHandler(BiFunction<? super NettyInboun
 							onSetup.accept(ch);
 						}
 						return HttpClientOperations.bindHttp(ch, handler, c);
-					} : EMPTY).onPipeline(this);
+					} : EMPTY);
+
+			if (handler == null) {
+				h.onPipeline(ACTIVE_CONFIGURATOR.andThen(this));
+			}
+			else {
+				h.onPipeline(this);
+			}
+			return h;
 		}
 
 		@Override

src/main/java/reactor/ipc/netty/http/server/HttpPredicate.java

@@ -266,13 +266,23 @@ static final class UriPathTemplate {
 
 		private final Pattern uriPattern;
 
+		static String filterQueryParams(String uri) {
+			int hasQuery = uri.lastIndexOf("?");
+			if (hasQuery != -1) {
+				return uri.substring(0, hasQuery);
+			}
+			else {
+				return uri;
+			}
+		}
+
 		/**
 		 * Creates a new {@code UriPathTemplate} from the given {@code uriPattern}.
 		 *
 		 * @param uriPattern The pattern to be used by the template
 		 */
-		public UriPathTemplate(String uriPattern) {
-			String s = "^" + uriPattern;
+		UriPathTemplate(String uriPattern) {
+			String s = "^" + filterQueryParams(uriPattern);
 
 			Matcher m = NAME_SPLAT_PATTERN.matcher(s);
 			while (m.find()) {
@@ -347,6 +357,7 @@ final Map<String, String> match(String uri) {
 		}
 
 		private Matcher matcher(String uri) {
+			uri = filterQueryParams(uri);
 			Matcher m = matchers.get(uri);
 			if (null == m) {
 				m = uriPattern.matcher(uri);

src/main/java/reactor/ipc/netty/tcp/TcpClient.java

@@ -19,23 +19,28 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Objects;
+import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelPipeline;
 import io.netty.channel.pool.ChannelPool;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.logging.LoggingHandler;
+import io.netty.util.AttributeKey;
 import io.netty.util.NetUtil;
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
 import reactor.ipc.netty.NettyConnector;
 import reactor.ipc.netty.NettyContext;
 import reactor.ipc.netty.NettyInbound;
 import reactor.ipc.netty.NettyOutbound;
+import reactor.ipc.netty.channel.AbortedException;
 import reactor.ipc.netty.channel.ChannelOperations;
 import reactor.ipc.netty.channel.ContextHandler;
 import reactor.ipc.netty.options.ClientOptions;
@@ -194,11 +199,59 @@ protected Mono<NettyContext> newHandler(BiFunction<? super NettyInbound, ? super
 				contextHandler.setFuture(b.connect());
 			}
 			else {
-				contextHandler.setFuture(pool.acquire());
+				contextHandler.setFuture(pool.acquire()
+				                             .addListener(f -> {
+					                             if (f.isSuccess()) {
+						                             Channel c = (Channel) f.getNow();
+
+						                             ActiveChannelOperationFactory acof =
+								                             new ActiveChannelOperationFactory(contextHandler, c, sink);
+
+						                             c.attr(ACTIVE)
+						                              .get()
+						                              .subscribe(null, acof, acof);
+
+					}
+				}));
 			}
 		});
 	}
 
+	static final class ActiveChannelOperationFactory implements Runnable,
+	                                                            Consumer<Throwable> {
+
+		final ContextHandler<SocketChannel> contextHandler;
+		final Channel c;
+		final MonoSink<NettyContext> sink;
+
+		ActiveChannelOperationFactory(ContextHandler<SocketChannel> contextHandler,
+				Channel c,
+				MonoSink<NettyContext> sink) {
+			this.sink = sink;
+			this.contextHandler = contextHandler;
+			this.c = c;
+		}
+
+		@Override
+		public void accept(Throwable throwable) {
+			sink.error(throwable);
+		}
+
+		@Override
+		public void run() {
+			if (c.eventLoop()
+			     .inEventLoop()) {
+				if (contextHandler.createOperations(c, null) == null) {
+					sink.error(new AbortedException("Failed to acquire"));
+				}
+			}
+			else {
+				c.eventLoop()
+				 .execute(this);
+			}
+		}
+	}
+
 	/**
 	 * Create a {@link ContextHandler} for {@link Bootstrap#handler()}
 	 *
@@ -216,21 +269,43 @@ protected ContextHandler<SocketChannel> doHandler(BiFunction<? super NettyInboun
 			SocketAddress providedAddress,
 			ChannelPool pool,
 			Consumer<? super Channel> onSetup) {
-		return ContextHandler.newClientContext(sink,
+		ContextHandler<SocketChannel> h = ContextHandler.newClientContext(sink,
 				options,
 				loggingHandler,
 				secure,
 				providedAddress,
 				pool,
 				handler == null ? EMPTY :
 						(ch, c, msg) -> ChannelOperations.bind(ch, handler, c));
+
+		if (handler == null) {
+			h.onPipeline(ACTIVE_CONFIGURATOR);
+		}
+
+		return h;
 	}
 
-	protected static final ChannelOperations.OnNew<SocketChannel> EMPTY = (a,b,c) -> null;
 
 	static final LoggingHandler loggingHandler = new LoggingHandler(TcpClient.class);
 
 
+	static final AttributeKey<DirectProcessor<Void>> ACTIVE = AttributeKey.valueOf(
+			"$POOLED_ACTIVE_EVENT_DISPATCHER");
+
+	protected static final BiConsumer<ChannelPipeline, ContextHandler<Channel>> ACTIVE_CONFIGURATOR =	(p, h) -> {
+		p.channel()
+		 .attr(ACTIVE)
+		 .compareAndSet(null, DirectProcessor.create());
+	};
+
+	protected static final ChannelOperations.OnNew<SocketChannel> EMPTY = (a,b,c) -> {
+		a.attr(ACTIVE)
+		 .get()
+		 .onComplete();
+
+		return null;
+	};
+
 	public static final class Builder {
 		private Consumer<? super ClientOptions.Builder<?>> options;
 

src/test/java/reactor/ipc/netty/http/server/UriPathTemplateTest.java

@@ -27,9 +27,46 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.hasEntry;
+import static org.testng.AssertJUnit.assertEquals;
+import static org.testng.AssertJUnit.assertTrue;
 
 public class UriPathTemplateTest {
 
+    @Test
+    public void patternShouldMatchPathWithOnlyLetters() {
+        UriPathTemplate uriPathTemplate = new UriPathTemplate("/test/{order}");
+        // works as expected
+        assertThat(uriPathTemplate.match("/test/1").get("order"), is("1"));
+    }
+
+    @Test
+    public void patternShouldMatchPathWithDots() {
+        UriPathTemplate uriPathTemplate = new UriPathTemplate("/test/{order}");
+        // does not match, the dot in the segment parameter breaks matching
+        // expected: a map containing {"order": "2.0"}, found: empty map
+        assertThat(uriPathTemplate.match("/test/2.0").get("order"), is("2.0"));
+    }
+
+    @Test
+    public void staticPatternShouldMatchPathWithQueryParams() {
+        UriPathTemplate uriPathTemplate = new UriPathTemplate("/test/3");
+        // does not match, the query parameter breaks matching
+        // expected: true, found: false
+	    assertTrue(uriPathTemplate.matches("/test/3?q=reactor"));
+//        assertThat(uriPathTemplate.matches("/test/3?q=reactor"), is(true));
+    }
+
+    @Test
+    public void parameterizedPatternShouldMatchPathWithQueryParams() {
+        UriPathTemplate uriPathTemplate = new UriPathTemplate("/test/{order}");
+        // does not match, the query parameter breaks matching
+        // expected: a map containing {"order": "3"}, found: a map containing {"order": "3?q=reactor"}
+	    assertEquals("3",
+			    uriPathTemplate.match("/test/3?q=reactor")
+			                   .get("order"));
+//        assertThat(uriPathTemplate.match("/test/3?q=reactor").get("order"), is("3"));
+    }
+
     @Test
     public void staticPathShouldBeMatched() {
         UriPathTemplate template = new UriPathTemplate("/comments");