anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Minor refactory, renamed var names from session to channel

Browse Source
This commit is contained in:
andsel
2016-01-24 19:21:24 +01:00
parent 3be2667844
commit 43f7223a15
3 changed files with 100 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/ConnectionDescriptor.java
+4 -4
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+44 -44
View File
@@ -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<String> 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) {
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+52 -79
View File
@@ -57,7 +57,7 @@ public class ProtocolProcessorTest {
final static List<InterceptHandler> 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<Subscription> 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<QOSType> 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<Subscription> 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<String> 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<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
List<Subscription> 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<Subscription> 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<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {