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

@Override
public void messageReceived(final ChannelHandlerContext ctx, final MessageEvent evt) throws Exception {

    /* Get audio channel from the enclosing RaopAudioHandler */
    Channel tempAudioChannel = null;
    synchronized(RaopAudioHandler.this) {
        tempAudioChannel = audioChannel;
    }

    if ((tempAudioChannel != null) && tempAudioChannel.isOpen() && tempAudioChannel.isReadable()) {
        tempAudioChannel.getPipeline().sendUpstream(new UpstreamMessageEvent(
            tempAudioChannel,
            evt.getMessage(),
            evt.getRemoteAddress())
        );
    }
}

@Override
public void messageReceived(final ChannelHandlerContext ctx, final MessageEvent evt) throws Exception {

    /* Get audio channel from the enclosing RaopAudioHandler */
    Channel tempAudioChannel = null;
    synchronized(RaopAudioHandler.this) {
        tempAudioChannel = audioChannel;
    }

    if ((tempAudioChannel != null) && tempAudioChannel.isOpen() && tempAudioChannel.isReadable()) {
        tempAudioChannel.getPipeline().sendUpstream(new UpstreamMessageEvent(
            tempAudioChannel,
            evt.getMessage(),
            evt.getRemoteAddress())
        );
    }
}

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

/**
 * Handles a prepared WebSocket API invocation
 * @param asMap true if the JSON reponse is a map, false if it is an array
 * @param request The prepared HTTP request so we can piggy-back on the existing RpcHandler services.
 * @param response The JSONResponse to write back to
 * @throws IOException thrown on IO errors
 */
protected void invoke(boolean asMap, HttpRequest request, JSONResponse response) throws IOException {
    try {
        JsonGenerator generator = response.writeHeader(asMap);
        InvocationChannel ichannel = new InvocationChannel();
        HttpQuery query = new HttpQuery(tsdb, request, ichannel);
        String baseRoute = query.getQueryBaseRoute();
        rpcHandler.messageReceived(null, new UpstreamMessageEvent(ichannel, request, null));
        HttpResponse resp = (HttpResponse)ichannel.getWrites().get(0);          
        ChannelBuffer content = resp.getContent();
        ChannelBufferInputStream cbis =  new ChannelBufferInputStream(content);
        ObjectReader reader = jsonMapper.reader();
        JsonNode contentNode = reader.readTree(cbis);
        cbis.close();
        if(asMap) {
            ObjectNode on = (ObjectNode)contentNode;
            Iterator<Map.Entry<String, JsonNode>> nodeIter = on.fields();
            while(nodeIter.hasNext()) {
                Map.Entry<String, JsonNode> node = nodeIter.next();
                generator.writeObjectField(node.getKey(), node.getValue());
            }
        } else {
            ArrayNode an = (ArrayNode)contentNode;
            for(int i = 0; i < an.size(); i++) {
                generator.writeObject(an.get(i));
            }           
        }
        response.closeGenerator();
    } catch (Exception ex) {
        throw new RuntimeException(ex);
    }

}

@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
    try {
        super.handleUpstream(ctx, e);
    } catch (Exception ex) {
        Channel channel = ctx.getChannel();
        if (!channel.isOpen()) {
            return;
        }
        ctx.sendUpstream(new UpstreamMessageEvent(channel, new NaviBadRequest(ex), channel.getRemoteAddress()));
    }
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {
            from("netty-http:http://localhost:{{port}}/myapp/myservice").process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    String body = exchange.getIn().getBody(String.class);

                    // for unit testing make sure we got right message
                    assertEquals("The body message is wrong", "b1=x&b2=y", body);
                    assertEquals("Get a wrong query parameter from the message header", "a", exchange.getIn().getHeader("query1"));
                    assertEquals("Get a wrong query parameter from the message header", "b", exchange.getIn().getHeader("query2"));
                    assertEquals("Get a wrong form parameter from the message header", "x", exchange.getIn().getHeader("b1"));
                    assertEquals("Get a wrong form parameter from the message header", "y", exchange.getIn().getHeader("b2"));
                    assertEquals("Get a wrong form parameter from the message header", "localhost:" + getPort(), exchange.getIn().getHeader("host"));

                    UpstreamMessageEvent event = (UpstreamMessageEvent) exchange.getIn().getHeader("CamelNettyMessageEvent");
                    DefaultHttpRequest request = (DefaultHttpRequest) event.getMessage();
                    assertNotEquals("Relative path should NOT be used in POST", "/myapp/myservice?query1=a&query2=b", request.getUri());

                    // send a response
                    exchange.getOut().getHeaders().clear();
                    exchange.getOut().setHeader(Exchange.CONTENT_TYPE, "text/plain");
                    exchange.getOut().setBody("Request message is OK");
                }
            });
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {
            from("netty-http:http://localhost:{{port}}/myapp/myservice").process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    String body = exchange.getIn().getBody(String.class);

                    // for unit testing make sure we got right message
                    assertEquals("The body message is wrong", "b1=x&b2=y", body);
                    assertEquals("Get a wrong query parameter from the message header", "a", exchange.getIn().getHeader("query1"));
                    assertEquals("Get a wrong query parameter from the message header", "b", exchange.getIn().getHeader("query2"));
                    assertEquals("Get a wrong form parameter from the message header", "x", exchange.getIn().getHeader("b1"));
                    assertEquals("Get a wrong form parameter from the message header", "y", exchange.getIn().getHeader("b2"));
                    assertEquals("Get a wrong form parameter from the message header", "localhost:" + getPort(), exchange.getIn().getHeader("host"));

                    UpstreamMessageEvent event = (UpstreamMessageEvent) exchange.getIn().getHeader("CamelNettyMessageEvent");
                    DefaultHttpRequest request = (DefaultHttpRequest) event.getMessage();
                    assertEquals("Relative path not used in POST", "/myapp/myservice?query1=a&query2=b", request.getUri());

                    // send a response
                    exchange.getOut().getHeaders().clear();
                    exchange.getOut().setHeader(Exchange.CONTENT_TYPE, "text/plain");
                    exchange.getOut().setBody("Request message is OK");
                }
            });
        }
    };
}

