@Test public void testOnInputMessageError() throws Exception { ErrorMessage message = mock(ErrorMessage.class); when(message.getInformation()).thenReturn(""); when(message.getType()).thenReturn(MessageType.Error); when(message.getPayload()).thenReturn(new byte[] {}); exception.expect(ErrorMessageReceivedException.class); sxpLegacy.onInputMessage(channelHandlerContext, connection, message); }
@Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageException, UpdateMessageConnectionStateException, ErrorMessageReceivedException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessageLegacy) { OpenMessageLegacy legacyOpenMsg = (OpenMessageLegacy) message; if (legacyOpenMsg.getType().equals(MessageType.Open)) { if (connection.isStateDeleteHoldDown()) { connection.setReconciliationTimer(); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); ByteBuf response = LegacyMessageFactory.createOpenResp(connection.getVersion(), connection.getMode()); LOG.info("{} Sent RESP {}", connection, MessageFactory.toString(response)); ctx.writeAndFlush(response); return; } } else if (legacyOpenMsg.getType().equals(MessageType.OpenResp)) { if (!connection.getVersion().equals(legacyOpenMsg.getVersion())) { throw new ErrorMessageException(ErrorCodeNonExtended.VersionMismatch, new IncompatiblePeerVersionException(connection.getVersion(), legacyOpenMsg.getVersion())); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); return; } } } else if (message instanceof UpdateMessageLegacy) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessageLegacy) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); }
SxpLegacy extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageException, UpdateMessageConnectionStateException, ErrorMessageReceivedException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessageLegacy) { OpenMessageLegacy legacyOpenMsg = (OpenMessageLegacy) message; if (legacyOpenMsg.getType().equals(MessageType.Open)) { if (connection.isStateDeleteHoldDown()) { connection.setReconciliationTimer(); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); ByteBuf response = LegacyMessageFactory.createOpenResp(connection.getVersion(), connection.getMode()); LOG.info("{} Sent RESP {}", connection, MessageFactory.toString(response)); ctx.writeAndFlush(response); return; } } else if (legacyOpenMsg.getType().equals(MessageType.OpenResp)) { if (!connection.getVersion().equals(legacyOpenMsg.getVersion())) { throw new ErrorMessageException(ErrorCodeNonExtended.VersionMismatch, new IncompatiblePeerVersionException(connection.getVersion(), legacyOpenMsg.getVersion())); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); return; } } } else if (message instanceof UpdateMessageLegacy) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessageLegacy) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } }
SxpLegacy extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageException, UpdateMessageConnectionStateException, ErrorMessageReceivedException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessageLegacy) { OpenMessageLegacy legacyOpenMsg = (OpenMessageLegacy) message; if (legacyOpenMsg.getType().equals(MessageType.Open)) { if (connection.isStateDeleteHoldDown()) { connection.setReconciliationTimer(); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); ByteBuf response = LegacyMessageFactory.createOpenResp(connection.getVersion(), connection.getMode()); LOG.info("{} Sent RESP {}", connection, MessageFactory.toString(response)); ctx.writeAndFlush(response); return; } } else if (legacyOpenMsg.getType().equals(MessageType.OpenResp)) { if (!connection.getVersion().equals(legacyOpenMsg.getVersion())) { throw new ErrorMessageException(ErrorCodeNonExtended.VersionMismatch, new IncompatiblePeerVersionException(connection.getVersion(), legacyOpenMsg.getVersion())); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); return; } } } else if (message instanceof UpdateMessageLegacy) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessageLegacy) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } SxpLegacy(Version sxpVersion); }
SxpLegacy extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageException, UpdateMessageConnectionStateException, ErrorMessageReceivedException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessageLegacy) { OpenMessageLegacy legacyOpenMsg = (OpenMessageLegacy) message; if (legacyOpenMsg.getType().equals(MessageType.Open)) { if (connection.isStateDeleteHoldDown()) { connection.setReconciliationTimer(); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); ByteBuf response = LegacyMessageFactory.createOpenResp(connection.getVersion(), connection.getMode()); LOG.info("{} Sent RESP {}", connection, MessageFactory.toString(response)); ctx.writeAndFlush(response); return; } } else if (legacyOpenMsg.getType().equals(MessageType.OpenResp)) { if (!connection.getVersion().equals(legacyOpenMsg.getVersion())) { throw new ErrorMessageException(ErrorCodeNonExtended.VersionMismatch, new IncompatiblePeerVersionException(connection.getVersion(), legacyOpenMsg.getVersion())); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); return; } } } else if (message instanceof UpdateMessageLegacy) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessageLegacy) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } SxpLegacy(Version sxpVersion); @Override void onChannelActivation(ChannelHandlerContext ctx, SxpConnection connection); @Override void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message); @Override Notification onParseInput(ByteBuf request); @Override ByteBuf onUpdateMessage(SxpConnection connection, List<T> deleteBindings,
            List<T> addBindings, SxpBindingFilter bindingFilter); }
SxpLegacy extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageException, UpdateMessageConnectionStateException, ErrorMessageReceivedException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessageLegacy) { OpenMessageLegacy legacyOpenMsg = (OpenMessageLegacy) message; if (legacyOpenMsg.getType().equals(MessageType.Open)) { if (connection.isStateDeleteHoldDown()) { connection.setReconciliationTimer(); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); ByteBuf response = LegacyMessageFactory.createOpenResp(connection.getVersion(), connection.getMode()); LOG.info("{} Sent RESP {}", connection, MessageFactory.toString(response)); ctx.writeAndFlush(response); return; } } else if (legacyOpenMsg.getType().equals(MessageType.OpenResp)) { if (!connection.getVersion().equals(legacyOpenMsg.getVersion())) { throw new ErrorMessageException(ErrorCodeNonExtended.VersionMismatch, new IncompatiblePeerVersionException(connection.getVersion(), legacyOpenMsg.getVersion())); } setNodeIdRemote(connection); if (!checkModeMismatch(connection, legacyOpenMsg, ctx)) { connection.closeChannelHandlerContextComplements(ctx); connection.markChannelHandlerContext(ctx); connection.setStateOn(); LOG.info("{} Connected", connection); return; } } } else if (message instanceof UpdateMessageLegacy) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessageLegacy) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } SxpLegacy(Version sxpVersion); @Override void onChannelActivation(ChannelHandlerContext ctx, SxpConnection connection); @Override void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message); @Override Notification onParseInput(ByteBuf request); @Override ByteBuf onUpdateMessage(SxpConnection connection, List<T> deleteBindings,
            List<T> addBindings, SxpBindingFilter bindingFilter);  }