@Test(timeout = 60000) public void testErrorPostInputSubscribe() throws Exception { ConnectionSubscriber subscriber = connectionHandlerRule.enableConnectionSubscriberAndAssert(true); connectionHandlerRule.activateConnectionAndAssert(subscriber); ConnectionInputSubscriber inputSubscriber = subscriber.getInputSubscriber(); assertThat("Unexpected on next notifications count pre exception.", inputSubscriber.getOnNextEvents(), hasSize(0)); assertThat("Unexpected on error notifications count pre exception.", inputSubscriber.getOnErrorEvents(), hasSize(0)); assertThat("Unexpected on completed notifications count pre exception.", inputSubscriber.getOnCompletedEvents(), hasSize(0)); final NullPointerException exception = new NullPointerException(); connectionHandlerRule.handler.exceptionCaught(connectionHandlerRule.ctx, exception); inputSubscriber.assertTerminalEvent(); assertThat("Unexpected on next notifications count post exception.", inputSubscriber.getOnNextEvents(), hasSize(0)); assertThat("Unexpected on error notifications count post exception.", inputSubscriber.getOnErrorEvents(), hasSize(1)); }
@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (!connectionEmitted && isValidToEmit(newChannelSub)) { newChannelSub.onError(cause); } else if (isValidToEmitToReadSubscriber(readProducer)) { readProducer.sendOnError(cause); } else { logger.info("Exception in the pipeline and none of the subscribers are active.", cause); } }
AbstractConnectionToChannelBridge extends BackpressureManagingHandler { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (!connectionEmitted && isValidToEmit(newChannelSub)) { newChannelSub.onError(cause); } else if (isValidToEmitToReadSubscriber(readProducer)) { readProducer.sendOnError(cause); } else { logger.info("Exception in the pipeline and none of the subscribers are active.", cause); } } }
AbstractConnectionToChannelBridge extends BackpressureManagingHandler { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (!connectionEmitted && isValidToEmit(newChannelSub)) { newChannelSub.onError(cause); } else if (isValidToEmitToReadSubscriber(readProducer)) { readProducer.sendOnError(cause); } else { logger.info("Exception in the pipeline and none of the subscribers are active.", cause); } } protected  AbstractConnectionToChannelBridge(String thisHandlerName, ConnectionEventListener eventListener,
                                                EventPublisher eventPublisher); protected  AbstractConnectionToChannelBridge(String thisHandlerName,
                                                AttributeKey<ConnectionEventListener> eventListenerAttributeKey,
                                                AttributeKey<EventPublisher> eventPublisherAttributeKey); }
AbstractConnectionToChannelBridge extends BackpressureManagingHandler { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (!connectionEmitted && isValidToEmit(newChannelSub)) { newChannelSub.onError(cause); } else if (isValidToEmitToReadSubscriber(readProducer)) { readProducer.sendOnError(cause); } else { logger.info("Exception in the pipeline and none of the subscribers are active.", cause); } } protected  AbstractConnectionToChannelBridge(String thisHandlerName, ConnectionEventListener eventListener,
                                                EventPublisher eventPublisher); protected  AbstractConnectionToChannelBridge(String thisHandlerName,
                                                AttributeKey<ConnectionEventListener> eventListenerAttributeKey,
                                                AttributeKey<EventPublisher> eventPublisherAttributeKey); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void channelUnregistered(ChannelHandlerContext ctx); @Override void userEventTriggered(ChannelHandlerContext ctx, Object evt); @SuppressWarnings("unchecked") @Override void newMessage(ChannelHandlerContext ctx, Object msg); @Override boolean shouldReadMore(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); }
AbstractConnectionToChannelBridge extends BackpressureManagingHandler { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (!connectionEmitted && isValidToEmit(newChannelSub)) { newChannelSub.onError(cause); } else if (isValidToEmitToReadSubscriber(readProducer)) { readProducer.sendOnError(cause); } else { logger.info("Exception in the pipeline and none of the subscribers are active.", cause); } } protected  AbstractConnectionToChannelBridge(String thisHandlerName, ConnectionEventListener eventListener,
                                                EventPublisher eventPublisher); protected  AbstractConnectionToChannelBridge(String thisHandlerName,
                                                AttributeKey<ConnectionEventListener> eventListenerAttributeKey,
                                                AttributeKey<EventPublisher> eventPublisherAttributeKey); @Override void handlerAdded(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void channelUnregistered(ChannelHandlerContext ctx); @Override void userEventTriggered(ChannelHandlerContext ctx, Object evt); @SuppressWarnings("unchecked") @Override void newMessage(ChannelHandlerContext ctx, Object msg); @Override boolean shouldReadMore(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause);  }