@Test public void testOnInputMessageOpenBothModeConnectionErrHandling() throws Exception { OpenMessage message = mock(OpenMessage.class); when(message.getVersion()).thenReturn(Version.Version4); when(message.getType()).thenReturn(MessageType.Open); when(message.getPayload()).thenReturn(new byte[]{}); when(message.getAttribute()).thenReturn(getHoldTime()); when(message.getSxpMode()).thenReturn(ConnectionMode.Listener); when(connection.getMode()).thenReturn(ConnectionMode.Both); when(connection.isModeBoth()).thenReturn(true); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection, never()).setConnection(eq(message)); verify(connection, never()).setStateOff(channelHandlerContext); }
@Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageReceivedException, ErrorMessageException, UpdateMessageConnectionStateException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessage) { OpenMessage openMsg = (OpenMessage) message; if (openMsg.getType().equals(MessageType.Open)) { handleOpenMessage(ctx, connection, openMsg); return; } else if (openMsg.getType().equals(MessageType.OpenResp)) { handleOpenResponseMessage(ctx, connection, openMsg); return; } } else if (message instanceof UpdateMessage) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessage) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } else if (message instanceof KeepaliveMessage) { connection.setUpdateOrKeepaliveMessageTimestamp(); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); }
Sxpv4 extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageReceivedException, ErrorMessageException, UpdateMessageConnectionStateException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessage) { OpenMessage openMsg = (OpenMessage) message; if (openMsg.getType().equals(MessageType.Open)) { handleOpenMessage(ctx, connection, openMsg); return; } else if (openMsg.getType().equals(MessageType.OpenResp)) { handleOpenResponseMessage(ctx, connection, openMsg); return; } } else if (message instanceof UpdateMessage) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessage) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } else if (message instanceof KeepaliveMessage) { connection.setUpdateOrKeepaliveMessageTimestamp(); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } }
Sxpv4 extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageReceivedException, ErrorMessageException, UpdateMessageConnectionStateException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessage) { OpenMessage openMsg = (OpenMessage) message; if (openMsg.getType().equals(MessageType.Open)) { handleOpenMessage(ctx, connection, openMsg); return; } else if (openMsg.getType().equals(MessageType.OpenResp)) { handleOpenResponseMessage(ctx, connection, openMsg); return; } } else if (message instanceof UpdateMessage) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessage) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } else if (message instanceof KeepaliveMessage) { connection.setUpdateOrKeepaliveMessageTimestamp(); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } Sxpv4(SxpNode ownerNode); }
Sxpv4 extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageReceivedException, ErrorMessageException, UpdateMessageConnectionStateException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessage) { OpenMessage openMsg = (OpenMessage) message; if (openMsg.getType().equals(MessageType.Open)) { handleOpenMessage(ctx, connection, openMsg); return; } else if (openMsg.getType().equals(MessageType.OpenResp)) { handleOpenResponseMessage(ctx, connection, openMsg); return; } } else if (message instanceof UpdateMessage) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessage) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } else if (message instanceof KeepaliveMessage) { connection.setUpdateOrKeepaliveMessageTimestamp(); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } Sxpv4(SxpNode ownerNode); @Override void onChannelActivation(final 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); }
Sxpv4 extends AbstractStrategy { @Override public void onInputMessage(ChannelHandlerContext ctx, SxpConnection connection, Notification message) throws ErrorMessageReceivedException, ErrorMessageException, UpdateMessageConnectionStateException { LOG.info("{} Handle {}", connection, MessageFactory.toString(message)); if (message instanceof OpenMessage) { OpenMessage openMsg = (OpenMessage) message; if (openMsg.getType().equals(MessageType.Open)) { handleOpenMessage(ctx, connection, openMsg); return; } else if (openMsg.getType().equals(MessageType.OpenResp)) { handleOpenResponseMessage(ctx, connection, openMsg); return; } } else if (message instanceof UpdateMessage) { if (!connection.isStateOn(SxpConnection.ChannelHandlerContextType.LISTENER_CNTXT)) { throw new UpdateMessageConnectionStateException(connection.getState()); } connection.setUpdateOrKeepaliveMessageTimestamp(); connection.processUpdateMessage((UpdateMessage) message); return; } else if (message instanceof ErrorMessage) { throw new ErrorMessageReceivedException(((ErrorMessage) message).getInformation()); } else if (message instanceof PurgeAllMessage) { connection.getOwner().getSvcBindingHandler().processPurgeAllMessage(connection); return; } else if (message instanceof KeepaliveMessage) { connection.setUpdateOrKeepaliveMessageTimestamp(); return; } LOG.warn("{} Cannot handle message, ignoring: {}", connection, MessageFactory.toString(message)); } Sxpv4(SxpNode ownerNode); @Override void onChannelActivation(final 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);  }