From 43f7223a1511086f3e113ab210234d12d0a3683c Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 24 Jan 2016 19:17:31 +0100 Subject: [PATCH] Minor refactory, renamed var names from session to channel --- .../moquette/server/ConnectionDescriptor.java | 8 +- .../moquette/spi/impl/ProtocolProcessor.java | 88 ++++++------ .../spi/impl/ProtocolProcessorTest.java | 131 +++++++----------- 3 files changed, 100 insertions(+), 127 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 3249e65e..e9026c6f 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -27,12 +27,12 @@ import io.netty.channel.Channel; public class ConnectionDescriptor { public final String clientID; - public final Channel session; + public final Channel channel; public final boolean cleanSession; public ConnectionDescriptor(String clientID, Channel session, boolean cleanSession) { this.clientID = clientID; - this.session = session; + this.channel = session; this.cleanSession = cleanSession; } @@ -49,14 +49,14 @@ public class ConnectionDescriptor { ConnectionDescriptor that = (ConnectionDescriptor) o; if (clientID != null ? !clientID.equals(that.clientID) : that.clientID != null) return false; - return !(session != null ? !session.equals(that.session) : that.session != null); + return !(channel != null ? !channel.equals(that.channel) : that.channel != null); } @Override public int hashCode() { int result = clientID != null ? clientID.hashCode() : 0; - result = 31 * result + (session != null ? session.hashCode() : 0); + result = 31 * result + (channel != null ? channel.hashCode() : 0); return result; } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index a2b89fd0..da074903 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -135,21 +135,21 @@ public class ProtocolProcessor { m_sessionsStore = sessionsStore; } - public void processConnect(Channel session, ConnectMessage msg) { + public void processConnect(Channel channel, ConnectMessage msg) { LOG.debug("CONNECT for client <{}>", msg.getClientID()); if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); LOG.warn("processConnect sent bad proto ConnAck"); - session.writeAndFlush(badProto); - session.close(); + channel.writeAndFlush(badProto); + channel.close(); return; } if (msg.getClientID() == null || msg.getClientID().length() == 0) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); - session.writeAndFlush(okResp); + channel.writeAndFlush(okResp); m_interceptor.notifyClientConnected(msg); return; } @@ -160,17 +160,17 @@ public class ProtocolProcessor { if (msg.isPasswordFlag()) { pwd = msg.getPassword(); } else if (!this.allowAnonymous) { - failedCredentials(session); + failedCredentials(channel); return; } if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { - failedCredentials(session); - session.close(); + failedCredentials(channel); + channel.close(); return; } - NettyUtils.userName(session, msg.getUsername()); + NettyUtils.userName(channel, msg.getUsername()); } else if (!this.allowAnonymous) { - failedCredentials(session); + failedCredentials(channel); return; } @@ -178,28 +178,28 @@ public class ProtocolProcessor { if (m_clientIDs.containsKey(msg.getClientID())) { LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); //clean the subscriptions if the old used a cleanSession = true - Channel oldSession = m_clientIDs.get(msg.getClientID()).session; + Channel oldChannel = m_clientIDs.get(msg.getClientID()).channel; ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID()); oldClientSession.disconnect(); - NettyUtils.sessionStolen(oldSession, true); - oldSession.close(); + NettyUtils.sessionStolen(oldChannel, true); + oldChannel.close(); LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); } - ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); + ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), channel, msg.isCleanSession()); m_clientIDs.put(msg.getClientID(), connDescr); int keepAlive = msg.getKeepAlive(); LOG.debug("Connect with keepAlive {} s", keepAlive); - NettyUtils.keepAlive(session, keepAlive); + NettyUtils.keepAlive(channel, keepAlive); //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); - NettyUtils.cleanSession(session, msg.isCleanSession()); + NettyUtils.cleanSession(channel, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); - NettyUtils.clientID(session, msg.getClientID()); - LOG.debug("Connect create session <{}>", session); + NettyUtils.clientID(channel, msg.getClientID()); + LOG.debug("Connect create session <{}>", channel); - setIdleTime(session.pipeline(), Math.round(keepAlive * 1.5f)); + setIdleTime(channel.pipeline(), Math.round(keepAlive * 1.5f)); //Handle will flag if (msg.isWillFlag()) { @@ -219,7 +219,7 @@ public class ProtocolProcessor { if (!msg.isCleanSession() && isSessionAlreadyStored) { okResp.setSessionPresent(true); } - session.writeAndFlush(okResp); + channel.writeAndFlush(okResp); m_interceptor.notifyClientConnected(msg); if (!isSessionAlreadyStored) { @@ -485,22 +485,22 @@ public class ProtocolProcessor { // could happen is not an error HANDLE IT throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); } - Channel session = m_clientIDs.get(clientId).session; - LOG.debug("Session for clientId {} is {}", clientId, session); + Channel channel = m_clientIDs.get(clientId).channel; + LOG.debug("Session for clientId {} is {}", clientId, channel); - String user = NettyUtils.userName(session); + String user = NettyUtils.userName(channel); if (!m_authorizator.canRead(topic, user, clientId)) { LOG.debug("topic {} doesn't have read credentials", topic); return; } - session.writeAndFlush(pubMessage); //Or writeAdnFlush but flush is costly!! + channel.writeAndFlush(pubMessage); //Or writeAdnFlush but flush is costly!! } private void sendPubRec(String clientID, int messageID) { LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); - m_clientIDs.get(clientID).session.writeAndFlush(pubRecMessage); + m_clientIDs.get(clientID).channel.writeAndFlush(pubRecMessage); } private void sendPubAck(String clientId, int messageID) { @@ -516,7 +516,7 @@ public class ProtocolProcessor { if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } - m_clientIDs.get(clientId).session.writeAndFlush(pubAckMessage); + m_clientIDs.get(clientId).channel.writeAndFlush(pubAckMessage); } catch(Throwable t) { LOG.error(null, t); } @@ -526,8 +526,8 @@ public class ProtocolProcessor { * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. * */ - public void processPubRel(Channel session, PubRelMessage msg) { - String clientID = NettyUtils.clientID(session); + public void processPubRel(Channel channel, PubRelMessage msg) { + String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -552,11 +552,11 @@ public class ProtocolProcessor { PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - m_clientIDs.get(clientID).session.writeAndFlush(pubCompMessage); + m_clientIDs.get(clientID).channel.writeAndFlush(pubCompMessage); } - public void processPubRec(Channel session, PubRecMessage msg) { - String clientID = NettyUtils.clientID(session); + public void processPubRec(Channel channel, PubRecMessage msg) { + String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); @@ -569,11 +569,11 @@ public class ProtocolProcessor { pubRelMessage.setMessageID(messageID); pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE); - session.writeAndFlush(pubRelMessage); + channel.writeAndFlush(pubRelMessage); } - public void processPubComp(Channel session, PubCompMessage msg) { - String clientID = NettyUtils.clientID(session); + public void processPubComp(Channel channel, PubCompMessage msg) { + String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory @@ -582,15 +582,15 @@ public class ProtocolProcessor { targetSession.secondPhaseAcknowledged(messageID); } - public void processDisconnect(Channel session) throws InterruptedException { - String clientID = NettyUtils.clientID(session); - boolean cleanSession = NettyUtils.cleanSession(session); + public void processDisconnect(Channel channel) throws InterruptedException { + String clientID = NettyUtils.clientID(channel); + boolean cleanSession = NettyUtils.cleanSession(channel); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); clientSession.disconnect(); m_clientIDs.remove(clientID); - session.close(); + channel.close(); //cleanup the will store m_willStore.remove(clientID); @@ -621,10 +621,10 @@ public class ProtocolProcessor { * Remove the clientID from topic subscription, if not previously subscribed, * doesn't reply any error */ - public void processUnsubscribe(Channel session, UnsubscribeMessage msg) { + public void processUnsubscribe(Channel channel, UnsubscribeMessage msg) { List topics = msg.topicFilters(); int messageID = msg.getMessageID(); - String clientID = NettyUtils.clientID(session); + String clientID = NettyUtils.clientID(channel); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); @@ -634,7 +634,7 @@ public class ProtocolProcessor { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { //close the connection, not valid topicFilter is a protocol violation - session.close(); + channel.close(); LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID); return; } @@ -649,11 +649,11 @@ public class ProtocolProcessor { ackMessage.setMessageID(messageID); LOG.info("replying with UnsubAck to MSG ID {}", messageID); - session.writeAndFlush(ackMessage); + channel.writeAndFlush(ackMessage); } - public void processSubscribe(Channel session, SubscribeMessage msg) { - String clientID = NettyUtils.clientID(session); + public void processSubscribe(Channel channel, SubscribeMessage msg) { + String clientID = NettyUtils.clientID(channel); LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); @@ -679,7 +679,7 @@ public class ProtocolProcessor { if (LOG.isTraceEnabled()) { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } - session.writeAndFlush(ackMessage); + channel.writeAndFlush(ackMessage); //fire the publish for(Subscription subscription : newSubscriptions) { diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 4bbdc327..db48d3e8 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -57,7 +57,7 @@ public class ProtocolProcessorTest { final static List EMPTY_OBSERVERS = Collections.emptyList(); final static BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); - EmbeddedChannel m_session; + EmbeddedChannel m_channel; ConnectMessage connMsg; ProtocolProcessor m_processor; @@ -71,7 +71,7 @@ public class ProtocolProcessorTest { connMsg = new ConnectMessage(); connMsg.setProtocolVersion((byte) 0x03); - m_session = new EmbeddedChannel(); + m_channel = new EmbeddedChannel(); //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); @@ -103,7 +103,7 @@ public class ProtocolProcessorTest { @Override public List matches(String topic) { if (topic.equals(FAKE_TOPIC)) { - return Arrays.asList(subscription); + return Collections.singletonList(subscription); } else { throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); } @@ -120,7 +120,7 @@ public class ProtocolProcessorTest { connectMessage.setClientID(FAKE_CLIENT_ID); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); connectMessage.setCleanSession(true); - m_processor.processConnect(m_session, connectMessage); + m_processor.processConnect(m_channel, connectMessage); //Exercise @@ -130,11 +130,11 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - NettyUtils.userName(m_session, "FakeCLI"); - m_processor.processPublish(m_session, msg); + NettyUtils.userName(m_channel, "FakeCLI"); + m_processor.processPublish(m_channel, msg); //Verify - assertNotNull(m_session.readOutbound()); + assertNotNull(m_channel.readOutbound()); //TODO check received message attributes } @@ -163,23 +163,23 @@ public class ProtocolProcessorTest { subs.init(storageService.sessionsStore()); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); - EmbeddedChannel firstReceiverSession = new EmbeddedChannel(); + EmbeddedChannel firstReceiverChannel = new EmbeddedChannel(); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(FAKE_CLIENT_ID); connectMessage.setCleanSession(true); - m_processor.processConnect(firstReceiverSession, connectMessage); - assertConnAckAccepted(firstReceiverSession); + m_processor.processConnect(firstReceiverChannel, connectMessage); + assertConnAckAccepted(firstReceiverChannel); //connect the second fake subscriber - EmbeddedChannel secondReceiverSession = new EmbeddedChannel(); + EmbeddedChannel secondReceiverChannel = new EmbeddedChannel(); ConnectMessage connectMessage2 = new ConnectMessage(); connectMessage2.setProtocolVersion((byte) 3); connectMessage2.setClientID(FAKE_CLIENT_ID2); connectMessage2.setCleanSession(true); connectMessage2.setCleanSession(true); - m_processor.processConnect(secondReceiverSession, connectMessage2); - assertConnAckAccepted(secondReceiverSession); + m_processor.processConnect(secondReceiverChannel, connectMessage2); + assertConnAckAccepted(secondReceiverChannel); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -189,17 +189,17 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - NettyUtils.userName(m_session, "FakeCLI"); - m_processor.processPublish(m_session, msg); + NettyUtils.userName(m_channel, "FakeCLI"); + m_processor.processPublish(m_channel, msg); //Verify - PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverSession.readOutbound(); + PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverChannel.readOutbound(); assertNotNull(pub2FirstSubscriber); String firstMessageContent = DebugUtils.payload2Str(pub2FirstSubscriber.getPayload()); assertEquals("Hello", firstMessageContent); - PublishMessage pub2SecondSubscriber = (PublishMessage) secondReceiverSession.readOutbound(); + PublishMessage pub2SecondSubscriber = (PublishMessage) secondReceiverChannel.readOutbound(); assertNotNull(pub2SecondSubscriber); String secondMessageContent = DebugUtils.payload2Str(pub2SecondSubscriber.getPayload()); assertEquals("Hello", secondMessageContent); @@ -210,13 +210,13 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - NettyUtils.clientID(m_session, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_session, false); + NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); - m_processor.processSubscribe(m_session, msg); + m_processor.processSubscribe(m_channel, msg); //Verify - assertTrue(m_session.readOutbound() instanceof SubAckMessage); + assertTrue(m_channel.readOutbound() instanceof SubAckMessage); Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); } @@ -225,15 +225,15 @@ public class ProtocolProcessorTest { public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - NettyUtils.clientID(m_session, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_session, false); + NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); - m_processor.processSubscribe(m_session, msg); + m_processor.processSubscribe(m_channel, msg); //Exercise - m_processor.processSubscribe(m_session, msg); + m_processor.processSubscribe(m_channel, msg); //Verify assertEquals(1, subscriptions.size()); @@ -246,17 +246,17 @@ public class ProtocolProcessorTest { public void testSubscribeWithBadFormattedTopic() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); - NettyUtils.clientID(m_session, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_session, false); + NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); //Exercise - m_processor.processSubscribe(m_session, msg); + m_processor.processSubscribe(m_channel, msg); //Verify assertEquals(0, subscriptions.size()); - Object recvSubAckMessage = m_session.readOutbound(); + Object recvSubAckMessage = m_channel.readOutbound(); assertTrue(recvSubAckMessage instanceof SubAckMessage); List qosSubAcked = ((SubAckMessage)recvSubAckMessage).types(); assertEquals(1, qosSubAcked.size()); @@ -271,14 +271,14 @@ public class ProtocolProcessorTest { UnsubscribeMessage msg = new UnsubscribeMessage(); msg.setMessageID(1); msg.addTopicFilter(BAD_FORMATTED_TOPIC); - NettyUtils.clientID(m_session, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_session, false); + NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_channel, false); //Exercise - m_processor.processUnsubscribe(m_session, msg); + m_processor.processUnsubscribe(m_channel, msg); //Verify - assertFalse("If client unsubscribe with bad topic than channel must be closed", m_session.isOpen()); + assertFalse("If client unsubscribe with bad topic than channel must be closed", m_channel.isOpen()); } @@ -293,7 +293,7 @@ public class ProtocolProcessorTest { @Override public List matches(String topic) { if (topic.equals(FAKE_TOPIC)) { - return Arrays.asList(subscription); + return Collections.singletonList(subscription); } else { throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); } @@ -309,26 +309,26 @@ public class ProtocolProcessorTest { connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProtocolVersion((byte) 3); connectMessage.setCleanSession(true); - m_processor.processConnect(m_session, connectMessage); - assertConnAckAccepted(m_session); + m_processor.processConnect(m_channel, connectMessage); + assertConnAckAccepted(m_channel); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage pubmsg = new PublishMessage(); pubmsg.setTopicName(FAKE_TOPIC); pubmsg.setQos(QOSType.MOST_ONE); pubmsg.setPayload(buffer); pubmsg.setRetainFlag(true); - NettyUtils.clientID(m_session, FAKE_PUBLISHER_ID); - m_processor.processPublish(m_session, pubmsg); - NettyUtils.cleanSession(m_session, false); + NettyUtils.clientID(m_channel, FAKE_PUBLISHER_ID); + m_processor.processPublish(m_channel, pubmsg); + NettyUtils.cleanSession(m_channel, false); //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(QOSType.MOST_ONE.byteValue(), "#")); - m_processor.processSubscribe(m_session, msg); + m_processor.processSubscribe(m_channel, msg); //Verify //wait the latch - Object pubMessage = m_session.readOutbound(); + Object pubMessage = m_channel.readOutbound(); assertNotNull(pubMessage); assertTrue(pubMessage instanceof PublishMessage); assertEquals(FAKE_TOPIC, ((PublishMessage) pubMessage).getTopicName()); @@ -351,7 +351,7 @@ public class ProtocolProcessorTest { connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProtocolVersion((byte) 3); connectMessage.setCleanSession(false); - m_processor.processConnect(m_session, connectMessage); + m_processor.processConnect(m_channel, connectMessage); //Verify no messages are still stored Collection guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID); @@ -365,10 +365,11 @@ public class ProtocolProcessorTest { SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE); - List inactiveSubscriptions = Arrays.asList(inactiveSub); + List inactiveSubscriptions = Collections.singletonList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), + NO_OBSERVERS_INTERCEPTOR); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -377,41 +378,13 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - NettyUtils.clientID(m_session, "Publisher"); - m_processor.processPublish(m_session, msg); + NettyUtils.clientID(m_channel, "Publisher"); + m_processor.processPublish(m_channel, msg); //Verify no message is received - assertNull(m_session.readOutbound()); + assertNull(m_channel.readOutbound()); } - - @Test - public void publishToAnInactiveSubscriptionsCleanSession() { - //create an inactive session for Subscriber - m_sessionStore.createNewSession("Subscriber", false).deactivate(); - SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); - Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE); - List inactiveSubscriptions = Arrays.asList(inactiveSub); - when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); - m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR); - - //Exercise - ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); - PublishMessage msg = new PublishMessage(); - msg.setTopicName("/topic"); - msg.setQos(QOSType.MOST_ONE); - msg.setPayload(buffer); - msg.setRetainFlag(true); - NettyUtils.clientID(m_session, "Publisher"); - m_processor.processPublish(m_session, msg); - - //Verify no message is received - assertNull(m_session.readOutbound()); - } - - /** * Verify that receiving a publish with retained message and with Q0S = 0 * clean the existing retained messages for that topic. @@ -420,9 +393,9 @@ public class ProtocolProcessorTest { public void testCleanRetainedStoreAfterAQoS0AndRetainedTrue() { //force a connect connMsg.setClientID("Publisher"); - m_processor.processConnect(m_session, connMsg); + m_processor.processConnect(m_channel, connMsg); //prepare and existing retained store - NettyUtils.clientID(m_session, "Publisher"); + NettyUtils.clientID(m_channel, "Publisher"); ByteBuffer payload = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage msg = new PublishMessage(); msg.setTopicName(FAKE_TOPIC); @@ -430,7 +403,7 @@ public class ProtocolProcessorTest { msg.setPayload(payload); msg.setRetainFlag(true); msg.setMessageID(100); - m_processor.processPublish(m_session, msg); + m_processor.processPublish(m_channel, msg); //Exercise PublishMessage cleanPubMsg = new PublishMessage(); @@ -438,7 +411,7 @@ public class ProtocolProcessorTest { cleanPubMsg.setPayload(payload); cleanPubMsg.setQos(QOSType.MOST_ONE); cleanPubMsg.setRetainFlag(true); - m_processor.processPublish(m_session, cleanPubMsg); + m_processor.processPublish(m_channel, cleanPubMsg); //Verify Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() {