Java 类org.jboss.netty.channel.ChannelState 实例源码

private void handleServerChannel(ChannelEvent e) {
    IsdnServerChannel channel = (IsdnServerChannel) e.getChannel();
    ChannelFuture future = e.getFuture();
    if (e instanceof ChannelStateEvent) {
        ChannelStateEvent stateEvent = (ChannelStateEvent) e;
        ChannelState state = stateEvent.getState();
        Object value = stateEvent.getValue();
        switch (state) {
        case OPEN:
            if (Boolean.FALSE.equals(value)) {
                close(channel, future);
            }
            break;
        case BOUND:
            if (value != null) {
                bind(channel, future, (IsdnSocketAddress) value);
            } else {
                logger.warn("eventSunk() :: UNHANDLED (BOUND value=null) --> {}", e);
                close(channel, future);
            }
            break;
        default:
            logger.warn("eventSunk() :: UNHANDLED --> {}", e);
        }
    }
}

@Test
public void multiple_connected_same_ip() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress);

    final ChannelEvent openEvent = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, openEvent);
    subject.handleUpstream(ctx, openEvent);
    subject.handleUpstream(ctx, openEvent);

    verify(ctx, times(2)).sendUpstream(openEvent);
    verify(channel, times(1)).write(argThat(new ArgumentMatcher<Object>() {
        @Override
        public boolean matches(Object argument) {
            return QueryMessages.connectionsExceeded(MAX_CONNECTIONS_PER_IP).equals(argument);
        }
    }));
    verify(channelFuture, times(1)).addListener(ChannelFutureListener.CLOSE);
    verify(whoisLog).logQueryResult(anyString(), eq(0), eq(0), eq(QueryCompletionInfo.REJECTED), eq(0L), (InetAddress) Mockito.anyObject(), Mockito.anyInt(), eq(""));
    verify(ctx, times(2)).sendUpstream(openEvent);
}

@Test
public void multiple_connected_limit_disabled() throws Exception {
    subject.setMaxConnectionsPerIp(0);

    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress);

    final ChannelEvent openEvent = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, openEvent);
    subject.handleUpstream(ctx, openEvent);
    subject.handleUpstream(ctx, openEvent);

    final ChannelEvent closeEvent = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.FALSE);
    subject.handleUpstream(ctx, closeEvent);
    subject.handleUpstream(ctx, closeEvent);
    subject.handleUpstream(ctx, closeEvent);

    verify(ctx, times(3)).sendUpstream(openEvent);
    verify(ctx, times(3)).sendUpstream(closeEvent);
    verify(channel, never()).close();
    verify(channel, never()).write(anyObject());
    verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
}

@Test
public void multiple_connected_unlimited_allowed() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);

    when(ipResourceConfiguration.isUnlimitedConnections(any(IpInterval.class))).thenReturn(true);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress);

    final ChannelEvent event = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, event);
    subject.handleUpstream(ctx, event);
    subject.handleUpstream(ctx, event);

    verify(ctx, times(3)).sendUpstream(event);
    verify(channel, never()).close();
    verify(channel, never()).write(anyObject());
    verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
}

@Test
public void multiple_connected_proxy_allowed() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);

    when(ipResourceConfiguration.isProxy(any(IpInterval.class))).thenReturn(true);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress);

    final ChannelEvent event = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, event);
    subject.handleUpstream(ctx, event);
    subject.handleUpstream(ctx, event);

    verify(ctx, times(3)).sendUpstream(event);
    verify(channel, never()).close();
    verify(channel, never()).write(anyObject());
    verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
}

@Test
public void multiple_connected_different_ip() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);
    final InetSocketAddress remoteAddress2 = new InetSocketAddress("10.0.0.1", 43);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress).thenReturn(remoteAddress).thenReturn(remoteAddress2);

    final ChannelEvent event = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, event);
    subject.handleUpstream(ctx, event);

    final ChannelEvent event2 = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, event2);

    verify(ctx, times(2)).sendUpstream(event);
    verify(ctx, times(1)).sendUpstream(event2);
    verify(channel, never()).close();
    verify(channel, never()).write(anyObject());
    verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
}

