@Test public void testChannelRead0() throws Exception { when(connection.getMode()).thenReturn(ConnectionMode.Speaker); ByteBuf byteBuf = MessageFactory.createOpen(Version.Version4, ConnectionMode.Speaker, sxpNode.getNodeId(), 0); decoder.channelRead(channelHandlerContext, byteBuf); verify(connection).setConnection(any(OpenMessage.class)); }
@Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf message) { final SxpConnection connection = owner.getConnection(ctx.channel().remoteAddress()); if (connection == null) { LOG.warn(getLogMessage(owner, ctx, "Channel read0")); return; } while (message.readableBytes() != 0) { try { Notification notification = connection.getContext().executeParseInput(message); connection.getContext().executeInputMessageStrategy(ctx, connection, notification); } catch (ErrorMessageException messageValidationException) { Exception carriedException = messageValidationException.getCarriedException(); if (carriedException != null) { LOG.warn(getLogMessage(owner, ctx, "", carriedException)); } sendErrorMessage(ctx, messageValidationException, connection); connection.setStateOff(ctx); break; } catch (ErrorMessageReceivedException | UpdateMessageConnectionStateException e) { LOG.warn(getLogMessage(owner, ctx, "", e)); connection.setStateOff(ctx); break; } catch (Exception e) { LOG.warn(getLogMessage(owner, ctx, "Channel read") + ": {}", MessageFactory.toString(message), e); break; } } }
MessageDecoder extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf message) { final SxpConnection connection = owner.getConnection(ctx.channel().remoteAddress()); if (connection == null) { LOG.warn(getLogMessage(owner, ctx, "Channel read0")); return; } while (message.readableBytes() != 0) { try { Notification notification = connection.getContext().executeParseInput(message); connection.getContext().executeInputMessageStrategy(ctx, connection, notification); } catch (ErrorMessageException messageValidationException) { Exception carriedException = messageValidationException.getCarriedException(); if (carriedException != null) { LOG.warn(getLogMessage(owner, ctx, "", carriedException)); } sendErrorMessage(ctx, messageValidationException, connection); connection.setStateOff(ctx); break; } catch (ErrorMessageReceivedException | UpdateMessageConnectionStateException e) { LOG.warn(getLogMessage(owner, ctx, "", e)); connection.setStateOff(ctx); break; } catch (Exception e) { LOG.warn(getLogMessage(owner, ctx, "Channel read") + ": {}", MessageFactory.toString(message), e); break; } } } }
MessageDecoder extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf message) { final SxpConnection connection = owner.getConnection(ctx.channel().remoteAddress()); if (connection == null) { LOG.warn(getLogMessage(owner, ctx, "Channel read0")); return; } while (message.readableBytes() != 0) { try { Notification notification = connection.getContext().executeParseInput(message); connection.getContext().executeInputMessageStrategy(ctx, connection, notification); } catch (ErrorMessageException messageValidationException) { Exception carriedException = messageValidationException.getCarriedException(); if (carriedException != null) { LOG.warn(getLogMessage(owner, ctx, "", carriedException)); } sendErrorMessage(ctx, messageValidationException, connection); connection.setStateOff(ctx); break; } catch (ErrorMessageReceivedException | UpdateMessageConnectionStateException e) { LOG.warn(getLogMessage(owner, ctx, "", e)); connection.setStateOff(ctx); break; } catch (Exception e) { LOG.warn(getLogMessage(owner, ctx, "Channel read") + ": {}", MessageFactory.toString(message), e); break; } } } private  MessageDecoder(SxpNode owner, Profile profile); }
MessageDecoder extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf message) { final SxpConnection connection = owner.getConnection(ctx.channel().remoteAddress()); if (connection == null) { LOG.warn(getLogMessage(owner, ctx, "Channel read0")); return; } while (message.readableBytes() != 0) { try { Notification notification = connection.getContext().executeParseInput(message); connection.getContext().executeInputMessageStrategy(ctx, connection, notification); } catch (ErrorMessageException messageValidationException) { Exception carriedException = messageValidationException.getCarriedException(); if (carriedException != null) { LOG.warn(getLogMessage(owner, ctx, "", carriedException)); } sendErrorMessage(ctx, messageValidationException, connection); connection.setStateOff(ctx); break; } catch (ErrorMessageReceivedException | UpdateMessageConnectionStateException e) { LOG.warn(getLogMessage(owner, ctx, "", e)); connection.setStateOff(ctx); break; } catch (Exception e) { LOG.warn(getLogMessage(owner, ctx, "Channel read") + ": {}", MessageFactory.toString(message), e); break; } } } private  MessageDecoder(SxpNode owner, Profile profile); static ChannelInboundHandler createClientProfile(SxpNode owner); static ChannelInboundHandler createServerProfile(SxpNode owner); static void sendErrorMessage(ChannelHandlerContext ctx, ErrorMessageException messageValidationException,
            SxpConnection connection); @Override void channelActive(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void channelRead0(ChannelHandlerContext ctx, ByteBuf message); @Override void channelReadComplete(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); }
MessageDecoder extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf message) { final SxpConnection connection = owner.getConnection(ctx.channel().remoteAddress()); if (connection == null) { LOG.warn(getLogMessage(owner, ctx, "Channel read0")); return; } while (message.readableBytes() != 0) { try { Notification notification = connection.getContext().executeParseInput(message); connection.getContext().executeInputMessageStrategy(ctx, connection, notification); } catch (ErrorMessageException messageValidationException) { Exception carriedException = messageValidationException.getCarriedException(); if (carriedException != null) { LOG.warn(getLogMessage(owner, ctx, "", carriedException)); } sendErrorMessage(ctx, messageValidationException, connection); connection.setStateOff(ctx); break; } catch (ErrorMessageReceivedException | UpdateMessageConnectionStateException e) { LOG.warn(getLogMessage(owner, ctx, "", e)); connection.setStateOff(ctx); break; } catch (Exception e) { LOG.warn(getLogMessage(owner, ctx, "Channel read") + ": {}", MessageFactory.toString(message), e); break; } } } private  MessageDecoder(SxpNode owner, Profile profile); static ChannelInboundHandler createClientProfile(SxpNode owner); static ChannelInboundHandler createServerProfile(SxpNode owner); static void sendErrorMessage(ChannelHandlerContext ctx, ErrorMessageException messageValidationException,
            SxpConnection connection); @Override void channelActive(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void channelRead0(ChannelHandlerContext ctx, ByteBuf message); @Override void channelReadComplete(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause);  }