@Test public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); subscriptions.clearAllSubscriptions(); assertEquals(0, subscriptions.size()); m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); assertEquals(1, subscriptions.size()); Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); assertTrue(subscriptions.contains(expectedSubscription)); }
void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); subscribeSingleTopic(newSubscription, req.getTopic()); } SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } LOG.info("replying with SubAck to MSG ID " + msg.getMessageID()); session.write(ackMessage); }
ProtocolProcessor { void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); subscribeSingleTopic(newSubscription, req.getTopic()); } SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } LOG.info("replying with SubAck to MSG ID " + msg.getMessageID()); session.write(ackMessage); } }
ProtocolProcessor { void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); subscribeSingleTopic(newSubscription, req.getTopic()); } SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } LOG.info("replying with SubAck to MSG ID " + msg.getMessageID()); session.write(ackMessage); } ProtocolProcessor(); }
ProtocolProcessor { void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); subscribeSingleTopic(newSubscription, req.getTopic()); } SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } LOG.info("replying with SubAck to MSG ID " + msg.getMessageID()); session.write(ackMessage); } ProtocolProcessor();  }
ProtocolProcessor { void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); subscribeSingleTopic(newSubscription, req.getTopic()); } SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } LOG.info("replying with SubAck to MSG ID " + msg.getMessageID()); session.write(ackMessage); } ProtocolProcessor();   }