AddressResolverHandler extends SimpleChannelDownstreamHandler { @Override public void connectRequested(final ChannelHandlerContext ctx, final ChannelStateEvent e) throws Exception { if (!(e.getValue() instanceof InetSocketAddress)) { super.connectRequested(ctx, e); return; } final InetSocketAddress originalAddress = (InetSocketAddress) e.getValue(); if (!originalAddress.isUnresolved()) { super.connectRequested(ctx, e); return; } ListenableFuture<InetSocketAddress> resolvedFuture = resolveService.submit(new Callable<InetSocketAddress>() { @Override public InetSocketAddress call() throws Exception { return resolvedAddressProvider.createResolved(originalAddress); } }); Futures.addCallback(resolvedFuture, new FutureCallback<InetSocketAddress>() { @Override public void onSuccess(InetSocketAddress resolvedAddress) { try { DownstreamChannelStateEvent forwardedEvent = new DownstreamChannelStateEvent(e.getChannel(), e.getFuture(), e.getState(), resolvedAddress); AddressResolverHandler.super.connectRequested(ctx, forwardedEvent); } catch (Exception cause) { failConnect(cause); } } @Override public void onFailure(Throwable cause) { failConnect(cause); } private void failConnect(Throwable cause) { LOGGER.warn("fail connect to unresolved address:{}", originalAddress, cause); Channels.fireExceptionCaughtLater(ctx, cause); } }); } AddressResolverHandler(ListeningExecutorService resolveService);   AddressResolverHandler(ListeningExecutorService resolveService, ResolvedAddressProvider resolvedAddressProvider); @Override void connectRequested(final ChannelHandlerContext ctx, final ChannelStateEvent e);  }
@Test public void shouldResolveAddressAndForwardEventWhenConnectIsCalledWithAnUnresolvedInetSocketAddress() throws Exception { DefaultChannelFuture originalEventFuture = new DefaultChannelFuture(channel, false); ChannelStateEvent event = new DownstreamChannelStateEvent( channel, originalEventFuture, ChannelState.CONNECTED, createUnresolvedRemoteAddress() ); handler.connectRequested(ctx, event); ArgumentCaptor<DownstreamChannelStateEvent> eventCaptor = ArgumentCaptor.forClass(DownstreamChannelStateEvent.class); verify(ctx).sendDownstream(eventCaptor.capture()); DownstreamChannelStateEvent forwardedEvent = eventCaptor.getValue(); assertThat(forwardedEvent.getChannel(), equalTo(channel)); assertThat(forwardedEvent.getState(), equalTo(ChannelState.CONNECTED)); assertThat(forwardedEvent.getFuture(), Matchers.<ChannelFuture>equalTo(originalEventFuture)); InetSocketAddress forwardedAddress = (InetSocketAddress) forwardedEvent.getValue(); assertThat(forwardedAddress.isUnresolved(), equalTo(false)); }
@Test public void shouldSimplyForwardEventWhenConnectIsCalledWithAResolvedInetSocketAddress() throws Exception { ChannelStateEvent event = new DownstreamChannelStateEvent( channel, new DefaultChannelFuture(channel, false), ChannelState.CONNECTED, new InetSocketAddress("localhost", 9999) ); handler.connectRequested(ctx, event); verify(ctx).sendDownstream(refEq(event)); }
@Test public void shouldSimplyForwardEventWhenConnectIsCalledWithNonInetSocketAddress() throws Exception { ChannelStateEvent event = new DownstreamChannelStateEvent( channel, new DefaultChannelFuture(channel, false), ChannelState.CONNECTED, new LocalAddress(LocalAddress.EPHEMERAL) ); handler.connectRequested(ctx, event); verify(ctx).sendDownstream(refEq(event)); }
@Test public void shouldFireExceptionIfAddressResolutionTaskFails() throws Exception { setupPipelineToRunFireEventLaterInline(); final IllegalStateException resolveFailureCause = new IllegalStateException("simulate resolve failure"); AddressResolverHandler.ResolvedAddressProvider failingResolver = new AddressResolverHandler.ResolvedAddressProvider() { @Override public InetSocketAddress createResolved(InetSocketAddress unresolvedAddress) throws Exception { throw resolveFailureCause; } }; AddressResolverHandler failingHandler = new AddressResolverHandler(MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor()), failingResolver); DefaultChannelFuture originalEventFuture = new DefaultChannelFuture(channel, false); ChannelStateEvent event = new DownstreamChannelStateEvent( channel, originalEventFuture, ChannelState.CONNECTED, createUnresolvedRemoteAddress() ); failingHandler.connectRequested(ctx, event); ArgumentCaptor<DefaultExceptionEvent> exceptionEventCaptor = ArgumentCaptor.forClass(DefaultExceptionEvent.class); verify(ctx).sendUpstream(exceptionEventCaptor.capture()); DefaultExceptionEvent exceptionEvent = exceptionEventCaptor.getValue(); assertThat(exceptionEvent.getCause(), Matchers.<Throwable>is(resolveFailureCause)); }
@Test public void shouldFireExceptionIfConnectEventCannotBeForwarded() throws Exception { setupPipelineToRunFireEventLaterInline(); IllegalStateException resolveFailureCause = new IllegalStateException("downstream event failed"); doThrow(resolveFailureCause).when(ctx).sendDownstream(any(DownstreamChannelStateEvent.class)); DefaultChannelFuture originalEventFuture = new DefaultChannelFuture(channel, false); ChannelStateEvent event = new DownstreamChannelStateEvent( channel, originalEventFuture, ChannelState.CONNECTED, createUnresolvedRemoteAddress() ); handler.connectRequested(ctx, event); ArgumentCaptor<DefaultExceptionEvent> exceptionEventCaptor = ArgumentCaptor.forClass(DefaultExceptionEvent.class); verify(ctx).sendUpstream(exceptionEventCaptor.capture()); DefaultExceptionEvent exceptionEvent = exceptionEventCaptor.getValue(); assertThat(exceptionEvent.getCause(), Matchers.<Throwable>is(resolveFailureCause)); }