Обработка исключений vertx RecordParser ⇐ JAVA

[Anonymous]

Я использую NetServer в Vertx, и когда подключается новый клиент, создается поток, отправляющий пакеты, полученные на этом NetClient. Чтобы понять, как правильно обрабатывать исключения, я добавил в поток несколько случайных исключений:

Код: Выделить всё

private RecordParser createParser(NetState state, FlowableEmitter emitter) {
RecordParser parser = RecordParser.newFixed(4);
parser.handler(buffer -> {
if(Math.random() > 0.8) {
logger.error("I am dying now");
throw new RuntimeException();
}

// protocol handling
}).exceptionHandler((throwable) -> {
logger.error("Got an exception: ", throwable);
emitter.onError(throwable);
});
return parser;
}

Я использую это следующим образом:

Код: Выделить всё

NetState state = new NetState();
// client -> NetClient
return Flowable.create(emitter -> client.handler(createParser(state, emitter)), BackpressureStrategy.ERROR);

Проблема в том, что мое исключение, созданное в анализаторе записей, никогда не достигает ни одного обработчика исключений. Ни тот, который принадлежит самому RecordParser, ни тот, который я прикрепил к клиенту. Все это относится к корневому контексту. Как мне отловить исключения, возникающие во время анализа записи (например, чтобы закрыть клиент, который отправил неверные данные)

Подробнее здесь: https://stackoverflow.com/questions/527 ... n-handling