@Test
public void multiple_connected_same_ip_and_closed() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress);

    final ChannelEvent openEvent = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, openEvent);
    subject.handleUpstream(ctx, openEvent);

    final ChannelEvent closeEvent = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.FALSE);
    subject.handleUpstream(ctx, closeEvent);
    subject.handleUpstream(ctx, closeEvent);

    subject.handleUpstream(ctx, openEvent);
    subject.handleUpstream(ctx, openEvent);

    verify(ctx, times(4)).sendUpstream(openEvent);
    verify(ctx, times(2)).sendUpstream(closeEvent);
    verify(channel, never()).close();
    verify(channel, never()).write(anyObject());
    verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
}

public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent event) throws Exception {
    if (event instanceof MessageEvent) {
        MessageEvent msgEvent = (MessageEvent) event;
        Object msg = msgEvent.getMessage();
        if (msg instanceof ChannelBuffer) {
            callDecode(ctx, (ChannelBuffer) msg, msgEvent.getRemoteAddress());
            return;
        }
    } else if (event instanceof ChannelStateEvent) {
        ChannelStateEvent stateEvent = (ChannelStateEvent) event;
        if (stateEvent.getState() == ChannelState.CONNECTED) {
            if (stateEvent.getValue() != null) {
                lengthBytesToRead = lengthFieldLength;
                lengthBuffer = getBuffer(ctx.getChannel().getConfig().getBufferFactory(),
                        lengthBytesToRead);
            }
        }
    }
    ctx.sendUpstream(event);
}

public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent event) throws Exception {
    if (event instanceof MessageEvent) {
        MessageEvent msgEvent = (MessageEvent) event;
        Object msg = msgEvent.getMessage();
        if (msg instanceof ChannelBuffer) {
            callDecode(ctx, (ChannelBuffer) msg, msgEvent.getRemoteAddress());
            return;
        }
    } else if (event instanceof ChannelStateEvent) {
        ChannelStateEvent stateEvent = (ChannelStateEvent) event;
        if (stateEvent.getState() == ChannelState.CONNECTED) {
            if (stateEvent.getValue() != null) {
                lengthBytesToRead = lengthFieldLength;
                lengthBuffer = getBuffer(ctx.getChannel().getConfig().getBufferFactory(),
                        lengthBytesToRead);
            }
        }
    }
    ctx.sendUpstream(event);
}

@Test
public void shouldThrowExceptionIfBadHandshakeIsReceived() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress(0);

    // start off by simulating a 'channelConnected' event
    // this should set the internal state properly
    handler.channelConnected(ctx, new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, remoteAddress));

    // we shouldn't forward the event on
    Mockito.verifyNoMoreInteractions(ctx);

    // now simulate an incoming message
    // the handler is expecting a handshake message
    // but we're going to feed it something else, and we expect an exception as a result
    ChannelBuffer badHandshakeBuffer = ChannelBuffers.wrappedBuffer(new byte[]{0, 1, 3, 4});
    expectedException.expect(IOException.class);
    handler.messageReceived(ctx, new UpstreamMessageEvent(channel, badHandshakeBuffer, remoteAddress));
}

