@Test public void testConnectionInProcessWrites() throws Exception { ChannelPromise connectPromise = channel.newPromise(); when(connectionManager.connectionState()).thenReturn(ClientConnectionState.NOT_CONNECTED); when(connectionManager.connect()).thenReturn(connectPromise); Request request1 = requestFactory("req1"); Request request2 = requestFactory("req2"); Request request3 = requestFactory("req3"); Optional<ChannelFuture> result1Future = subject.write(request1); when(connectionManager.connectionState()).thenReturn(ClientConnectionState.CONNECTING); Optional<ChannelFuture> result2Future = subject.write(request2); Optional<ChannelFuture> result3Future = subject.write(request3); Request proxiedRequest1BeforeConnection = channel.readOutbound(); assertNull(proxiedRequest1BeforeConnection); connectPromise.setSuccess(); Request proxiedRequest1AfterConnectionComplete = channel.readOutbound(); Request proxiedRequest2AfterConnectionComplete = channel.readOutbound(); Request proxiedRequest3AfterConnectionComplete = channel.readOutbound(); assertEquals(proxiedRequest1AfterConnectionComplete.path(), request1.path()); assertEquals(proxiedRequest2AfterConnectionComplete.path(), request2.path()); assertEquals(proxiedRequest3AfterConnectionComplete.path(), request3.path()); assertTrue(result1Future.get().isDone() && result1Future.get().isSuccess()); assertTrue(result2Future.get().isDone() && result2Future.get().isSuccess()); assertTrue(result3Future.get().isDone() && result3Future.get().isSuccess()); }
public Optional<ChannelFuture> write(Request request) { ChannelPromise promise; if (manager.connectionState() == ClientConnectionState.NOT_CONNECTED) { log.debug( "== No channel exists, lets connect on client: " + this + " with request: " + request); ChannelFuture connectFuture = manager.connect(); promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); connectFuture.addListener(this::executeBufferedRequests); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTING) { promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTED) { log.debug("== already connected, just writing req: " + request + " on client: " + this); return Optional.of(this.rawWrite(request)); } else { log.error("Connect failed on client: " + this); return Optional.empty(); } }
Client { public Optional<ChannelFuture> write(Request request) { ChannelPromise promise; if (manager.connectionState() == ClientConnectionState.NOT_CONNECTED) { log.debug( "== No channel exists, lets connect on client: " + this + " with request: " + request); ChannelFuture connectFuture = manager.connect(); promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); connectFuture.addListener(this::executeBufferedRequests); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTING) { promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTED) { log.debug("== already connected, just writing req: " + request + " on client: " + this); return Optional.of(this.rawWrite(request)); } else { log.error("Connect failed on client: " + this); return Optional.empty(); } } }
Client { public Optional<ChannelFuture> write(Request request) { ChannelPromise promise; if (manager.connectionState() == ClientConnectionState.NOT_CONNECTED) { log.debug( "== No channel exists, lets connect on client: " + this + " with request: " + request); ChannelFuture connectFuture = manager.connect(); promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); connectFuture.addListener(this::executeBufferedRequests); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTING) { promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTED) { log.debug("== already connected, just writing req: " + request + " on client: " + this); return Optional.of(this.rawWrite(request)); } else { log.error("Connect failed on client: " + this); return Optional.empty(); } } Client(ClientState state, ClientConnectionManager manager); }
Client { public Optional<ChannelFuture> write(Request request) { ChannelPromise promise; if (manager.connectionState() == ClientConnectionState.NOT_CONNECTED) { log.debug( "== No channel exists, lets connect on client: " + this + " with request: " + request); ChannelFuture connectFuture = manager.connect(); promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); connectFuture.addListener(this::executeBufferedRequests); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTING) { promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTED) { log.debug("== already connected, just writing req: " + request + " on client: " + this); return Optional.of(this.rawWrite(request)); } else { log.error("Connect failed on client: " + this); return Optional.empty(); } } Client(ClientState state, ClientConnectionManager manager); InetSocketAddress remoteAddress(); Optional<ChannelFuture> write(Request request); void prepareForReuse(Supplier<ChannelHandler> handlerSupplier); void recycle(); boolean isReusable(); }
Client { public Optional<ChannelFuture> write(Request request) { ChannelPromise promise; if (manager.connectionState() == ClientConnectionState.NOT_CONNECTED) { log.debug( "== No channel exists, lets connect on client: " + this + " with request: " + request); ChannelFuture connectFuture = manager.connect(); promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); connectFuture.addListener(this::executeBufferedRequests); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTING) { promise = manager.currentChannel().newPromise(); log.debug("== Adding req: " + request + " to queue on client: " + this); this.requestQueue.add(new Client.ClientPayload(request, promise)); return Optional.of(promise); } else if (manager.connectionState() == ClientConnectionState.CONNECTED) { log.debug("== already connected, just writing req: " + request + " on client: " + this); return Optional.of(this.rawWrite(request)); } else { log.error("Connect failed on client: " + this); return Optional.empty(); } } Client(ClientState state, ClientConnectionManager manager); InetSocketAddress remoteAddress(); Optional<ChannelFuture> write(Request request); void prepareForReuse(Supplier<ChannelHandler> handlerSupplier); void recycle(); boolean isReusable();  }