@Test public void testOnInputMessageOpen() throws Exception { OpenMessageLegacy message = mock(OpenMessageLegacy.class); when(message.getVersion()).thenReturn(Version.Version1); when(message.getType()).thenReturn(MessageType.Open); when(message.getPayload()).thenReturn(new byte[] {}); when(message.getSxpMode()).thenReturn(ConnectionMode.Listener); when(connection.isModeListener()).thenReturn(true); sxpLegacy.onInputMessage(channelHandlerContext, connection, message); verify(connection).setStateOff(any(ChannelHandlerContext.class)); verify(connection, times(0)).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(connection, times(0)).setStateOn(); verify(channelHandlerContext).writeAndFlush(isNotNull()); when(connection.isModeListener()).thenReturn(false); when(connection.isModeSpeaker()).thenReturn(true); sxpLegacy.onInputMessage(channelHandlerContext, connection, message); verify(connection).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(connection).setStateOn(); verify(channelHandlerContext, times(2)).writeAndFlush(isNotNull()); when(message.getSxpMode()).thenReturn(ConnectionMode.Speaker); when(connection.isModeListener()).thenReturn(true); when(connection.isModeSpeaker()).thenReturn(false); sxpLegacy.onInputMessage(channelHandlerContext, connection, message); verify(connection, times(2)).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(connection, times(2)).setStateOn(); verify(channelHandlerContext, times(3)).writeAndFlush(isNotNull()); when(message.getType()).thenReturn(MessageType.OpenResp); sxpLegacy.onInputMessage(channelHandlerContext, connection, message); when(message.getSxpMode()).thenReturn(ConnectionMode.Listener); when(connection.isModeListener()).thenReturn(true); sxpLegacy.onInputMessage(channelHandlerContext, connection, message); verify(connection, times(2)).setStateOff(any(ChannelHandlerContext.class)); verify(connection, times(3)).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(connection, times(3)).setStateOn(); verify(channelHandlerContext, times(4)).writeAndFlush(isNotNull()); when(connection.getVersion()).thenReturn(Version.Version4); exception.expect(ErrorMessageException.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);  }