@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 shouldFireExceptionIfConnectEventCannotBeForwarded() throws Exception {
    setupPipelineToRunFireEventLaterInline();

    // setup the ctx to throw an exception when we attempt to send the connected event downstream
    IllegalStateException resolveFailureCause = new IllegalStateException("downstream event failed");
    doThrow(resolveFailureCause).when(ctx).sendDownstream(any(DownstreamChannelStateEvent.class));

    // initiate the connect request
    DefaultChannelFuture originalEventFuture = new DefaultChannelFuture(channel, false);
    ChannelStateEvent event = new DownstreamChannelStateEvent(
            channel,
            originalEventFuture,
            ChannelState.CONNECTED,
            createUnresolvedRemoteAddress()
    );

    handler.connectRequested(ctx, event);

    // firing the connect downstream should cause a "fireExceptionCaughtLater" call
    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 shouldPassOnProxyConnectRequestedEvent() throws Exception {
    ChannelFuture originalConnectFuture = Channels.future(channel);
    InetSocketAddress originalValue = new InetSocketAddress(0);

    // initiate the connect request
    handler.connectRequested(ctx, new DownstreamChannelStateEvent(channel, originalConnectFuture, ChannelState.CONNECTED, originalValue));

    // verify that a proxy event is sent instead of the original event
    ArgumentCaptor<DownstreamChannelStateEvent> downstreamEvent = ArgumentCaptor.forClass(DownstreamChannelStateEvent.class);
    verify(ctx, atLeastOnce()).getChannel();
    verify(ctx).sendDownstream(downstreamEvent.capture());
    verifyNoMoreInteractions(ctx);

    DownstreamChannelStateEvent forwardedEvent = downstreamEvent.getValue();
    assertThat(forwardedEvent.getChannel(), is(channel));
    assertThat(forwardedEvent.getFuture(), notNullValue());
    assertThat(forwardedEvent.getFuture(), not(originalConnectFuture));
    assertThat(forwardedEvent.getState(), is(ChannelState.CONNECTED));
    assertThat((InetSocketAddress) forwardedEvent.getValue(), is(originalValue));
}

@Test
public void shouldFailOriginalConnectFutureWhenProxyConnectFutureIsMarkedAsFailed() throws Exception {
    ChannelFuture originalConnectFuture = Channels.future(channel);
    InetSocketAddress originalValue = new InetSocketAddress(0);

    // initiate the connect request
    handler.connectRequested(ctx, new DownstreamChannelStateEvent(channel, originalConnectFuture, ChannelState.CONNECTED, originalValue));

    // get the proxy event
    ArgumentCaptor<DownstreamChannelStateEvent> downstreamEvent = ArgumentCaptor.forClass(DownstreamChannelStateEvent.class);
    verify(ctx, atLeastOnce()).getChannel();
    verify(ctx).sendDownstream(downstreamEvent.capture());
    verifyNoMoreInteractions(ctx);

    // mark the proxy event future as having failed
    DownstreamChannelStateEvent forwardedEvent = downstreamEvent.getValue();

    Exception failureCause = new Exception();
    forwardedEvent.getFuture().setFailure(failureCause);

    // and check that we've triggered the original future as well
    assertThat(originalConnectFuture.isDone(), equalTo(true));
    assertThat(originalConnectFuture.isSuccess(), equalTo(false));
}

@Test
public void shouldNotTriggerOriginalConnectFutureIfProxyConnectFutureSucceeds() throws Exception {
    ChannelFuture originalConnectFuture = Channels.future(channel);
    InetSocketAddress originalValue = new InetSocketAddress(0);

    // initiate the connect request
    handler.connectRequested(ctx, new DownstreamChannelStateEvent(channel, originalConnectFuture, ChannelState.CONNECTED, originalValue));

    // get the proxy event
    ArgumentCaptor<DownstreamChannelStateEvent> downstreamEvent = ArgumentCaptor.forClass(DownstreamChannelStateEvent.class);
    verify(ctx, atLeastOnce()).getChannel();
    verify(ctx).sendDownstream(downstreamEvent.capture());

    // mark the proxy event future as having succeeded
    DownstreamChannelStateEvent forwardedEvent = downstreamEvent.getValue();
    forwardedEvent.getFuture().setSuccess();

    // and check that original future _is not triggered_ and no events are forwarded!
    assertThat(originalConnectFuture.isDone(), equalTo(false));
    verifyNoMoreInteractions(ctx);
}

/**
 * <p>
 * {@inheritDoc}
 * </p>
 * <p>
 * Overriden to log ChannelStateEvents if they have a state other than
 * {@link ChannelState#INTEREST_OPS}, i. e. OPEN, BOUND, CONNECTED.
 * </p>
 * 
 * @param ctx
 *            the context object for this handler
 * @param e
 *            the upstream event to process or intercept
 */
@Override
public void handleUpstream(final ChannelHandlerContext ctx, final ChannelEvent e) throws Exception {
    if (e instanceof ChannelStateEvent) {
        ChannelStateEvent stateEvent = (ChannelStateEvent) e;
        if (stateEvent.getState() != ChannelState.INTEREST_OPS) {
            log.info(e.toString());
            if (stateEvent.getState() == ChannelState.CONNECTED && stateEvent.getValue() == null) {
                // Remove channel from container when client disconnects
                channelContainer.removeChannel(e.getChannel());
            }
        }

    }
    super.handleUpstream(ctx, e);
}

@Test
public void testDisconnect() {
    emulateConnectAndOpen();

    channel.write(NettyTestUtils.createData(1234L));
    assertEquals(1, sink.events.size());
    checkIsSendDataRequestWithData(sink.events.poll(),
            NettyTestUtils.createData(1234L));

    channel.disconnect();
    Channels.fireMessageReceived(channel,
            HttpTunnelMessageUtils.createSendDataResponse());
    assertEquals(1, sink.events.size());

    HttpRequest request = NettyTestUtils.checkIsDownstreamMessageEvent(
            sink.events.poll(), HttpRequest.class);
    assertTrue(HttpTunnelMessageUtils.isCloseTunnelRequest(request, USER_AGENT));
    assertEquals("newTunnel",
            HttpTunnelMessageUtils.extractTunnelId(request));
    Channels.fireMessageReceived(channel,
            HttpTunnelMessageUtils.createTunnelCloseResponse());
    assertEquals(1, sink.events.size());
    NettyTestUtils.checkIsStateEvent(sink.events.poll(),
            ChannelState.CONNECTED, null);
}

@Test
public void testClose() {
    emulateConnectAndOpen();

    channel.close();
    assertEquals(1, sink.events.size());
    HttpRequest request = NettyTestUtils.checkIsDownstreamMessageEvent(
            sink.events.poll(), HttpRequest.class);
    assertTrue(HttpTunnelMessageUtils.isCloseTunnelRequest(request, USER_AGENT));
    assertEquals("newTunnel",
            HttpTunnelMessageUtils.extractTunnelId(request));
    Channels.fireMessageReceived(channel,
            HttpTunnelMessageUtils.createTunnelCloseResponse());
    assertEquals(1, sink.events.size());
    NettyTestUtils.checkIsStateEvent(sink.events.poll(), ChannelState.OPEN,
            false);
}

@Test
public void testBind_preResolvedAddress_ipv6() {
    ChannelFuture bindFuture = Channels.bind(channel,
            RESOLVED_LOCAL_ADDRESS_IPV6);
    assertFalse(bindFuture.isDone());

    assertEquals(1, sendSink.events.size());
    assertEquals(1, pollSink.events.size());

    ChannelEvent sendChannelEvent = sendSink.events.poll();
    NettyTestUtils.checkIsStateEvent(sendChannelEvent, ChannelState.BOUND,
            RESOLVED_LOCAL_ADDRESS_IPV6);
    ChannelEvent pollChannelEvent = pollSink.events.poll();
    NettyTestUtils.checkIsStateEvent(pollChannelEvent, ChannelState.BOUND,
            RESOLVED_LOCAL_ADDRESS_IPV6_EPHEMERAL_PORT);

    sendChannel.emulateBound(RESOLVED_LOCAL_ADDRESS_IPV6,
            sendChannelEvent.getFuture());
    assertFalse(bindFuture.isDone());
    pollChannel.emulateBound(RESOLVED_LOCAL_ADDRESS_IPV4_SELECTED_PORT,
            pollChannelEvent.getFuture());
    assertTrue(bindFuture.isDone());
    assertTrue(bindFuture.isSuccess());

    assertEquals(channel.getLocalAddress(), RESOLVED_LOCAL_ADDRESS_IPV6);
}

@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
    if(e instanceof ChannelStateEvent && ((ChannelStateEvent) e).getState() != ChannelState.INTEREST_OPS) {
        LOG.debug(e.toString());
    }
    super.handleUpstream(ctx, e);
}

