Можно ли поймать исключение netty в Camel?

Question

Мне кажется, что у netty есть свои собственные обработчики исключений, и они не распространяют исключения (т. Е. IOException) обратно на верблюжий маршрут.Есть ли способ узнать, что клиент отключился?

Answer 1

Отвечая на мой собственный вопрос.Моя проблема заключалась в освобождении клиентов, которые просто ждали бы вечно, чтобы получить какой-то ответ от netty, главным образом в случае соединений, закрытых удаленными хостами во время обработки конвейера.

Что нужно сделать, это добавить пользовательский обработчик в конвейер, который должен расширять ChannelDuplexHandler и переопределять connect и писать methods или SimpleChannelInboundHandler и переопределять channelInactive.Я использовал ChannelDuplexHandler.

public class ExceptionHandler extends ChannelDuplexHandler {

private final NettyProducer producer;

@Override
public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
                    ChannelPromise promise)
        throws Exception {
    ctx.connect(remoteAddress, localAddress, promise)
            .addListener((future -> {
                if (!future.isSuccess()) {
                    // no need to do anything here, camel will manage it on its own
                }
            }));
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
    ctx.write(msg, promise).addListener(future -> {
        if (!future.isSuccess()) {
            reportStatusBackToCamel(ctx);
        }
    });
}

private void reportStatusBackToCamel(ChannelHandlerContext ctx) {
    NettyCamelState nettyCamelState = producer.getCorrelationManager().getState(ctx, ctx.channel(),
            new IOException());
    Exchange exchange = nettyCamelState.getExchange();
    AsyncCallback callback = nettyCamelState.getCallback();
    exchange.setException(new RuntimeException("Client disconnected"));
    callback.done(false);
    }
}

В случае SimpleChannelInboundHandler просто поместите обработку обмена в метод channelInactive.

В вашем ClientInitializerFactory in initChannel вы добавите этот обработчик в конвейер:

 pipeline.addLast(new ExceptionHandler(producer));

producer выдается при запуске приложения.Если вам нужны дополнительные бины с инжекцией пружины, как я сделал, у вас просто получится пара конструкторов в вашем классе фабрики, один из которых @Autowired (с вашими введенными полями) вызывает другой параметр, устанавливающий дополнительное поле производителя.