Java 类org.apache.camel.component.netty.NettyConstants 实例源码

@Override
protected Object getRequestBody(Exchange exchange) throws Exception {
    // creating the url to use takes 2-steps
    String uri = NettyHttpHelper.createURL(exchange, getEndpoint());
    URI u = NettyHttpHelper.createURI(exchange, uri, getEndpoint());

    HttpRequest request = getEndpoint().getNettyHttpBinding().toNettyRequest(exchange.getIn(), u.toString(), getConfiguration());
    String actualUri = request.getUri();
    exchange.getIn().setHeader(Exchange.HTTP_URL, actualUri);
    // Need to check if we need to close the connection or not
    if (!HttpHeaders.isKeepAlive(request)) {
        // just want to make sure we close the channel if the keepAlive is not true
        exchange.setProperty(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
    }
    if (getConfiguration().isBridgeEndpoint()) {
        // Need to remove the Host key as it should be not used when bridging/proxying
        exchange.getIn().removeHeader("host");
    }

    return request;
}

@Test
public void testSSLInOutWithNettyConsumer() throws Exception {
    // ibm jdks dont have sun security algorithms
    if (isJavaVendor("ibm")) {
        return;
    }

    getMockEndpoint("mock:input").expectedBodiesReceived("Hello World");

    context.addRoutes(new RouteBuilder() {
        public void configure() {
            from("netty-http:https://localhost:{{port}}?ssl=true&passphrase=changeit&keyStoreResource=jsse/localhost.ks&trustStoreResource=jsse/localhost.ks")
                    .to("mock:input")
                    .process(new Processor() {
                        public void process(Exchange exchange) throws Exception {
                            SSLSession session = exchange.getIn().getHeader(NettyConstants.NETTY_SSL_SESSION, SSLSession.class);
                            if (session != null) {
                                exchange.getOut().setBody("Bye World");
                            } else {
                                exchange.getOut().setBody("Cannot start conversion without SSLSession");
                            }
                        }
                    });
        }
    });
    context.start();

    String out = template.requestBody("https://localhost:{{port}}", "Hello World", String.class);
    assertEquals("Bye World", out);

    assertMockEndpointsSatisfied();
}

@Override
public void operationComplete(ChannelFuture future) throws Exception {
    // if it was not a success then thrown an exception
    if (!future.isSuccess()) {
        Exception e = new CamelExchangeException("Cannot write response to " + remoteAddress, exchange, future.getCause());
        consumer.getExceptionHandler().handleException(e);
    }

    // should channel be closed after complete?
    Boolean close;
    if (exchange.hasOut()) {
        close = exchange.getOut().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
    } else {
        close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
    }

    // check the setting on the exchange property
    if (close == null) {
        close = exchange.getProperty(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
    }

    // should we disconnect, the header can override the configuration
    boolean disconnect = consumer.getConfiguration().isDisconnect();
    if (close != null) {
        disconnect = close;
    }
    if (disconnect) {
        if (LOG.isTraceEnabled()) {
            LOG.trace("Closing channel when complete at address: {}", remoteAddress);
        }
        NettyHelper.close(future.getChannel());
    }
}

@Override
public Set<Credential> extractCredentials() {
    HashSet<Credential> credentials = new HashSet<Credential>();
    ChannelHandlerContext handlerContext = getMessage().getHeader(NettyConstants.NETTY_CHANNEL_HANDLER_CONTEXT, ChannelHandlerContext.class);
    if (handlerContext != null) {
        SslHandler sslHandler = handlerContext.getPipeline().get(SslHandler.class);
        if (sslHandler != null) {
            credentials.addAll(new SSLSessionCredentialExtractor().extract(sslHandler.getEngine().getSession()));
        }
    }
    return credentials;
}

public void checkAdd(String id, Exchange exchange){

    if(exchange == null || id == null)
        return;

    ChannelHandlerContext channelHandlerContext = (ChannelHandlerContext) exchange.getIn().getHeader(NettyConstants.NETTY_CHANNEL_HANDLER_CONTEXT);
    Channel ioSession = channelHandlerContext.getChannel();

    if(ioSession == null){
        return;
    }

    Channel session = sessionMap.get(id);
    if(session == null){
        sessionMap.put(id, ioSession);
        idleMap.put(id, Boolean.TRUE);
        lastActTimestamp.put(id, new Date().getTime());
        seqMap.put(id, 0);
    }else{
        if(!session.getId().equals(ioSession.getId())){
            session.close();
            sessionMap.put(id, ioSession);
            idleMap.put(id, Boolean.TRUE);
            lastActTimestamp.put(id, new Date().getTime());
            seqMap.put(id, 0);
        }
    }

    concentratorManager.onLine(id);

}

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent messageEvent) throws Exception {
    messageReceived = true;

    if (LOG.isTraceEnabled()) {
        LOG.trace("Message received: {}", messageEvent);
    }

    ChannelHandler handler = ctx.getPipeline().get("timeout");
    if (handler != null) {
        LOG.trace("Removing timeout channel as we received message");
        ctx.getPipeline().remove(handler);
    }

    Exchange exchange = getExchange(ctx);
    if (exchange == null) {
        // we just ignore the received message as the channel is closed
        return;
    }     

    AsyncCallback callback = getAsyncCallback(ctx);

    Message message;
    try {
        message = getResponseMessage(exchange, messageEvent);
    } catch (Exception e) {
        exchange.setException(e);
        callback.done(false);
        return;
    }

    // set the result on either IN or OUT on the original exchange depending on its pattern
    if (ExchangeHelper.isOutCapable(exchange)) {
        exchange.setOut(message);
    } else {
        exchange.setIn(message);
    }

    try {
        // should channel be closed after complete?
        Boolean close;
        if (ExchangeHelper.isOutCapable(exchange)) {
            close = exchange.getOut().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
        } else {
            close = exchange.getIn().getHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
        }

        // check the setting on the exchange property
        if (close == null) {
            close = exchange.getProperty(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, Boolean.class);
        }

        // should we disconnect, the header can override the configuration
        boolean disconnect = producer.getConfiguration().isDisconnect();
        if (close != null) {
            disconnect = close;
        }
        if (disconnect) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Closing channel when complete at address: {}", producer.getConfiguration().getAddress());
            }
            NettyHelper.close(ctx.getChannel());
        }
    } finally {
        // signal callback
        callback.done(false);
    }
}

public void testResp(Exchange exchange){
    //System.out.print(exchange.getIn().getBody());

    //producerTemplate.sendBody("jms:queue:test:2", "小葱:" + count++);

    String r = (String) exchange.getIn().getBody();
    logger.debug("接收:" + r);



    //IoSession ioSession = (IoSession) exchange.getIn().getHeader(Mina2Constants.MINA_IOSESSION);

    ChannelHandlerContext channelHandlerContext = (ChannelHandlerContext) exchange.getIn().getHeader(NettyConstants.NETTY_CHANNEL_HANDLER_CONTEXT);
    channelHandlerContext.getChannel().write("you send : " + r);

    System.out.println(channelHandlerContext.getChannel().getId());

    /*if(!r.equals("bye")){
        exchange.getOut().setBody("you send : " + r);
    }else{
        exchange.getOut().setHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
        exchange.getOut().setBody("bye!!!");
    }*/

}