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

Merge branch 'split_protocol_processor_in_message_handlers'

Browse Source
This commit is contained in:
andsel
2018-07-20 18:20:29 +02:00
parent 7546a509da ba8ad02289
commit 5d6435cf40
16 changed files with 940 additions and 715 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/connections/IConnectionsManager.java
+1 -3
View File
@@ -39,10 +39,8 @@ public interface IConnectionsManager {
boolean removeConnection(ConnectionDescriptor descriptor);
ConnectionDescriptor getConnection(String clientID);
/**
* Determines wether a MQTT client is connected to the broker.
* Determines weather a MQTT client is connected to the broker.
*
* @param clientID
* @return
broker/src/main/java/io/moquette/server/ConnectionDescriptor.java
+8 -8
View File
@@ -22,6 +22,7 @@ import io.moquette.server.netty.metrics.BytesMetricsHandler;
import io.moquette.server.netty.metrics.MessageMetrics;
import io.moquette.server.netty.metrics.MessageMetricsHandler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,8 +43,8 @@ public class ConnectionDescriptor {
public enum ConnectionState {
// Connection states
DISCONNECTED,
INIT_SESSION,
SENDACK,
SESSION_CREATED,
MESSAGES_REPUBLISHED,
ESTABLISHED,
// Disconnection states
@@ -67,6 +68,10 @@ public class ConnectionDescriptor {
this.channel.writeAndFlush(payload).addListener(FIRE_EXCEPTION_ON_FAILURE);
}
public void writeAndFlush(Object payload, ChannelFutureListener listener) {
this.channel.writeAndFlush(payload).addListener(listener);
}
public boolean doesNotUseChannel(Channel channel) {
return !(this.channel.equals(channel));
}
@@ -86,13 +91,8 @@ public class ConnectionDescriptor {
}
public void abort() {
LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID);
// try {
// this.channel.disconnect().sync();
this.channel.close().addListener(CLOSE_ON_FAILURE); // .sync();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
LOG.info("Closing connection descriptor clientId = {}.", clientID);
this.channel.close().addListener(CLOSE_ON_FAILURE);
}
public boolean assignState(ConnectionState expected, ConnectionState newState) {
broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java
-5
View File
@@ -62,11 +62,6 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
return connectionDescriptors.remove(descriptor.clientID, descriptor);
}
@Override
public ConnectionDescriptor getConnection(String clientID) {
return connectionDescriptors.get(clientID);
}
@Override
public boolean isConnected(String clientID) {
return connectionDescriptors.containsKey(clientID);
broker/src/main/java/io/moquette/spi/ClientSession.java
+10 -2
View File
@@ -16,8 +16,10 @@
package io.moquette.spi;
import io.moquette.spi.impl.WillMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import java.util.HashSet;
import java.util.Set;
@@ -111,10 +113,16 @@ public abstract class ClientSession {
public abstract int getInflightMessagesNo();
public abstract void wipeSubscriptions();
public abstract void storeWillMessage(MqttConnectMessage msg, String clientId);
public abstract void removeWill();
public abstract WillMessage willMessage();
@Override
public String toString() {
return getClass().getSimpleName() + "{clientID='" + clientID + "'}";
}
public abstract void wipeSubscriptions();
}
broker/src/main/java/io/moquette/spi/DurableSession.java
+36 -1
View File
@@ -15,14 +15,20 @@
*/
package io.moquette.spi;
import io.moquette.spi.impl.WillMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
public class DurableSession extends ClientSession {
@@ -56,11 +62,14 @@ public class DurableSession extends ClientSession {
private final OutboundFlightZone outboundFlightZone;
private final InboundFlightZone inboundFlightZone;
private final ConcurrentMap<String, WillMessage> willMessages;
public DurableSession(String clientID, ISessionsStore sessions, ISubscriptionsStore subscriptionsStore) {
public DurableSession(String clientID, ISessionsStore sessions, ISubscriptionsStore subscriptionsStore,
ConcurrentMap<String, WillMessage> willMessages) {
super(clientID);
this.subscriptionsStore = subscriptionsStore;
this.sessionsStore = sessions;
this.willMessages = willMessages;
this.outboundFlightZone = new OutboundFlightZone();
this.inboundFlightZone = new InboundFlightZone();
}
@@ -215,4 +224,30 @@ public class DurableSession extends ClientSession {
public void wipeSubscriptions() {
this.subscriptionsStore.wipeSubscriptions(clientID);
}
@Override
public void storeWillMessage(MqttConnectMessage msg, String clientId) {
// Handle will flag
if (msg.variableHeader().isWillFlag()) {
MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos());
LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}",
clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain());
byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8);
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
// save the will testament in the clientID store
WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos);
willMessages.put(clientId, will);
LOG.debug("MQTT last will and testament has been configured. CId={}", clientId);
}
}
@Override
public void removeWill() {
willMessages.remove(clientID);
}
@Override
public WillMessage willMessage() {
return willMessages.get(clientID);
}
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+1
View File
@@ -18,6 +18,7 @@ package io.moquette.spi;
import io.moquette.persistence.PersistentSession;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.WillMessage;
import java.time.LocalDateTime;
import java.util.Collection;
broker/src/main/java/io/moquette/spi/TransientSession.java
+37 -1
View File
@@ -16,11 +16,17 @@
package io.moquette.spi;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.ProtocolProcessor;
import io.moquette.spi.impl.WillMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -37,9 +43,12 @@ public class TransientSession extends ClientSession {
private final ConcurrentMap<Integer, StoredMessage> inboundInflightMap = new ConcurrentHashMap<>();
private final ConcurrentMap<Integer, StoredMessage> outboundInflightMap = new ConcurrentHashMap<>();
private final Map<Integer, StoredMessage> secondPhaseStore = new ConcurrentHashMap<>();
private WillMessage will;
private final ConcurrentMap<String, WillMessage> willMessages;
public TransientSession(String clientID) {
public TransientSession(String clientID, ConcurrentMap<String, WillMessage> willMessages) {
super(clientID);
this.willMessages = willMessages;
}
@Override
@@ -186,4 +195,31 @@ public class TransientSession extends ClientSession {
public void wipeSubscriptions() {
// it's transient, no need to wide subs because wasn't stored
}
@Override
public void storeWillMessage(MqttConnectMessage msg, String clientId) {
// Handle will flag
if (msg.variableHeader().isWillFlag()) {
MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos());
LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}",
clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain());
byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8);
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
// save the will testament in the clientID store
will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos);
willMessages.put(clientId, will);
LOG.debug("MQTT last will and testament has been configured. CId={}", clientId);
}
}
@Override
public void removeWill() {
willMessages.remove(clientID);
will = null;
}
@Override
public WillMessage willMessage() {
return will;
}
}
broker/src/main/java/io/moquette/spi/impl/ConnectHandler.java
+285
View File
@@ -0,0 +1,285 @@
package io.moquette.spi.impl;
import io.moquette.connections.IConnectionsManager;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.netty.AutoFlushHandler;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.NoSuchElementException;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
public class ConnectHandler {
private static final Logger LOG = LoggerFactory.getLogger(ConnectHandler.class);
private IConnectionsManager connectionDescriptors;
private BrokerInterceptor m_interceptor;
private SessionsRepository sessionsRepository;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
private IAuthorizator m_authorizator;
private ISubscriptionsDirectory subscriptions;
private boolean allowAnonymous;
private boolean allowZeroByteClientId;
private boolean reauthorizeSubscriptionsOnConnect;
private InternalRepublisher internalRepublisher;
ConnectHandler(IConnectionsManager connectedClients, BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
ISessionsStore sessionsStore, IAuthenticator authenticator, IAuthorizator authorizator,
ISubscriptionsDirectory subscriptions, boolean allowAnonymous, boolean allowZeroByteClientId,
boolean reauthorizeSubscriptionsOnConnect, InternalRepublisher internalRepublisher) {
this.connectionDescriptors = connectedClients;
this.m_interceptor = interceptor;
this.sessionsRepository = sessionsRepository;
this.m_sessionsStore = sessionsStore;
this.m_authenticator = authenticator;
this.m_authorizator = authorizator;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
this.allowZeroByteClientId = allowZeroByteClientId;
this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect;
this.internalRepublisher = internalRepublisher;
}
public void processConnect(Channel channel, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientId = payload.clientIdentifier();
final String username = payload.userName();
LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, username);
if (isNotProtocolVersion(msg, MqttVersion.MQTT_3_1) && isNotProtocolVersion(msg, MqttVersion.MQTT_3_1_1)) {
MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
LOG.error("MQTT protocol version is not valid. CId={}", clientId);
channel.writeAndFlush(badProto).addListener(FIRE_EXCEPTION_ON_FAILURE);
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
final boolean cleanSession = msg.variableHeader().isCleanSession();
if (clientId == null || clientId.length() == 0) {
if (!cleanSession || !this.allowZeroByteClientId) {
MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
channel.writeAndFlush(badId).addListener(FIRE_EXCEPTION_ON_FAILURE);
channel.close().addListener(CLOSE_ON_FAILURE);
LOG.error("MQTT client ID cannot be empty. Username={}", username);
return;
}
// Generating client id.
clientId = UUID.randomUUID().toString().replace("-", "");
LOG.info("Client has connected with server generated id={}, username={}", clientId, username);
}
if (!login(channel, msg, clientId)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession);
final ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
if (existing != null) {
LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId);
existing.abort();
//return;
this.connectionDescriptors.removeConnection(existing);
this.connectionDescriptors.addConnection(descriptor);
}
if (!descriptor.assignState(DISCONNECTED, INIT_SESSION)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
LOG.debug("Initializing client session {}", clientId);
ClientSession existingSession = this.sessionsRepository.sessionForClient(clientId);
boolean isSessionAlreadyStored = existingSession != null;
final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession();
if (isSessionAlreadyStored && msgCleanSessionFlag) {
for (Subscription existingSub : existingSession.getSubscriptions()) {
this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientId);
}
}
initializeKeepAliveTimeout(channel, msg, clientId);
final ClientSession clientSession = this.sessionsRepository.createOrLoadClientSession(clientId, cleanSession);
clientSession.storeWillMessage(msg, clientId);
int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */;
setupAutoFlusher(channel, flushIntervalMs);
if (!cleanSession && reauthorizeSubscriptionsOnConnect) {
reauthorizeOnExistingSubscriptions(clientId, username);
}
if (!descriptor.assignState(INIT_SESSION, SENDACK)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
LOG.debug("Sending CONNACK. CId={}", clientId);
MqttConnAckMessage okResp = createConnectAck(msg, clientId);
final String connectClientId = clientId;
descriptor.writeAndFlush(okResp, new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
if (future.isSuccess()) {
LOG.debug("CONNACK has been sent. CId={}", connectClientId);
if (!descriptor.assignState(SENDACK, MESSAGES_REPUBLISHED)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
m_interceptor.notifyClientConnected(msg);
if (!msg.variableHeader().isCleanSession()) {
// force the republish of stored QoS1 and QoS2
internalRepublisher.publishStored(clientSession);
}
if (!descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED)) {
channel.close().addListener(CLOSE_ON_FAILURE);
}
LOG.info("Connected client <{}> with login <{}>", connectClientId, username);
} else {
future.channel().pipeline().fireExceptionCaught(future.cause());
}
}
});
}
private boolean isNotProtocolVersion(MqttConnectMessage msg, MqttVersion version) {
return msg.variableHeader().version() != version.protocolLevel();
}
private void reauthorizeOnExistingSubscriptions(String clientId, String username) {
if (!m_sessionsStore.contains(clientId)) {
return;
}
final Collection<Subscription> clientSubscriptions = m_sessionsStore.subscriptionStore()
.listClientSubscriptions(clientId);
for (Subscription sub : clientSubscriptions) {
final Topic topicToReauthorize = sub.getTopicFilter();
final boolean readAuthorized = m_authorizator.canRead(topicToReauthorize, username, clientId);
if (!readAuthorized) {
subscriptions.removeSubscription(topicToReauthorize, clientId);
}
}
}
private void setupAutoFlusher(Channel channel, int flushIntervalMs) {
try {
channel.pipeline().addAfter(
"idleEventHandler",
"autoFlusher",
new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
} catch (NoSuchElementException nseex) {
// the idleEventHandler is not present on the pipeline
channel.pipeline()
.addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
}
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
return connAck(returnCode, false);
}
private MqttConnAckMessage connAckWithSessionPresent(MqttConnectReturnCode returnCode) {
return connAck(returnCode, true);
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE,
false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);
return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) {
// handle user authentication
if (msg.variableHeader().hasUserName()) {
byte[] pwd = null;
if (msg.variableHeader().hasPassword()) {
pwd = msg.payload().password().getBytes(StandardCharsets.UTF_8);
} else if (!this.allowAnonymous) {
LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId);
failedCredentials(channel);
return false;
}
final String login = msg.payload().userName();
if (!m_authenticator.checkValid(clientId, login, pwd)) {
LOG.error("Authenticator has rejected the MQTT credentials CId={}, username={}", clientId, login);
failedCredentials(channel);
return false;
}
NettyUtils.userName(channel, login);
} else if (!this.allowAnonymous) {
LOG.error("Client didn't supply any credentials and MQTT anonymous mode is disabled. CId={}", clientId);
failedCredentials(channel);
return false;
}
return true;
}
private MqttConnAckMessage createConnectAck(MqttConnectMessage msg, String clientId) {
MqttConnAckMessage okResp;
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientId);
boolean isSessionAlreadyStored = clientSession != null;
final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession();
if (!msgCleanSessionFlag && isSessionAlreadyStored) {
okResp = connAckWithSessionPresent(CONNECTION_ACCEPTED);
} else {
okResp = connAck(CONNECTION_ACCEPTED);
}
return okResp;
}
private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) {
int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
NettyUtils.keepAlive(channel, keepAlive);
NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
NettyUtils.clientID(channel, clientId);
int idleTime = Math.round(keepAlive * 1.5f);
setIdleTime(channel.pipeline(), idleTime);
LOG.debug("Connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}",
clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
}
private void failedCredentials(Channel session) {
session.writeAndFlush(connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD))
.addListener(FIRE_EXCEPTION_ON_FAILURE);
LOG.info("Client {} failed to connect with bad username or password.", session);
}
private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
if (pipeline.names().contains("idleStateHandler")) {
pipeline.remove("idleStateHandler");
}
pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0));
}
}
broker/src/main/java/io/moquette/spi/impl/DisconnectHandler.java
+133
View File
@@ -0,0 +1,133 @@
package io.moquette.spi.impl;
import io.moquette.connections.IConnectionsManager;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Optional;
import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
public class DisconnectHandler {
private static final Logger LOG = LoggerFactory.getLogger(DisconnectHandler.class);
private final IConnectionsManager connectionDescriptors;
private final SessionsRepository sessionsRepository;
private final ISubscriptionsDirectory subscriptions;
private final BrokerInterceptor m_interceptor;
public DisconnectHandler(IConnectionsManager connectionDescriptors, SessionsRepository sessionsRepository,
ISubscriptionsDirectory subscriptions, BrokerInterceptor interceptor) {
this.connectionDescriptors = connectionDescriptors;
this.sessionsRepository = sessionsRepository;
this.subscriptions = subscriptions;
this.m_interceptor = interceptor;
}
public void processDisconnect(Channel channel, String clientID) {
final Optional<ConnectionDescriptor> existingDescriptorOpt = this.connectionDescriptors.lookupDescriptor(clientID);
if (!existingDescriptorOpt.isPresent()) {
// another client with same ID removed the descriptor, we must exit
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
final ConnectionDescriptor existingDescriptor = existingDescriptorOpt.get();
if (existingDescriptor.doesNotUseChannel(channel)) {
// another client saved it's descriptor, exit
LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!removeSubscriptions(existingDescriptor, clientID)) {
LOG.warn("Unable to remove subscriptions. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!dropStoredMessages(existingDescriptor, clientID)) {
LOG.warn("Unable to drop stored messages. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) {
LOG.warn("Unable to drop will message. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!existingDescriptor.close()) {
LOG.debug("Connection has been closed. CId={}", clientID);
return;
}
boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor);
if (!stillPresent) {
// another descriptor was inserted
LOG.warn("Another descriptor has been inserted. CId={}", clientID);
return;
}
this.sessionsRepository.disconnect(clientID);
LOG.info("Client <{}> disconnected", clientID);
}
private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) {
final boolean success = descriptor.assignState(ESTABLISHED, SUBSCRIPTIONS_REMOVED);
if (!success) {
return false;
}
if (descriptor.cleanSession) {
LOG.trace("Removing saved subscriptions. CId={}", descriptor.clientID);
final ClientSession session = this.sessionsRepository.sessionForClient(clientID);
session.wipeSubscriptions();
for (Subscription existingSub : session.getSubscriptions()) {
this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientID);
}
LOG.trace("Saved subscriptions have been removed. CId={}", descriptor.clientID);
}
return true;
}
private boolean dropStoredMessages(ConnectionDescriptor descriptor, String clientID) {
final boolean success = descriptor.assignState(SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED);
if (!success) {
return false;
}
final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
if (clientSession.isCleanSession()) {
clientSession.dropQueue();
}
return true;
}
private boolean cleanWillMessageAndNotifyInterceptor(ConnectionDescriptor descriptor, String clientID) {
final boolean success = descriptor.assignState(MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED);
if (!success) {
return false;
}
LOG.trace("Removing will message. ClientId={}", descriptor.clientID);
// cleanup the will store
final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
clientSession.removeWill();
String username = descriptor.getUsername();
m_interceptor.notifyClientDisconnected(clientID, username);
return true;
}
}
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+1
View File
@@ -59,6 +59,7 @@ class InternalRepublisher {
return;
}
LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
EnqueuedMessage pubEvt;
while ((pubEvt = clientSession.poll()) != null) {
// put in flight zone
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+26 -383
View File
@@ -19,13 +19,14 @@ package io.moquette.spi.impl;
import io.moquette.connections.IConnectionsManager;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.persistence.MemorySessionStore;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.AutoFlushHandler;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.*;
import io.moquette.spi.ClientSession;
import io.moquette.spi.EnqueuedMessage;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
@@ -33,18 +34,16 @@ import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
@@ -52,7 +51,6 @@ import static io.moquette.spi.impl.Utils.messageId;
import static io.moquette.spi.impl.Utils.readBytesAndRewind;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.*;
@@ -64,37 +62,6 @@ import static io.netty.handler.codec.mqtt.MqttQoS.*;
*/
public class ProtocolProcessor {
static final class WillMessage {
private final String topic;
private final ByteBuffer payload;
private final boolean retained;
private final MqttQoS qos;
WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) {
this.topic = topic;
this.payload = payload;
this.retained = retained;
this.qos = qos;
}
public String getTopic() {
return topic;
}
public ByteBuffer getPayload() {
return payload;
}
public boolean isRetained() {
return retained;
}
public MqttQoS getQos() {
return qos;
}
}
private enum SubscriptionState {
STORED, VERIFIED
}
@@ -105,16 +72,12 @@ public class ProtocolProcessor {
private ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
private ISubscriptionsDirectory subscriptions;
private boolean allowAnonymous;
private boolean allowZeroByteClientId;
private boolean reauthorizeSubscriptionsOnConnect;
private IAuthorizator m_authorizator;
private IMessagesStore m_messagesStore;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
private BrokerInterceptor m_interceptor;
private Qos0PublishHandler qos0PublishHandler;
@@ -124,8 +87,8 @@ public class ProtocolProcessor {
private InternalRepublisher internalRepublisher;
SessionsRepository sessionsRepository;
// maps clientID to Will testament, if specified on CONNECT
private ConcurrentMap<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
private ConnectHandler connectHandler;
private DisconnectHandler disconnectHandler;
ProtocolProcessor() {
}
@@ -170,25 +133,21 @@ public class ProtocolProcessor {
boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
boolean reauthorizeSubscriptionsOnConnect) {
LOG.debug("Initializing MQTT protocol processor...");
LOG.info("Initializing MQTT protocol processor..");
this.connectionDescriptors = connectionDescriptors;
this.subscriptionInCourse = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect;
this.allowZeroByteClientId = allowZeroByteClientId;
m_authorizator = authorizator;
if (LOG.isDebugEnabled()) {
LOG.debug("Initial subscriptions tree={}", subscriptions.dumpTree());
}
m_authenticator = authenticator;
m_messagesStore = storageService;
m_sessionsStore = sessionsStore;
this.sessionsRepository = sessionsRepository;
LOG.info("Initializing messages publisher...");
LOG.debug("Initializing messages publisher...");
final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors);
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, messageSender,
subscriptions, this.sessionsRepository);
@@ -203,240 +162,16 @@ public class ProtocolProcessor {
LOG.debug("Initializing internal republisher...");
this.internalRepublisher = new InternalRepublisher(messageSender);
connectHandler = new ConnectHandler(connectionDescriptors, m_interceptor, sessionsRepository, m_sessionsStore,
authenticator, m_authorizator, subscriptions,
allowAnonymous, allowZeroByteClientId, reauthorizeSubscriptionsOnConnect, internalRepublisher);
disconnectHandler = new DisconnectHandler(connectionDescriptors, sessionsRepository, subscriptions, m_interceptor);
LOG.info("Initialized");
}
public void processConnect(Channel channel, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientId = payload.clientIdentifier();
final String username = payload.userName();
LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, username);
if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel()
&& msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) {
MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
LOG.error("MQTT protocol version is not valid. CId={}", clientId);
channel.writeAndFlush(badProto).addListener(FIRE_EXCEPTION_ON_FAILURE);
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
final boolean cleanSession = msg.variableHeader().isCleanSession();
if (clientId == null || clientId.length() == 0) {
if (!cleanSession || !this.allowZeroByteClientId) {
MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
channel.writeAndFlush(badId).addListener(FIRE_EXCEPTION_ON_FAILURE);
channel.close().addListener(CLOSE_ON_FAILURE);
LOG.error("The MQTT client ID cannot be empty. Username={}", username);
return;
}
// Generating client id.
clientId = UUID.randomUUID().toString().replace("-", "");
LOG.info("Client has connected with a server generated identifier. CId={}, username={}", clientId,
username);
}
if (!login(channel, msg, clientId)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession);
final ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
if (existing != null) {
LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId);
existing.abort();
//return;
this.connectionDescriptors.removeConnection(existing);
this.connectionDescriptors.addConnection(descriptor);
}
initializeKeepAliveTimeout(channel, msg, clientId);
storeWillMessage(msg, clientId);
if (!cleanSession && reauthorizeSubscriptionsOnConnect) {
reauthorizeOnExistingSubscriptions(clientId, username);
}
if (!sendAck(descriptor, msg, clientId)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
m_interceptor.notifyClientConnected(msg);
if (!descriptor.assignState(SENDACK, SESSION_CREATED)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
final ClientSession clientSession = this.sessionsRepository.createOrLoadClientSession(clientId, cleanSession);
if (!republish(descriptor, msg, clientSession)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */;
setupAutoFlusher(channel, flushIntervalMs);
final boolean success = descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED);
if (!success) {
channel.close().addListener(CLOSE_ON_FAILURE);
}
LOG.info("Connected client <{}> with login <{}>", clientId, username);
}
private void reauthorizeOnExistingSubscriptions(String clientId, String username) {
if (!m_sessionsStore.contains(clientId)) {
return;
}
final Collection<Subscription> clientSubscriptions = m_sessionsStore.subscriptionStore()
.listClientSubscriptions(clientId);
for (Subscription sub : clientSubscriptions) {
final Topic topicToReauthorize = sub.getTopicFilter();
final boolean readAuthorized = m_authorizator.canRead(topicToReauthorize, username, clientId);
if (!readAuthorized) {
subscriptions.removeSubscription(topicToReauthorize, clientId);
}
}
}
private void setupAutoFlusher(Channel channel, int flushIntervalMs) {
try {
channel.pipeline().addAfter(
"idleEventHandler",
"autoFlusher",
new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
} catch (NoSuchElementException nseex) {
// the idleEventHandler is not present on the pipeline
channel.pipeline()
.addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
}
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
return connAck(returnCode, false);
}
private MqttConnAckMessage connAckWithSessionPresent(MqttConnectReturnCode returnCode) {
return connAck(returnCode, true);
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE,
false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);
return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) {
// handle user authentication
if (msg.variableHeader().hasUserName()) {
byte[] pwd = null;
if (msg.variableHeader().hasPassword()) {
pwd = msg.payload().password().getBytes(StandardCharsets.UTF_8);
} else if (!this.allowAnonymous) {
LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId);
failedCredentials(channel);
return false;
}
final String login = msg.payload().userName();
if (!m_authenticator.checkValid(clientId, login, pwd)) {
LOG.error("Authenticator has rejected the MQTT credentials CId={}, username={}", clientId, login);
failedCredentials(channel);
return false;
}
NettyUtils.userName(channel, login);
} else if (!this.allowAnonymous) {
LOG.error("Client didn't supply any credentials and MQTT anonymous mode is disabled. CId={}", clientId);
failedCredentials(channel);
return false;
}
return true;
}
private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, String clientId) {
LOG.debug("Sending CONNACK. CId={}", clientId);
final boolean success = descriptor.assignState(DISCONNECTED, SENDACK);
if (!success) {
return false;
}
MqttConnAckMessage okResp;
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientId);
boolean isSessionAlreadyStored = clientSession != null;
final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession();
if (!msgCleanSessionFlag && isSessionAlreadyStored) {
okResp = connAckWithSessionPresent(CONNECTION_ACCEPTED);
} else {
okResp = connAck(CONNECTION_ACCEPTED);
}
descriptor.writeAndFlush(okResp);
LOG.debug("CONNACK has been sent. CId={}", clientId);
if (isSessionAlreadyStored && msgCleanSessionFlag) {
for (Subscription existingSub : clientSession.getSubscriptions()) {
this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientId);
}
}
return true;
}
private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) {
int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
NettyUtils.keepAlive(channel, keepAlive);
NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
NettyUtils.clientID(channel, clientId);
int idleTime = Math.round(keepAlive * 1.5f);
setIdleTime(channel.pipeline(), idleTime);
LOG.debug("Connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}",
clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
}
private void storeWillMessage(MqttConnectMessage msg, String clientId) {
// Handle will flag
if (msg.variableHeader().isWillFlag()) {
MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos());
LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}",
clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain());
byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8);
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
// save the will testament in the clientID store
WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(),
willQos);
m_willStore.put(clientId, will);
LOG.debug("MQTT last will and testament has been configured. CId={}", clientId);
}
}
private boolean republish(ConnectionDescriptor descriptor, MqttConnectMessage msg, ClientSession clientSession) {
final boolean success = descriptor.assignState(SESSION_CREATED, MESSAGES_REPUBLISHED);
if (!success) {
return false;
}
if (!msg.variableHeader().isCleanSession()) {
// force the republish of stored QoS1 and QoS2
LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
this.internalRepublisher.publishStored(clientSession);
}
return true;
}
private void failedCredentials(Channel session) {
session.writeAndFlush(connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD))
.addListener(FIRE_EXCEPTION_ON_FAILURE);
LOG.info("Client {} failed to connect with bad username or password.", session);
}
private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
if (pipeline.names().contains("idleStateHandler")) {
pipeline.remove("idleStateHandler");
}
pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0));
public void processConnect(Channel channel, MqttConnectMessage connectMessage) {
this.connectHandler.processConnect(channel, connectMessage);
}
public void processPubAck(Channel channel, MqttPubAckMessage msg) {
@@ -476,7 +211,7 @@ public class ProtocolProcessor {
final MqttQoS qos = msg.fixedHeader().qosLevel();
final String clientId = NettyUtils.clientID(channel);
LOG.info("Processing PUBLISH message. CId={}, topic={}, messageId={}, qos={}", clientId,
msg.variableHeader().topicName(), msg.variableHeader().messageId(), qos);
msg.variableHeader().topicName(), msg.variableHeader().messageId(), qos);
switch (qos) {
case AT_MOST_ONCE:
this.qos0PublishHandler.receivedPublishQos0(channel, msg);
@@ -601,100 +336,7 @@ public class ProtocolProcessor {
final String clientID = NettyUtils.clientID(channel);
LOG.debug("Processing DISCONNECT message. CId={}", clientID);
channel.flush();
final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID);
if (existingDescriptor == null) {
// another client with same ID removed the descriptor, we must exit
channel.close().addListener(CLOSE_ON_FAILURE);
return;
}
if (existingDescriptor.doesNotUseChannel(channel)) {
// another client saved it's descriptor, exit
LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!removeSubscriptions(existingDescriptor, clientID)) {
LOG.warn("Unable to remove subscriptions. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!dropStoredMessages(existingDescriptor, clientID)) {
LOG.warn("Unable to drop stored messages. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) {
LOG.warn("Unable to drop will message. Closing connection. CId={}", clientID);
existingDescriptor.abort();
return;
}
if (!existingDescriptor.close()) {
LOG.debug("Connection has been closed. CId={}", clientID);
return;
}
boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor);
if (!stillPresent) {
// another descriptor was inserted
LOG.warn("Another descriptor has been inserted. CId={}", clientID);
return;
}
this.sessionsRepository.disconnect(clientID);
LOG.info("Client <{}> disconnected", clientID);
}
private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) {
final boolean success = descriptor.assignState(ESTABLISHED, SUBSCRIPTIONS_REMOVED);
if (!success) {
return false;
}
if (descriptor.cleanSession) {
LOG.trace("Removing saved subscriptions. CId={}", descriptor.clientID);
final ClientSession session = this.sessionsRepository.sessionForClient(clientID);
session.wipeSubscriptions();
for (Subscription existingSub : session.getSubscriptions()) {
this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientID);
}
LOG.trace("Saved subscriptions have been removed. CId={}", descriptor.clientID);
}
return true;
}
private boolean dropStoredMessages(ConnectionDescriptor descriptor, String clientID) {
final boolean success = descriptor.assignState(SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED);
if (!success) {
return false;
}
final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
if (clientSession.isCleanSession()) {
clientSession.dropQueue();
// LOG.debug("Removing messages of session. CId={}", descriptor.clientID);
// this.m_sessionsStore.dropQueue(clientID);
// LOG.debug("Messages of the session have been removed. CId={}", descriptor.clientID);
}
return true;
}
private boolean cleanWillMessageAndNotifyInterceptor(ConnectionDescriptor descriptor, String clientID) {
final boolean success = descriptor.assignState(MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED);
if (!success) {
return false;
}
LOG.trace("Removing will message. ClientId={}", descriptor.clientID);
// cleanup the will store
m_willStore.remove(clientID);
String username = descriptor.getUsername();
m_interceptor.notifyClientDisconnected(clientID, username);
return true;
disconnectHandler.processDisconnect(channel, clientID);
}
public void processConnectionLost(String clientID, Channel channel) {
@@ -702,10 +344,11 @@ public class ProtocolProcessor {
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
connectionDescriptors.removeConnection(oldConnDescr);
// publish the Will message (if any) for the clientID
if (m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID);
final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
WillMessage will = clientSession.willMessage();
if (will != null) {
forwardPublishWill(will, clientID);
m_willStore.remove(clientID);
clientSession.removeWill();
}
String username = NettyUtils.userName(channel);
broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java
+5 -3
View File
@@ -24,7 +24,6 @@ import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -50,6 +49,9 @@ public class SessionsRepository {
private ISubscriptionsStore subscriptionsStore;
private ScheduledExecutorService scheduler;
private final ConcurrentMap<String, ClientSession> sessionsCache = new ConcurrentHashMap<>();
// maps clientID to Will testament, if specified on CONNECT
// maps clientID to Will testament, if specified on CONNECT
private ConcurrentMap<String, WillMessage> willMessages = new ConcurrentHashMap<>();
public SessionsRepository(ISessionsStore sessionsStore, ScheduledExecutorService scheduler) {
this.sessions = sessionsStore;
@@ -80,9 +82,9 @@ public class SessionsRepository {
private ClientSession newClientSessionAndCacheIt(String clientID, boolean cleanSession) {
ClientSession session;
if (cleanSession) {
session = new TransientSession(clientID);
session = new TransientSession(clientID, this.willMessages);
} else {
DurableSession durableSession = new DurableSession(clientID, this.sessions, this.subscriptionsStore);
DurableSession durableSession = new DurableSession(clientID, this.sessions, this.subscriptionsStore, willMessages);
durableSession.reloadAllSubscriptionsFromStore();
session = durableSession;
}
broker/src/main/java/io/moquette/spi/impl/WillMessage.java
+36
View File
@@ -0,0 +1,36 @@
package io.moquette.spi.impl;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.nio.ByteBuffer;
public final class WillMessage {
private final String topic;
private final ByteBuffer payload;
private final boolean retained;
private final MqttQoS qos;
public WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) {
this.topic = topic;
this.payload = payload;
this.retained = retained;
this.qos = qos;
}
public String getTopic() {
return topic;
}
public ByteBuffer getPayload() {
return payload;
}
public boolean isRetained() {
return retained;
}
public MqttQoS getQos() {
return qos;
}
}
broker/src/test/java/io/moquette/spi/impl/ConnectHandlerTest.java
+315
View File
@@ -0,0 +1,315 @@
package io.moquette.spi.impl;
import io.moquette.persistence.MemoryStorageService;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttVersion;
import org.junit.Before;
import org.junit.Test;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck;
import static io.moquette.spi.impl.ProtocolProcessorTest.*;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonMap;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ConnectHandlerTest {
EmbeddedChannel session;
MqttMessageBuilders.ConnectBuilder connMsg;
IMessagesStore m_messagesStore;
ISessionsStore m_sessionStore;
ISubscriptionsDirectory subscriptions;
MockAuthenticator m_mockAuthenticator;
private ConnectHandler sut;
private DisconnectHandler disconnectHandler;
private SessionsRepository sessionsRepository;
@Before
public void setUp() {
connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
session = new EmbeddedChannel();
MemoryStorageService memStorage = new MemoryStorageService(null, null);
m_messagesStore = memStorage.messagesStore();
m_sessionStore = memStorage.sessionsStore();
m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
subscriptions = new CTrieSubscriptionDirectory();
sessionsRepository = new SessionsRepository(m_sessionStore, null);
subscriptions.init(sessionsRepository);
final ConnectionDescriptorStore connectionsRegistry = new ConnectionDescriptorStore();
sut = new ConnectHandler(connectionsRegistry, NO_OBSERVERS_INTERCEPTOR, sessionsRepository, m_sessionStore, m_mockAuthenticator,
new PermitAllAuthorizator(), subscriptions, true, true, false, null);
disconnectHandler = new DisconnectHandler(connectionsRegistry, sessionsRepository,
subscriptions, NO_OBSERVERS_INTERCEPTOR);
}
@Test
public void testZeroByteClientIdWithCleanSession() {
// Allow zero byte client ids
sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator,
new PermitAllAuthorizator(), subscriptions, true, true, false, null);
// Connect message with clean session set to true and client id is null.
MqttConnectMessage msg = connMsg.clientId(null)
.protocolVersion(MqttVersion.MQTT_3_1_1)
.cleanSession(true)
.build();
sut.processConnect(session, msg);
assertEqualsConnAck("Connection MUST be accepted, unique clientid MUST be generated and clean session to true",
CONNECTION_ACCEPTED, session.readOutbound());
assertTrue("Connection should be valid and open,", session.isOpen());
}
@Test
public void invalidAuthentication() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER + "_fake")
.password(TEST_PWD)
.build();
// Exercise
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound());
assertFalse("Connection should be closed by the broker.", session.isOpen());
}
@Test
public void testConnect_badClientID() {
connMsg.clientId("extremely_long_clientID_greater_than_23").build();
// Exercise
sut.processConnect(session, connMsg.clientId("extremely_long_clientID_greater_than_23").build());
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
}
@Test
public void testWill() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true)
.willTopic("topic").willMessage("Topic message").build();
// Exercise
// m_handler.setMessaging(mockedMessaging);
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", session.isOpen());
// TODO verify the call
/*
* verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
* any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));
*/
}
@Test
public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() {
// Connect a client1
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER)
.password(TEST_PWD)
.build();
sut.processConnect(session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
// create another connect same clientID but with bad credentials
MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect()
.protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.EVIL_TEST_USER)
.password(ProtocolProcessorTest.EVIL_TEST_PWD)
.build();
EmbeddedChannel evilSession = new EmbeddedChannel();
// Exercise
sut.processConnect(evilSession, evilClientConnMsg);
// Verify
// the evil client gets a not auth notification
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, evilSession.readOutbound());
// the good client remains connected
assertTrue(session.isOpen());
assertFalse(evilSession.isOpen());
}
@Test
public void acceptAnonymousClient() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
// Exercise
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", session.isOpen());
}
@Test
public void validAuthentication() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER).password(TEST_PWD).build();
// Exercise
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", session.isOpen());
}
@Test
public void noPasswdAuthentication() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER)
.build();
// Exercise
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound());
assertFalse("Connection should be closed by the broker.", session.isOpen());
}
@Test
public void prohibitAnonymousClient() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
reinitProcessorProhibitingAnonymousClients();
// Exercise
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound());
assertFalse("Connection should be closed by the broker.", session.isOpen());
}
protected void reinitProcessorProhibitingAnonymousClients() {
sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator,
new PermitAllAuthorizator(), subscriptions, false, true, false, null);
}
@Test
public void prohibitAnonymousClient_providingUsername() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER + "_fake")
.build();
reinitProcessorProhibitingAnonymousClients();
// Exercise
sut.processConnect(session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound());
assertFalse("Connection should be closed by the broker.", session.isOpen());
}
@Test
public void testZeroByteClientIdNotAllowed() {
sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator,
new PermitAllAuthorizator(), subscriptions, false, false, false, null);
// Connect message with clean session set to true and client id is null.
MqttConnectMessage msg = connMsg.clientId(null)
.protocolVersion(MqttVersion.MQTT_3_1_1)
.cleanSession(true)
.build();
sut.processConnect(session, msg);
assertEqualsConnAck("Zero byte client identifiers are not allowed",
CONNECTION_REFUSED_IDENTIFIER_REJECTED, session.readOutbound());
assertFalse("Connection should closed.", session.isOpen());
}
@Test
public void testZeroByteClientIdWithoutCleanSession() {
// Allow zero byte client ids
reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients();
// Connect message without clean session set to true but client id is still null
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1)
.build();
sut.processConnect(session, msg);
assertEqualsConnAck(
"Identifier should be rejected due to having clean session set to false.",
CONNECTION_REFUSED_IDENTIFIER_REJECTED,
session.readOutbound());
assertFalse("Connection should be closed by the broker.", session.isOpen());
}
protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() {
sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator,
new PermitAllAuthorizator(), subscriptions, true, true, false, null);
}
@Test
public void testConnAckContainsSessionPresentFlag() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1_1).build();
NettyUtils.clientID(session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(session, false);
// Connect a first time
sut.processConnect(session, msg);
// disconnect
disconnectHandler.processDisconnect(session, FAKE_CLIENT_ID);
// Exercise, reconnect
EmbeddedChannel firstReceiverSession = new EmbeddedChannel();
sut.processConnect(firstReceiverSession, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen());
}
@Test
public void connectWithCleanSessionUpdateClientSession() {
// first connect with clean session true
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
sut.processConnect(session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
disconnectHandler.processDisconnect(session, FAKE_CLIENT_ID);
assertFalse(session.isOpen());
// second connect with clean session false
session = new EmbeddedChannel();
MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1).build();
sut.processConnect(session, secondConnMsg);
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
// Verify client session is clean false
ClientSession session = sessionsRepository.sessionForClient(FAKE_CLIENT_ID);
assertFalse(session.isCleanSession());
// Verify
// assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+32 -294
View File
@@ -18,7 +18,6 @@ package io.moquette.spi.impl;
import io.moquette.persistence.MemoryStorageService;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.PermitAllAuthorizator;
@@ -34,7 +33,7 @@ import org.junit.Test;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck;
import static io.moquette.spi.impl.ProtocolProcessorTest.*;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonMap;
import static org.junit.Assert.assertFalse;
@@ -42,322 +41,61 @@ import static org.junit.Assert.assertTrue;
public class ProtocolProcessor_CONNECT_Test {
EmbeddedChannel m_session;
MqttMessageBuilders.ConnectBuilder connMsg;
ProtocolProcessor m_processor;
IMessagesStore m_messagesStore;
ISessionsStore m_sessionStore;
ISubscriptionsDirectory subscriptions;
MockAuthenticator m_mockAuthenticator;
private EmbeddedChannel session;
private ProtocolProcessor processor;
private ISessionsStore sessionStore;
@Before
public void setUp() throws InterruptedException {
connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
m_session = new EmbeddedChannel();
public void setUp() {
session = new EmbeddedChannel();
MemoryStorageService memStorage = new MemoryStorageService(null, null);
m_messagesStore = memStorage.messagesStore();
m_sessionStore = memStorage.sessionsStore();
// m_messagesStore.initStore();
IMessagesStore m_messagesStore = memStorage.messagesStore();
sessionStore = memStorage.sessionsStore();
m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
MockAuthenticator mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
subscriptions = new CTrieSubscriptionDirectory();
SessionsRepository sessionsRepository = new SessionsRepository(m_sessionStore, null);
ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null);
subscriptions.init(sessionsRepository);
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
processor = new ProtocolProcessor();
processor.init(subscriptions, m_messagesStore, sessionStore, mockAuthenticator, true,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR,
new SessionsRepository(this.m_sessionStore, null), false);
new SessionsRepository(this.sessionStore, null), false);
}
@Test
public void testConnect_badClientID() {
connMsg.clientId("extremely_long_clientID_greater_than_23").build();
// Exercise
m_processor.processConnect(m_session, connMsg.clientId("extremely_long_clientID_greater_than_23").build());
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
}
@Test
public void testWill() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true)
.willTopic("topic").willMessage("Topic message").build();
// Exercise
// m_handler.setMessaging(mockedMessaging);
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
// TODO verify the call
/*
* verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
* any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));
*/
}
@Test
public void validAuthentication() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER).password(TEST_PWD).build();
// Exercise
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
}
@Test
public void noPasswdAuthentication() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER)
.build();
// Exercise
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
public void invalidAuthentication() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER + "_fake")
.password(TEST_PWD)
.build();
// Exercise
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
public void prohibitAnonymousClient() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
reinitProcessorProhibitingAnonymousClients();
// Exercise
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
protected void reinitProcessorProhibitingAnonymousClients() {
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), false);
}
@Test
public void prohibitAnonymousClient_providingUsername() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER + "_fake")
.build();
reinitProcessorProhibitingAnonymousClients();
// Exercise
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
public void acceptAnonymousClient() {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
// Exercise
m_processor.processConnect(m_session, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
}
@Test
public void connectWithCleanSessionUpdateClientSession() throws InterruptedException {
// first connect with clean session true
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
m_processor.processDisconnect(m_session);
assertFalse(m_session.isOpen());
// second connect with clean session false
m_session = new EmbeddedChannel();
MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1).build();
m_processor.processConnect(m_session, secondConnMsg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
// Verify client session is clean false
ClientSession session = m_processor.sessionsRepository.sessionForClient(FAKE_CLIENT_ID);
assertFalse(session.isCleanSession());
// Verify
// assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
}
@Test
public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() {
// Connect a client1
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER)
.password(TEST_PWD)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
// create another connect same clientID but with bad credentials
MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect()
.protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.EVIL_TEST_USER)
.password(ProtocolProcessorTest.EVIL_TEST_PWD)
.build();
EmbeddedChannel evilSession = new EmbeddedChannel();
// Exercise
m_processor.processConnect(evilSession, evilClientConnMsg);
// Verify
// the evil client gets a not auth notification
assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, evilSession.readOutbound());
// the good client remains connected
assertTrue(m_session.isOpen());
assertFalse(evilSession.isOpen());
}
@Test
public void testConnAckContainsSessionPresentFlag() throws InterruptedException {
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1_1).build();
NettyUtils.clientID(m_session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(m_session, false);
// Connect a first time
m_processor.processConnect(m_session, msg);
// disconnect
m_processor.processDisconnect(m_session);
// Exercise, reconnect
EmbeddedChannel firstReceiverSession = new EmbeddedChannel();
m_processor.processConnect(firstReceiverSession, msg);
// Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen());
}
@Test
public void testMultipleReconnection() throws InterruptedException {
public void testMultipleReconnection() {
// connect with clean a false and subscribe to a topic
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1_1).build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
processor.processConnect(session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", session.isOpen());
// subscribe
MqttSubscribeMessage subscribeMsg = MqttMessageBuilders.subscribe()
.addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.NEWS_TOPIC).messageId(10).build();
NettyUtils.clientID(m_session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(m_session, false);
m_processor.processSubscribe(m_session, subscribeMsg);
Subscription expectedSubscription = new Subscription(
FAKE_CLIENT_ID,
new Topic(ProtocolProcessorTest.NEWS_TOPIC),
MqttQoS.AT_MOST_ONCE);
verifySubscriptionExists(m_sessionStore, expectedSubscription);
assertEqualsSubAck(m_session.readOutbound());
NettyUtils.clientID(session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(session, false);
processor.processSubscribe(session, subscribeMsg);
final Topic newTopic = new Topic(ProtocolProcessorTest.NEWS_TOPIC);
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, newTopic, MqttQoS.AT_MOST_ONCE);
verifySubscriptionExists(sessionStore, expectedSubscription);
assertEqualsSubAck(session.readOutbound());
// disconnect
m_processor.processDisconnect(m_session);
assertFalse(m_session.isOpen());
processor.processDisconnect(session);
assertFalse(session.isOpen());
// reconnect clean session a false
m_session = new EmbeddedChannel();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
session = new EmbeddedChannel();
processor.processConnect(session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", session.isOpen());
// verify that the first subscription is still preserved
verifySubscriptionExists(m_sessionStore, expectedSubscription);
}
@Test
public void testZeroByteClientIdWithoutCleanSession() {
// Allow zero byte client ids
m_processor = new ProtocolProcessor();
reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients();
// Connect message without clean session set to true but client id is still null
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(
"Identifier should be rejected due to having clean session set to false.",
CONNECTION_REFUSED_IDENTIFIER_REJECTED,
m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() {
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), false);
}
@Test
public void testZeroByteClientIdWithCleanSession() {
// Allow zero byte client ids
m_processor = new ProtocolProcessor();
reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients();
// Connect message with clean session set to true and client id is null.
MqttConnectMessage msg = connMsg.clientId(null)
.protocolVersion(MqttVersion.MQTT_3_1_1)
.cleanSession(true)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck("Connection MUST be accepted, unique clientid MUST be generated and clean session to true",
CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection should be valid and open,", m_session.isOpen());
}
@Test
public void testZeroByteClientIdNotAllowed() {
// Connect message with clean session set to true and client id is null.
MqttConnectMessage msg = connMsg.clientId(null)
.protocolVersion(MqttVersion.MQTT_3_1_1)
.cleanSession(true)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck("Zero byte client identifiers are not allowed",
CONNECTION_REFUSED_IDENTIFIER_REJECTED, m_session.readOutbound());
assertFalse("Connection should closed.", m_session.isOpen());
verifySubscriptionExists(sessionStore, expectedSubscription);
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java
+14 -15
View File
@@ -32,7 +32,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testCleanSession_maintainClientSubscriptions() throws InterruptedException {
public void testCleanSession_maintainClientSubscriptions() {
connectAsClient(this.m_channel, "TestClient");
subscribe("/topic", AT_MOST_ONCE);
disconnect();
@@ -48,7 +48,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
@Ignore("At the end it's equal to testCleanSession_maintainClientSubscriptions")
@Test
public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws InterruptedException {
public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() {
connectAsClient("TestClient");
subscribe("/topic", AT_MOST_ONCE);
disconnect();
@@ -68,7 +68,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
* previous subscription
*/
@Test
public void testCleanSession_correctlyClientSubscriptions() throws InterruptedException {
public void testCleanSession_correctlyClientSubscriptions() {
connectAsClient("TestClient");
subscribe("/topic", AT_MOST_ONCE);
disconnect();
@@ -109,7 +109,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException {
public void testUnsubscribe_do_not_notify_anymore_new_session() {
connect();
subscribe("/topic", AT_MOST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false);
@@ -125,7 +125,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testPublishWithQoS1() throws InterruptedException {
public void testPublishWithQoS1() {
connect();
subscribe("/topic", AT_LEAST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, false);
@@ -148,21 +148,21 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
verifyPublishIsReceived();
}
protected void publishFromAnotherClient(String topic, MqttQoS qos) {
private void publishFromAnotherClient(String topic, MqttQoS qos) {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishToAs(anotherChannel, "AnotherClient", topic, qos, 67, false);
disconnect(anotherChannel);
}
protected void publishFromAnotherClient(String topic, String payload, MqttQoS qos) {
private void publishFromAnotherClient(String topic, String payload, MqttQoS qos) {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishToAs(anotherChannel, "AnotherClient", topic, payload, qos, 68, false);
disconnect(anotherChannel);
}
protected void publishQoS2FromAnotherClient(String topic) {
private void publishQoS2FromAnotherClient(String topic) {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishQoS2ToAs(anotherChannel, "AnotherClient", topic, 67, false);
@@ -187,7 +187,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testPublishWithQoS2() throws InterruptedException {
public void testPublishWithQoS2() {
connect();
subscribe("/topic", EXACTLY_ONCE);
disconnect();
@@ -201,8 +201,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1()
throws InterruptedException {
public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() {
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();
@@ -223,7 +222,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws InterruptedException {
public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() {
connect();
subscribe("/topic", AT_MOST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, true);
@@ -240,7 +239,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws InterruptedException {
public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() {
EmbeddedChannel publisher = new EmbeddedChannel();
connectAsClient(publisher, "Publisher");
@@ -261,7 +260,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void checkReplayStoredPublish_forNoCleanSession_qos1() throws InterruptedException {
public void checkReplayStoredPublish_forNoCleanSession_qos1() {
EmbeddedChannel publisher = new EmbeddedChannel();
connectAsClient(publisher, "Publisher");
@@ -283,7 +282,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
* "hello2"
*/
@Test
public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws InterruptedException {
public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() {
connect();
subscribe("topic", EXACTLY_ONCE);
disconnect();