@Test public void testOnInputMessageOpen() throws Exception { List<Attribute> attributes = new ArrayList<>(); attributes.addAll(getHoldTime()); attributes.addAll(getCapabilities()); 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(attributes); when(message.getSxpMode()).thenReturn(ConnectionMode.Listener); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection).setConnection(any(OpenMessage.class)); verify(connection, never()).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(channelHandlerContext).writeAndFlush(isNotNull()); when(message.getSxpMode()).thenReturn(ConnectionMode.Speaker); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection, times(2)).setConnection(any(OpenMessage.class)); verify(connection, never()).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(channelHandlerContext, times(2)).writeAndFlush(isNotNull()); when(connection.isStateDeleteHoldDown()).thenReturn(true); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection).closeChannelHandlerContextComplements(any(ChannelHandlerContext.class)); verify(connection, times(3)).setConnection(any(OpenMessage.class)); verify(channelHandlerContext, times(3)).writeAndFlush(isNotNull()); when(connection.isStateDeleteHoldDown()).thenReturn(false); when(connection.isStatePendingOn()).thenReturn(true); Channel channel = mock(Channel.class); InetSocketAddress localAddress = new InetSocketAddress("0.0.0.20", 0); InetSocketAddress remoteAddress = new InetSocketAddress("0.0.0.10", 0); when(channel.localAddress()).thenReturn(localAddress); when(channel.remoteAddress()).thenReturn(remoteAddress); when(channelHandlerContext.channel()).thenReturn(channel); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection).closeInitContextWithRemote(any(InetSocketAddress.class)); verify(channelHandlerContext, times(3)).writeAndFlush(isNotNull()); when(connection.getDestination()).thenReturn(remoteAddress); when(connection.isStateDeleteHoldDown()).thenReturn(false); when(connection.isStatePendingOn()).thenReturn(false); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection,times(4)).markChannelHandlerContext(any(ChannelHandlerContext.class)); verify(connection, times(4)).setConnection(any(OpenMessage.class)); verify(channelHandlerContext, times(4)).writeAndFlush(isNotNull()); when(message.getSxpMode()).thenReturn(ConnectionMode.Listener); when(connection.isModeBoth()).thenReturn(true); when(connection.isBidirectionalBoth()).thenReturn(false); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection).setConnectionSpeakerPart(any(OpenMessage.class)); verify(connection).markChannelHandlerContext(any(ChannelHandlerContext.class), any(SxpConnection.ChannelHandlerContextType.class)); verify(channelHandlerContext, times(5)).writeAndFlush(isNotNull()); when(message.getSxpMode()).thenReturn(ConnectionMode.Speaker); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(channelHandlerContext, times(6)).writeAndFlush(isNotNull()); verify(connection).setStateOff(any(ChannelHandlerContext.class)); when(connection.isBidirectionalBoth()).thenReturn(true); sxpv4.onInputMessage(channelHandlerContext, connection, message); verify(connection, times(5)).setConnection(any(OpenMessage.class)); }
@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);  }