@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
    if(e instanceof ChannelStateEvent && ((ChannelStateEvent) e).getState() != ChannelState.INTEREST_OPS) {
        LOG.debug(e.toString());
    }
    super.handleUpstream(ctx, e);
}

public void eventSunk(ChannelPipeline pipeline, ChannelEvent e) throws Exception {

        IsdnClientChannel channel = (IsdnClientChannel) e.getChannel();
        ChannelFuture future = e.getFuture();
        if (e instanceof ChannelStateEvent) {
            ChannelStateEvent stateEvent = (ChannelStateEvent) e;
            ChannelState state = stateEvent.getState();
            Object value = stateEvent.getValue();
            switch (state) {
            case OPEN:
                if (FALSE.equals(value)) {
                    IsdnWorker.close(channel, future);
                }
                break;
            case BOUND:
                if (value != null) {
                    bind(channel, future, (IsdnSocketAddress) value);
                } else {
                    IsdnWorker.close(channel, future);
                }
                break;
            case CONNECTED:
                if (value != null) {
                    connect(channel, future, (IsdnSocketAddress) value);
                } else {
                    IsdnWorker.close(channel, future);
                }
                break;
            case INTEREST_OPS:
                IsdnWorker.setInterestOps(channel, future, ((Integer) value).intValue());
                break;
            }
        } else if (e instanceof MessageEvent) {
            IsdnWorker.write(channel, future, ((MessageEvent) e).getMessage());
        }

    }

