import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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())
		);
	}
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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));
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
/**
 * 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);
	}
	
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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()));
    }
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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");
                }
            });
        }
    };
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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");
                }
            });
        }
    };
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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)));
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@Override
public void operationComplete(ChannelFuture future) throws Exception {
	if(upstreamContext != null && bgpEvent != null) {
		upstreamContext.sendUpstream(new UpstreamMessageEvent(upstreamContext.getChannel(), 
				bgpEvent, 
				upstreamContext.getChannel().getRemoteAddress()));
	}
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
@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));
}
 

import org.jboss.netty.channel.UpstreamMessageEvent; //导入依赖的package包/类
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());
}