ctx.getPipeline().remove(handler);
}
}
Exchange exchange = getExchange(ctx);
AsyncCallback callback = getAsyncCallback(ctx);
Object body = messageEvent.getMessage();
if (LOG.isDebugEnabled()) {
LOG.debug("Channel: {} received body: {}", new Object[]{messageEvent.getChannel(), body});
}
// if textline enabled then covert to a String which must be used for textline
if (producer.getConfiguration().isTextline()) {
try {
body = producer.getContext().getTypeConverter().mandatoryConvertTo(String.class, exchange, body);
} catch (NoTypeConversionAvailableException e) {
exchange.setException(e);
callback.done(false);
}
}
// set the result on either IN or OUT on the original exchange depending on its pattern
if (ExchangeHelper.isOutCapable(exchange)) {
NettyPayloadHelper.setOut(exchange, body);
} else {
NettyPayloadHelper.setIn(exchange, body);
}
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);
}
// 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);
}
}