@Test
public void one_connected() throws Exception {
    final InetSocketAddress remoteAddress = new InetSocketAddress("10.0.0.0", 43);
    when(channel.getRemoteAddress()).thenReturn(remoteAddress);

    final ChannelEvent event = new UpstreamChannelStateEvent(channel, ChannelState.OPEN, Boolean.TRUE);
    subject.handleUpstream(ctx, event);
    subject.handleUpstream(ctx, event);

    verify(ctx, times(2)).sendUpstream(event);
    verify(channel, never()).close();
    verify(channel, never()).write(anyObject());
    verify(channelFuture, never()).addListener(ChannelFutureListener.CLOSE);
}

@Test
public void shouldThrowExceptionIfMultipleChannelConnectedEventsAreReceived() throws Exception {
    // first channel connected event
    handler.channelConnected(ctx, new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, new InetSocketAddress(0)));

    // second channel connected event
    expectedException.expect(IllegalStateException.class);
    handler.channelConnected(ctx, new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, new InetSocketAddress(0)));
}

@Test
public void shouldNotForwardChannelConnectedEventUntilHandshakeMessageReceived() throws Exception {
    // simulate a channelConnected event
    handler.channelConnected(ctx, new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, new InetSocketAddress(0)));

    // verify that no event is forwarded on (if an event is forwarded, it _must_ use the ctx object)
    Mockito.verifyNoMoreInteractions(ctx);
}

@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();

    // setup the address resolver to throw an exception
    final IllegalStateException resolveFailureCause = new IllegalStateException("simulate resolve failure");
    AddressResolverHandler.ResolvedAddressProvider failingResolver = new AddressResolverHandler.ResolvedAddressProvider() {
        @Override
        public InetSocketAddress createResolved(InetSocketAddress unresolvedAddress) throws Exception {
            throw resolveFailureCause;
        }
    };

    // fire a connect request with an unresolved address
    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);

    // this should cause a "fireExceptionCaughtLater" call
    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 shouldDropConnectedEvent() {
    UpstreamChannelStateEvent event = spy(new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, InetSocketAddress.createUnresolved("remote-host", 8888)));

    handler.channelConnected(ctx, event);

    verifyNoMoreInteractions(ctx, event);
}

@Test
public void shouldDropChannelClosedEvent() {
    UpstreamChannelStateEvent event = spy(new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, null));

    handler.channelConnected(ctx, event);

    verifyNoMoreInteractions(ctx, event);
}