@Test
public void shouldThrowExceptionIfHandshakeReceivedBeforeChannelConnectedEvent() throws Exception {
    // simulate an incoming handshake message
    // since no channelConnected event was received first the handler should fail
    ChannelBuffer handshakeBuffer = Handshakers.createHandshakeMessage(S_01, mapper);
    expectedException.expect(NullPointerException.class);
    handler.messageReceived(ctx, new UpstreamMessageEvent(channel, handshakeBuffer, new InetSocketAddress(0)));
}

@Override
public void operationComplete(ChannelFuture future) throws Exception {
    if(upstreamContext != null && bgpEvent != null) {
        upstreamContext.sendUpstream(new UpstreamMessageEvent(upstreamContext.getChannel(), 
                bgpEvent, 
                upstreamContext.getChannel().getRemoteAddress()));
    }
}

@Test
public void shouldProperlyHandleIncomingHandshakeMessage() throws Exception {
    // the following actions should be performed for a incoming handshake
    // 1. set attachment to "S_01"
    // 2. remove self from pipeline
    // 3. forward channelConnected event on

    when(ctx.getChannel()).thenReturn(channel);
    when(ctx.getPipeline()).thenReturn(pipeline);

    // go through the full handshake flow:

    // address we expect in the channelConnected event
    final InetSocketAddress remoteAddress = new InetSocketAddress(0);

    // start off by simulating the original incoming '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 the incoming handshake message
    ChannelBuffer handshakeBuffer = Handshakers.createHandshakeMessage(S_01, mapper);
    handler.messageReceived(ctx, new UpstreamMessageEvent(channel, handshakeBuffer, remoteAddress));

    // captor for the event that's sent in response to this handshake
    ArgumentCaptor<ChannelEvent> upstreamEventCaptor = ArgumentCaptor.forClass(ChannelEvent.class);

    // verify the actions
    InOrder inOrder = Mockito.inOrder(channel, pipeline, ctx);
    inOrder.verify(ctx).getChannel();
    inOrder.verify(channel).setAttachment(S_01);
    inOrder.verify(ctx).getPipeline();
    inOrder.verify(pipeline).remove(handler);
    inOrder.verify(ctx).sendUpstream(upstreamEventCaptor.capture());
    inOrder.verifyNoMoreInteractions();

    ChannelEvent event = upstreamEventCaptor.getValue();
    assertThat(event, instanceOf(UpstreamChannelStateEvent.class));

    // now check that the event is actually a channelConnected event
    UpstreamChannelStateEvent channelStateEvent = (UpstreamChannelStateEvent) event;
    assertThat(channelStateEvent.getChannel(), is(channel));
    assertThat(channelStateEvent.getState(), is(ChannelState.CONNECTED));
    assertThat(channelStateEvent.getValue(), instanceOf(InetSocketAddress.class));
    assertThat((InetSocketAddress) channelStateEvent.getValue(), is(remoteAddress));
}

public static <T> T checkIsUpstreamMessageEvent(ChannelEvent event, Class<T> expectedMessageType) {
    assertTrue(event instanceof UpstreamMessageEvent);
    UpstreamMessageEvent messageEvent = (UpstreamMessageEvent) event;
    assertTrue(expectedMessageType.isInstance(messageEvent.getMessage()));
    return expectedMessageType.cast(messageEvent.getMessage());
}