@Test
public void shouldThrowExceptionIfMultipleConnectRequestedEventsAreReceived() throws Exception {
    // first connect request
    // no issues expected
    handler.connectRequested(ctx, new DownstreamChannelStateEvent(channel, Channels.future(channel), ChannelState.CONNECTED, new InetSocketAddress(0)));

    // second connect request
    // in real life this shouldn't happen
    // but, if we blindly expect this our internal state is corrupted
    expectedException.expect(IllegalStateException.class);
    handler.connectRequested(ctx, new DownstreamChannelStateEvent(channel, Channels.future(channel), ChannelState.CONNECTED, new InetSocketAddress(0)));
}

@Test
public void shouldFailOriginalConnectFutureWhenChannelIsClosed() throws Exception {
    ChannelFuture originalConnectFuture = Channels.future(channel);
    InetSocketAddress originalValue = new InetSocketAddress(0);

    // initiate the connect request
    handler.connectRequested(ctx, new DownstreamChannelStateEvent(channel, originalConnectFuture, ChannelState.CONNECTED, originalValue));

    // trigger the close future (i.e the channel closed)
    closeFuture.setSuccess();

    // and check that we've triggered the original future as well
    assertThat(originalConnectFuture.isDone(), equalTo(true));
    assertThat(originalConnectFuture.isSuccess(), equalTo(false));
}

@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
@Test
public void shouldFailOriginalConnectFutureAndCloseChannelIfHandshakeWriteFutureFails() throws Exception {
    ChannelFuture originalConnectFuture = Channels.future(channel);

    // pretend that a connectRequested event was sent over this channel
    handler.setConnectFutureForUnitTestOnly(originalConnectFuture);

    // signal that the connect succeeded
    handler.channelConnected(ctx, new UpstreamChannelStateEvent(channel, ChannelState.CONNECTED, new InetSocketAddress(0)));

    // instead of passing events upstream, the handler should
    // attempt to write an handshake out and send nothing upstream
    ArgumentCaptor<DownstreamMessageEvent> downstreamEvent = ArgumentCaptor.forClass(DownstreamMessageEvent.class);
    verify(ctx, atLeastOnce()).getChannel();
    verify(ctx).sendDownstream(downstreamEvent.capture());
    verifyNoMoreInteractions(ctx);

    // fail the write
    DownstreamMessageEvent handshakeEvent = downstreamEvent.getValue();
    IOException cause = new IOException();
    handshakeEvent.getFuture().setFailure(cause);

    // verify that the original future failed as well
    assertThat(originalConnectFuture.isDone(), equalTo(true));
    assertThat(originalConnectFuture.isSuccess(), equalTo(false));
    assertThat((IOException) originalConnectFuture.getCause(), is(cause));

    // and that the channel was closed
    verify(channel).close();
}

@Test
public void testConnect() throws InterruptedException, URISyntaxException {
    Channels.bind(channel, LOCAL_ADDRESS);
    Channels.connect(channel, REMOTE_ADDRESS);

    // this should result in BIND state events on both channels, and a
    // CONNECTED state event on the send channel, but not on the poll channel
    assertEquals(2, sendSink.events.size());
    assertEquals(1, pollSink.events.size());

    NettyTestUtils.checkIsStateEvent(sendSink.events.poll(), ChannelState.BOUND, LOCAL_ADDRESS);

    final ChannelEvent sendConnectedEvent = sendSink.events.poll();

       final Proxy proxy = ProxySelector.getDefault().select(new URI(String.format("http://%s:%d", REMOTE_ADDRESS.getHostString(), REMOTE_ADDRESS.getPort()))).get(0);
    final SocketAddress proxyAddress = proxy.address();

    NettyTestUtils.checkIsStateEvent(sendConnectedEvent, ChannelState.CONNECTED, proxyAddress == null ? REMOTE_ADDRESS : proxyAddress);

    // once the send channel indicates that it is connected, we should see
    // the tunnel open request being sent
    sendChannel.emulateConnected(LOCAL_ADDRESS, REMOTE_ADDRESS, ((ChannelStateEvent) sendConnectedEvent).getFuture());

    assertEquals(1, sendSink.events.size());
    ChannelEvent openTunnelRequest = sendSink.events.poll();
    NettyTestUtils.checkIsDownstreamMessageEvent(openTunnelRequest, ChannelBuffer.class);
}

private void checkBinding(InetSocketAddress requestedBindAddress,
        InetSocketAddress expectedPollBindRequest,
        InetSocketAddress expectedSendBindRequest,
        InetSocketAddress emulatedPollBindAddress,
        InetSocketAddress emulatedSendBindAddress) {

    ChannelFuture bindFuture = Channels.bind(channel, requestedBindAddress);
    assertFalse(bindFuture.isDone());

    assertEquals(1, sendSink.events.size());
    assertEquals(1, pollSink.events.size());

    ChannelEvent sendChannelEvent = sendSink.events.poll();
    NettyTestUtils.checkIsStateEvent(sendChannelEvent, ChannelState.BOUND,
            expectedPollBindRequest);
    ChannelEvent pollChannelEvent = pollSink.events.poll();
    NettyTestUtils.checkIsStateEvent(pollChannelEvent, ChannelState.BOUND,
            expectedSendBindRequest);

    sendChannel.emulateBound(emulatedPollBindAddress,
            sendChannelEvent.getFuture());
    assertFalse(bindFuture.isDone());
    pollChannel.emulateBound(emulatedSendBindAddress,
            pollChannelEvent.getFuture());
    assertTrue(bindFuture.isDone());
    assertTrue(bindFuture.isSuccess());

    assertEquals(channel.getLocalAddress(), emulatedPollBindAddress);
}

@Test
public void testChannelBoundEventFromReal_replicatedOnVirtual() {
    upstreamEvents.events.clear();
    InetSocketAddress boundAddr = InetSocketAddress.createUnresolved(
            "mycomputer", 12345);
    Channels.fireChannelBound(realChannelFactory.createdChannel, boundAddr);
    assertEquals(1, upstreamEvents.events.size());
    checkIsUpstreamChannelStateEvent(upstreamEvents.events.poll(),
            virtualChannel, ChannelState.BOUND, boundAddr);
}

@Test
public void testChannelUnboundEventFromReal_replicatedOnVirtual() {
    upstreamEvents.events.clear();
    Channels.fireChannelUnbound(realChannelFactory.createdChannel);
    assertEquals(1, upstreamEvents.events.size());
    checkIsUpstreamChannelStateEvent(upstreamEvents.events.poll(),
            virtualChannel, ChannelState.BOUND, null);
}

@Test
public void testChannelClosedEventFromReal_replicatedOnVirtual() {
    upstreamEvents.events.clear();
    Channels.fireChannelClosed(realChannelFactory.createdChannel);
    assertEquals(1, upstreamEvents.events.size());
    checkIsUpstreamChannelStateEvent(upstreamEvents.events.poll(),
            virtualChannel, ChannelState.OPEN, Boolean.FALSE);
}

private void checkIsUpstreamChannelStateEvent(ChannelEvent ev,
        Channel expectedChannel, ChannelState expectedState,
        Object expectedValue) {
    assertTrue(ev instanceof UpstreamChannelStateEvent);
    UpstreamChannelStateEvent checkedEv = (UpstreamChannelStateEvent) ev;
    assertSame(expectedChannel, checkedEv.getChannel());
    assertEquals(expectedState, checkedEv.getState());
    assertEquals(expectedValue, checkedEv.getValue());
}

public static ChannelStateEvent checkIsStateEvent(ChannelEvent event, ChannelState expectedState, Object expectedValue) {
    assertTrue(event instanceof ChannelStateEvent);
    ChannelStateEvent stateEvent = (ChannelStateEvent) event;
    Assert.assertEquals(expectedState, stateEvent.getState());
    Assert.assertEquals(expectedValue, stateEvent.getValue());
    return stateEvent;
}