Removed all the old ProtocolProcessor stuff
This commit is contained in:
andsel
@@ -1,285 +0,0 @@
|
||||
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.IAuthorizatorPolicy;
|
||||
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 IAuthorizatorPolicy 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, IAuthorizatorPolicy 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));
|
||||
}
|
||||
}
|
||||
@@ -1,133 +0,0 @@
|
||||
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;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -1,104 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.spi.ClientSession;
|
||||
import io.moquette.spi.EnqueuedMessage;
|
||||
import io.moquette.spi.IMessagesStore;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.handler.codec.mqtt.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Collection;
|
||||
|
||||
class InternalRepublisher {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class);
|
||||
|
||||
private final PersistentQueueMessageSender messageSender;
|
||||
|
||||
InternalRepublisher(PersistentQueueMessageSender messageSender) {
|
||||
this.messageSender = messageSender;
|
||||
}
|
||||
|
||||
void publishRetained(ClientSession targetSession, Collection<IMessagesStore.StoredMessage> messages) {
|
||||
for (IMessagesStore.StoredMessage storedMsg : messages) {
|
||||
// fire as retained the message
|
||||
MqttPublishMessage publishMsg = retainedPublish(storedMsg);
|
||||
if (storedMsg.getQos() != MqttQoS.AT_MOST_ONCE) {
|
||||
LOG.debug("Adding message to inflight zone. ClientId={}, topic={}", targetSession.clientID,
|
||||
storedMsg.getTopic());
|
||||
int packetID = targetSession.inFlightAckWaiting(storedMsg);
|
||||
|
||||
// set the PacketIdentifier only for QoS > 0
|
||||
publishMsg = retainedPublish(storedMsg, packetID);
|
||||
}
|
||||
|
||||
this.messageSender.sendPublish(targetSession, publishMsg);
|
||||
}
|
||||
}
|
||||
|
||||
void publishStored(ClientSession clientSession) {
|
||||
if (clientSession.isEmptyQueue()) {
|
||||
LOG.info("There are no stored publish events to CId={}", clientSession.clientID);
|
||||
return;
|
||||
}
|
||||
|
||||
LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
|
||||
EnqueuedMessage pubEvt;
|
||||
while ((pubEvt = clientSession.poll()) != null) {
|
||||
// put in flight zone
|
||||
LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, topic={}", clientSession.clientID,
|
||||
pubEvt.msg.getGuid(), pubEvt.msg.getTopic());
|
||||
|
||||
MqttPublishMessage publishMsg = notRetainedPublish(pubEvt.msg);
|
||||
// set the PacketIdentifier only for QoS > 0
|
||||
if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) {
|
||||
publishMsg = notRetainedPublish(pubEvt.msg, pubEvt.messageId);
|
||||
}
|
||||
this.messageSender.sendPublish(clientSession, publishMsg);
|
||||
}
|
||||
}
|
||||
|
||||
private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) {
|
||||
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), false,
|
||||
messageID);
|
||||
}
|
||||
|
||||
private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) {
|
||||
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), false,
|
||||
0);
|
||||
}
|
||||
|
||||
private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) {
|
||||
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), true,
|
||||
0);
|
||||
}
|
||||
|
||||
private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) {
|
||||
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), true,
|
||||
packetID);
|
||||
}
|
||||
|
||||
public static MqttPublishMessage createPublishForQos(String topic, MqttQoS qos, ByteBuf message, boolean retained,
|
||||
int messageId) {
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0);
|
||||
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
|
||||
return new MqttPublishMessage(fixedHeader, varHeader, message);
|
||||
}
|
||||
}
|
||||
@@ -1,115 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.connections.IConnectionsManager;
|
||||
import io.moquette.spi.ClientSession;
|
||||
import io.moquette.spi.IMessagesStore;
|
||||
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
|
||||
import io.moquette.spi.impl.subscriptions.Subscription;
|
||||
import io.moquette.spi.impl.subscriptions.Topic;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.handler.codec.mqtt.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
|
||||
|
||||
class MessagesPublisher {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class);
|
||||
private final IConnectionsManager connectionDescriptors;
|
||||
private final PersistentQueueMessageSender messageSender;
|
||||
private final ISubscriptionsDirectory subscriptions;
|
||||
private SessionsRepository sessionsRepository;
|
||||
|
||||
MessagesPublisher(IConnectionsManager connectionDescriptors,
|
||||
PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions,
|
||||
SessionsRepository sessionsRepository) {
|
||||
this.connectionDescriptors = connectionDescriptors;
|
||||
this.messageSender = messageSender;
|
||||
this.subscriptions = subscriptions;
|
||||
this.sessionsRepository = sessionsRepository;
|
||||
}
|
||||
|
||||
private static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
|
||||
return notRetainedPublishWithMessageId(topic, qos, message, 0);
|
||||
}
|
||||
|
||||
private static MqttPublishMessage notRetainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message,
|
||||
int messageId) {
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
|
||||
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
|
||||
return new MqttPublishMessage(fixedHeader, varHeader, message);
|
||||
}
|
||||
|
||||
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic, int messageID) {
|
||||
if (LOG.isTraceEnabled()) {
|
||||
LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
|
||||
"subscriptionTree={}", pubMsg.getClientID(), topic, messageID,
|
||||
DebugUtils.payload2Str(pubMsg.getPayload()),
|
||||
subscriptions.dumpTree());
|
||||
} else {
|
||||
LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}",
|
||||
pubMsg.getClientID(), topic, messageID);
|
||||
}
|
||||
publish2Subscribers(pubMsg, topic);
|
||||
}
|
||||
|
||||
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic) {
|
||||
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
|
||||
final String topic1 = pubMsg.getTopic();
|
||||
final MqttQoS publishingQos = pubMsg.getQos();
|
||||
final ByteBuf origPayload = pubMsg.getPayload();
|
||||
|
||||
for (final Subscription sub : topicMatchingSubscriptions) {
|
||||
MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
|
||||
ClientSession targetSession = this.sessionsRepository.sessionForClient(sub.getClientId());
|
||||
|
||||
boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());
|
||||
// TODO move all this logic into messageSender, which puts into the flightZone only the messages
|
||||
// that pull out of the queue.
|
||||
if (targetIsActive) {
|
||||
LOG.debug("Sending PUBLISH message to active subscriber. CId={}, topicFilter={}, qos={}",
|
||||
sub.getClientId(), sub.getTopicFilter(), qos);
|
||||
// we need to retain because duplicate only copy r/w indexes and don't retain() causing
|
||||
// refCnt = 0
|
||||
ByteBuf payload = origPayload.retainedDuplicate();
|
||||
MqttPublishMessage publishMsg;
|
||||
if (qos != MqttQoS.AT_MOST_ONCE) {
|
||||
// QoS 1 or 2
|
||||
int messageId = targetSession.inFlightAckWaiting(pubMsg);
|
||||
// set the PacketIdentifier only for QoS > 0
|
||||
publishMsg = notRetainedPublishWithMessageId(topic1, qos, payload, messageId);
|
||||
} else {
|
||||
publishMsg = notRetainedPublish(topic1, qos, payload);
|
||||
}
|
||||
this.messageSender.sendPublish(targetSession, publishMsg);
|
||||
} else {
|
||||
if (!targetSession.isCleanSession()) {
|
||||
LOG.debug("Storing pending PUBLISH inactive message. CId={}, topicFilter={}, qos={}",
|
||||
sub.getClientId(), sub.getTopicFilter(), qos);
|
||||
// store the message in targetSession queue to deliver
|
||||
targetSession.enqueue(pubMsg);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.connections.IConnectionsManager;
|
||||
import io.moquette.server.ConnectionDescriptor;
|
||||
import io.moquette.server.ConnectionDescriptorStore;
|
||||
import io.moquette.spi.ClientSession;
|
||||
import io.netty.handler.codec.mqtt.MqttMessageType;
|
||||
import io.netty.handler.codec.mqtt.MqttPublishMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttQoS;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
|
||||
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
|
||||
|
||||
class PersistentQueueMessageSender {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
|
||||
private final IConnectionsManager connectionDescriptorStore;
|
||||
|
||||
PersistentQueueMessageSender(IConnectionsManager connectionDescriptorStore) {
|
||||
this.connectionDescriptorStore = connectionDescriptorStore;
|
||||
}
|
||||
|
||||
void sendPublish(ClientSession clientsession, MqttPublishMessage pubMessage) {
|
||||
String clientId = clientsession.clientID;
|
||||
final int messageId = pubMessage.variableHeader().messageId();
|
||||
final String topicName = pubMessage.variableHeader().topicName();
|
||||
MqttQoS qos = pubMessage.fixedHeader().qosLevel();
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId,
|
||||
clientId, topicName, qos, DebugUtils.payload2Str(pubMessage.payload()));
|
||||
} else {
|
||||
LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName);
|
||||
}
|
||||
|
||||
boolean messageDelivered = false;
|
||||
|
||||
final Optional<ConnectionDescriptor> optDescriptor = connectionDescriptorStore.lookupDescriptor(clientId);
|
||||
if (optDescriptor.isPresent()) {
|
||||
final ConnectionDescriptor descriptor = optDescriptor.get();
|
||||
try {
|
||||
descriptor.writeAndFlush(pubMessage);
|
||||
messageDelivered = true;
|
||||
} catch (Throwable e) {
|
||||
LOG.error("Unable to send {} message. CId=<{}>, messageId={}", pubMessage.fixedHeader().messageType(),
|
||||
clientId, messageId, e);
|
||||
}
|
||||
}
|
||||
|
||||
if (!messageDelivered) {
|
||||
if (qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {
|
||||
LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, " +
|
||||
"qos={}, removeTemporaryQoS2={}", messageId, clientId, topicName, qos, false);
|
||||
clientsession.enqueue(asStoredMessage(pubMessage));
|
||||
} else {
|
||||
LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, " +
|
||||
"topic={}, qos={}, removeTemporaryQoS2={}", messageId, clientId, topicName, qos, true);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,583 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.connections.IConnectionsManager;
|
||||
import io.moquette.interception.InterceptHandler;
|
||||
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
|
||||
import io.moquette.server.ConnectionDescriptor;
|
||||
import io.moquette.server.ConnectionDescriptorStore;
|
||||
import io.moquette.server.netty.NettyUtils;
|
||||
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;
|
||||
import io.moquette.spi.security.IAuthenticator;
|
||||
import io.moquette.spi.security.IAuthorizatorPolicy;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.handler.codec.mqtt.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
|
||||
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.MqttMessageIdVariableHeader.from;
|
||||
import static io.netty.handler.codec.mqtt.MqttQoS.*;
|
||||
|
||||
/**
|
||||
* Class responsible to handle the logic of MQTT protocol it's the director of the protocol
|
||||
* execution.
|
||||
*
|
||||
* Used by the front facing class ProtocolProcessorBootstrapper.
|
||||
*/
|
||||
public class ProtocolProcessor {
|
||||
|
||||
private enum SubscriptionState {
|
||||
STORED, VERIFIED
|
||||
}
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
|
||||
|
||||
private IConnectionsManager connectionDescriptors;
|
||||
private ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
|
||||
|
||||
private ISubscriptionsDirectory subscriptions;
|
||||
private IAuthorizatorPolicy m_authorizator;
|
||||
|
||||
private IMessagesStore m_messagesStore;
|
||||
|
||||
private ISessionsStore m_sessionsStore;
|
||||
|
||||
private BrokerInterceptor m_interceptor;
|
||||
|
||||
private Qos0PublishHandler qos0PublishHandler;
|
||||
private Qos1PublishHandler qos1PublishHandler;
|
||||
private Qos2PublishHandler qos2PublishHandler;
|
||||
private MessagesPublisher messagesPublisher;
|
||||
private InternalRepublisher internalRepublisher;
|
||||
SessionsRepository sessionsRepository;
|
||||
|
||||
private ConnectHandler connectHandler;
|
||||
private DisconnectHandler disconnectHandler;
|
||||
|
||||
ProtocolProcessor() {
|
||||
}
|
||||
|
||||
public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
|
||||
IAuthenticator authenticator, boolean allowAnonymous, IAuthorizatorPolicy authorizator,
|
||||
BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
|
||||
boolean reauthorizeSubscriptionsOnConnect) {
|
||||
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false,
|
||||
authorizator, interceptor, sessionsRepository, reauthorizeSubscriptionsOnConnect);
|
||||
}
|
||||
|
||||
public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
|
||||
IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
|
||||
IAuthorizatorPolicy authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
|
||||
boolean reauthorizeSubscriptionsOnConnect) {
|
||||
init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore,
|
||||
authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository,
|
||||
reauthorizeSubscriptionsOnConnect);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param subscriptions
|
||||
* the subscription store where are stored all the existing clients subscriptions.
|
||||
* @param storageService
|
||||
* the persistent store to use for save/load of messages for QoS1 and QoS2 handling.
|
||||
* @param sessionsStore
|
||||
* the clients sessions store, used to persist subscriptions.
|
||||
* @param authenticator
|
||||
* the authenticator used in connect messages.
|
||||
* @param allowAnonymous
|
||||
* true connection to clients without credentials.
|
||||
* @param allowZeroByteClientId
|
||||
* true to allow clients connect without a clientid
|
||||
* @param authorizator
|
||||
* used to apply ACL policies to publishes and subscriptions.
|
||||
* @param interceptor
|
||||
* to notify events to an intercept handler
|
||||
*/
|
||||
void init(IConnectionsManager connectionDescriptors, ISubscriptionsDirectory subscriptions,
|
||||
IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
|
||||
boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizatorPolicy authorizator,
|
||||
BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
|
||||
boolean reauthorizeSubscriptionsOnConnect) {
|
||||
LOG.info("Initializing MQTT protocol processor..");
|
||||
this.connectionDescriptors = connectionDescriptors;
|
||||
this.subscriptionInCourse = new ConcurrentHashMap<>();
|
||||
this.m_interceptor = interceptor;
|
||||
this.subscriptions = subscriptions;
|
||||
m_authorizator = authorizator;
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("Initial subscriptions tree={}", subscriptions.dumpTree());
|
||||
}
|
||||
m_messagesStore = storageService;
|
||||
m_sessionsStore = sessionsStore;
|
||||
|
||||
this.sessionsRepository = sessionsRepository;
|
||||
|
||||
LOG.debug("Initializing messages publisher...");
|
||||
final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors);
|
||||
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, messageSender,
|
||||
subscriptions, this.sessionsRepository);
|
||||
|
||||
LOG.debug("Initializing QoS publish handlers...");
|
||||
this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
|
||||
this.messagesPublisher);
|
||||
this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
|
||||
this.connectionDescriptors, this.messagesPublisher);
|
||||
this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor,
|
||||
this.connectionDescriptors, this.messagesPublisher, this.sessionsRepository);
|
||||
|
||||
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 connectMessage) {
|
||||
this.connectHandler.processConnect(channel, connectMessage);
|
||||
}
|
||||
|
||||
public void processPubAck(Channel channel, MqttPubAckMessage msg) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
int messageID = msg.variableHeader().messageId();
|
||||
String username = NettyUtils.userName(channel);
|
||||
LOG.trace("retrieving inflight for messageID <{}>", messageID);
|
||||
|
||||
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
StoredMessage inflightMsg = targetSession.inFlightAcknowledged(messageID);
|
||||
|
||||
String topic = inflightMsg.getTopic();
|
||||
InterceptAcknowledgedMessage wrapped = new InterceptAcknowledgedMessage(inflightMsg, topic, username,
|
||||
messageID);
|
||||
m_interceptor.notifyMessageAcknowledged(wrapped);
|
||||
}
|
||||
|
||||
public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) {
|
||||
// TODO ugly, too much array copy
|
||||
ByteBuf payload = msg.payload();
|
||||
byte[] payloadContent = readBytesAndRewind(payload);
|
||||
|
||||
IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(payloadContent,
|
||||
msg.fixedHeader().qosLevel(), msg.variableHeader().topicName());
|
||||
stored.setRetained(msg.fixedHeader().isRetain());
|
||||
return stored;
|
||||
}
|
||||
|
||||
private static IMessagesStore.StoredMessage asStoredMessage(WillMessage will) {
|
||||
IMessagesStore.StoredMessage pub = new IMessagesStore.StoredMessage(will.getPayload().array(), will.getQos(),
|
||||
will.getTopic());
|
||||
pub.setRetained(will.isRetained());
|
||||
return pub;
|
||||
}
|
||||
|
||||
public void processPublish(Channel channel, MqttPublishMessage msg) {
|
||||
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);
|
||||
switch (qos) {
|
||||
case AT_MOST_ONCE:
|
||||
this.qos0PublishHandler.receivedPublishQos0(channel, msg);
|
||||
break;
|
||||
case AT_LEAST_ONCE:
|
||||
this.qos1PublishHandler.receivedPublishQos1(channel, msg);
|
||||
break;
|
||||
case EXACTLY_ONCE:
|
||||
this.qos2PublishHandler.receivedPublishQos2(channel, msg);
|
||||
break;
|
||||
default:
|
||||
LOG.error("Unknown QoS-Type:{}", qos);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Intended usage is only for embedded versions of the broker, where the hosting application
|
||||
* want to use the broker to send a publish message. Inspired by {@link #processPublish} but
|
||||
* with some changes to avoid security check, and the handshake phases for Qos1 and Qos2. It
|
||||
* also doesn't notifyTopicPublished because using internally the owner should already know
|
||||
* where it's publishing.
|
||||
*
|
||||
* @param msg
|
||||
* the message to publish.
|
||||
* @param clientId
|
||||
* the clientID
|
||||
*/
|
||||
public void internalPublish(MqttPublishMessage msg, final String clientId) {
|
||||
final MqttQoS qos = msg.fixedHeader().qosLevel();
|
||||
final Topic topic = new Topic(msg.variableHeader().topicName());
|
||||
LOG.info("Sending PUBLISH message. Topic={}, qos={}", topic, qos);
|
||||
|
||||
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
|
||||
if (clientId == null || clientId.isEmpty()) {
|
||||
toStoreMsg.setClientID("BROKER_SELF");
|
||||
} else {
|
||||
toStoreMsg.setClientID(clientId);
|
||||
}
|
||||
this.messagesPublisher.publish2Subscribers(toStoreMsg, topic);
|
||||
|
||||
if (!msg.fixedHeader().isRetain()) {
|
||||
return;
|
||||
}
|
||||
if (qos == AT_MOST_ONCE || msg.payload().readableBytes() == 0) {
|
||||
// QoS == 0 && retain => clean old retained
|
||||
m_messagesStore.cleanRetained(topic);
|
||||
return;
|
||||
}
|
||||
m_messagesStore.storeRetained(topic, toStoreMsg);
|
||||
}
|
||||
|
||||
/**
|
||||
* Specialized version to publish will testament message.
|
||||
*/
|
||||
private void forwardPublishWill(WillMessage will, String clientID) {
|
||||
LOG.info("Publishing will message. CId={}, topic={}", clientID, will.getTopic());
|
||||
// it has just to publish the message downstream to the subscribers
|
||||
// NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
|
||||
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
|
||||
tobeStored.setClientID(clientID);
|
||||
Topic topic = new Topic(tobeStored.getTopic());
|
||||
this.messagesPublisher.publish2Subscribers(tobeStored, topic);
|
||||
|
||||
//Stores retained message to the topic
|
||||
if (will.isRetained()) {
|
||||
m_messagesStore.storeRetained(topic, tobeStored);
|
||||
}
|
||||
}
|
||||
|
||||
static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) {
|
||||
if (qos.value() > sub.getRequestedQos().value()) {
|
||||
qos = sub.getRequestedQos();
|
||||
}
|
||||
return qos;
|
||||
}
|
||||
|
||||
/**
|
||||
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored
|
||||
* message and publish to all interested subscribers.
|
||||
*
|
||||
* @param channel
|
||||
* the channel of the incoming message.
|
||||
* @param msg
|
||||
* the decoded pubrel message.
|
||||
*/
|
||||
public void processPubRel(Channel channel, MqttMessage msg) {
|
||||
this.qos2PublishHandler.processPubRel(channel, msg);
|
||||
}
|
||||
|
||||
public void processPubRec(Channel channel, MqttMessage msg) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
int messageID = messageId(msg);
|
||||
LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID);
|
||||
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
// remove from the inflight and move to the QoS2 second phase queue
|
||||
StoredMessage ackedMsg = targetSession.inFlightAcknowledged(messageID);
|
||||
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID, ackedMsg);
|
||||
// once received a PUBREC reply with a PUBREL(messageID)
|
||||
LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID);
|
||||
|
||||
MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0);
|
||||
MqttMessage pubRelMessage = new MqttMessage(pubRelHeader, from(messageID));
|
||||
channel.writeAndFlush(pubRelMessage).addListener(FIRE_EXCEPTION_ON_FAILURE);
|
||||
}
|
||||
|
||||
public void processPubComp(Channel channel, MqttMessage msg) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
int messageID = messageId(msg);
|
||||
LOG.debug("Processing PUBCOMP message. CId={}, messageId={}", clientID, messageID);
|
||||
// once received the PUBCOMP then remove the message from the temp memory
|
||||
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
StoredMessage inflightMsg = targetSession.completeReleasedPublish(messageID);
|
||||
String username = NettyUtils.userName(channel);
|
||||
String topic = inflightMsg.getTopic();
|
||||
final InterceptAcknowledgedMessage interceptAckMsg = new InterceptAcknowledgedMessage(inflightMsg, topic,
|
||||
username, messageID);
|
||||
m_interceptor.notifyMessageAcknowledged(interceptAckMsg);
|
||||
}
|
||||
|
||||
public void processDisconnect(Channel channel) {
|
||||
final String clientID = NettyUtils.clientID(channel);
|
||||
LOG.debug("Processing DISCONNECT message. CId={}", clientID);
|
||||
channel.flush();
|
||||
disconnectHandler.processDisconnect(channel, clientID);
|
||||
}
|
||||
|
||||
public void processConnectionLost(String clientID, Channel channel) {
|
||||
LOG.info("Lost connection with client <{}>", clientID);
|
||||
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
|
||||
connectionDescriptors.removeConnection(oldConnDescr);
|
||||
// publish the Will message (if any) for the clientID
|
||||
final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
WillMessage will = clientSession.willMessage();
|
||||
if (will != null) {
|
||||
forwardPublishWill(will, clientID);
|
||||
clientSession.removeWill();
|
||||
}
|
||||
|
||||
String username = NettyUtils.userName(channel);
|
||||
m_interceptor.notifyClientConnectionLost(clientID, username);
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the clientID from topic subscription, if not previously subscribed, doesn't reply any
|
||||
* error.
|
||||
*
|
||||
* @param channel
|
||||
* the channel of the incoming message.
|
||||
* @param msg
|
||||
* the decoded unsubscribe message.
|
||||
*/
|
||||
public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) {
|
||||
List<String> topics = msg.payload().topics();
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
|
||||
LOG.debug("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics);
|
||||
|
||||
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
for (String t : topics) {
|
||||
Topic topic = new Topic(t);
|
||||
boolean validTopic = topic.isValid();
|
||||
if (!validTopic) {
|
||||
// close the connection, not valid topicFilter is a protocol violation
|
||||
channel.close().addListener(CLOSE_ON_FAILURE);
|
||||
LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic);
|
||||
return;
|
||||
}
|
||||
|
||||
LOG.trace("Removing subscription. CId={}, topic={}", clientID, topic);
|
||||
subscriptions.removeSubscription(topic, clientID);
|
||||
clientSession.unsubscribeFrom(topic);
|
||||
String username = NettyUtils.userName(channel);
|
||||
m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username);
|
||||
}
|
||||
|
||||
// ack the client
|
||||
int messageID = msg.variableHeader().messageId();
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_MOST_ONCE,
|
||||
false, 0);
|
||||
MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID));
|
||||
|
||||
LOG.debug("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID);
|
||||
channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE);
|
||||
LOG.info("Client <{}> unsubscribed from topics <{}>", clientID, topics);
|
||||
}
|
||||
|
||||
public void processSubscribe(Channel channel, MqttSubscribeMessage msg) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
int messageID = messageId(msg);
|
||||
LOG.debug("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID);
|
||||
|
||||
RunningSubscription executionKey = new RunningSubscription(clientID, messageID);
|
||||
SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED);
|
||||
if (currentStatus != null) {
|
||||
LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}",
|
||||
clientID, messageID);
|
||||
return;
|
||||
}
|
||||
String username = NettyUtils.userName(channel);
|
||||
List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg);
|
||||
MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID);
|
||||
if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) {
|
||||
LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " +
|
||||
"messageId={}", clientID, messageID);
|
||||
return;
|
||||
}
|
||||
|
||||
LOG.debug("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics);
|
||||
|
||||
List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID);
|
||||
|
||||
// save session, persist subscriptions from session
|
||||
for (Subscription subscription : newSubscriptions) {
|
||||
subscriptions.add(subscription);
|
||||
}
|
||||
|
||||
LOG.debug("Sending SUBACK response CId={}, messageId={}", clientID, messageID);
|
||||
channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE);
|
||||
|
||||
// fire the persisted messages in session
|
||||
for (Subscription subscription : newSubscriptions) {
|
||||
publishRetainedMessagesInSession(subscription, username);
|
||||
}
|
||||
|
||||
boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED);
|
||||
if (!success) {
|
||||
LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID);
|
||||
} else {
|
||||
LOG.info("Client <{}> subscribed to topics", clientID);
|
||||
}
|
||||
}
|
||||
|
||||
private List<Subscription> doStoreSubscription(List<MqttTopicSubscription> ackTopics, String clientID) {
|
||||
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
|
||||
List<Subscription> newSubscriptions = new ArrayList<>();
|
||||
for (MqttTopicSubscription req : ackTopics) {
|
||||
// TODO this is SUPER UGLY
|
||||
if (req.qualityOfService() == FAILURE) {
|
||||
continue;
|
||||
}
|
||||
final Topic topic = new Topic(req.topicName());
|
||||
Subscription newSubscription = new Subscription(clientID, topic, req.qualityOfService());
|
||||
|
||||
clientSession.subscribe(newSubscription);
|
||||
newSubscriptions.add(newSubscription);
|
||||
}
|
||||
return newSubscriptions;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param clientID
|
||||
* the clientID
|
||||
* @param username
|
||||
* the username
|
||||
* @param msg
|
||||
* the subscribe message to verify
|
||||
* @return the list of verified topics for the given subscribe message.
|
||||
*/
|
||||
private List<MqttTopicSubscription> doVerify(String clientID, String username, MqttSubscribeMessage msg) {
|
||||
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
List<MqttTopicSubscription> ackTopics = new ArrayList<>();
|
||||
|
||||
final int messageId = messageId(msg);
|
||||
for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
|
||||
Topic topic = new Topic(req.topicName());
|
||||
if (!m_authorizator.canRead(topic, username, clientSession.clientID)) {
|
||||
// send SUBACK with 0x80, the user hasn't credentials to read the topic
|
||||
LOG.warn("Client does not have read permissions on the topic CId={}, username={}, messageId={}, " +
|
||||
"topic={}", clientID, username, messageId, topic);
|
||||
ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE));
|
||||
} else {
|
||||
MqttQoS qos;
|
||||
if (topic.isValid()) {
|
||||
LOG.debug("Client will be subscribed to the topic CId={}, username={}, messageId={}, topic={}",
|
||||
clientID, username, messageId, topic);
|
||||
qos = req.qualityOfService();
|
||||
} else {
|
||||
LOG.warn("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", clientID,
|
||||
username, messageId, topic);
|
||||
qos = FAILURE;
|
||||
}
|
||||
ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));
|
||||
}
|
||||
}
|
||||
return ackTopics;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the SUBACK response from a list of topicFilters
|
||||
*/
|
||||
private MqttSubAckMessage doAckMessageFromValidateFilters(List<MqttTopicSubscription> topicFilters, int messageId) {
|
||||
List<Integer> grantedQoSLevels = new ArrayList<>();
|
||||
for (MqttTopicSubscription req : topicFilters) {
|
||||
grantedQoSLevels.add(req.qualityOfService().value());
|
||||
}
|
||||
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_MOST_ONCE, false, 0);
|
||||
MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);
|
||||
return new MqttSubAckMessage(fixedHeader, from(messageId), payload);
|
||||
}
|
||||
|
||||
private void publishRetainedMessagesInSession(Subscription newSubscription, String username) {
|
||||
LOG.debug("Retrieving retained messages CId={}, topics={}", newSubscription.getClientId(),
|
||||
newSubscription.getTopicFilter());
|
||||
|
||||
// scans retained messages to be published to the new subscription
|
||||
// TODO this is ugly, it does a linear scan on potential big dataset
|
||||
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore
|
||||
.searchMatching(key -> key.match(newSubscription.getTopicFilter()));
|
||||
|
||||
if (!messages.isEmpty()) {
|
||||
LOG.info("Publishing retained messages CId={}, topics={}, messagesNo={}",
|
||||
newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size());
|
||||
}
|
||||
ClientSession targetSession = this.sessionsRepository.sessionForClient(newSubscription.getClientId());
|
||||
|
||||
// adapt the message QoS to subscriber's accepted accepted
|
||||
for (IMessagesStore.StoredMessage msg : messages) {
|
||||
int lowestQoS = Math.min(msg.getQos().value(), newSubscription.getRequestedQos().value());
|
||||
MqttQoS qosToPublish = MqttQoS.valueOf(lowestQoS);
|
||||
msg.setQos(qosToPublish);
|
||||
}
|
||||
|
||||
this.internalRepublisher.publishRetained(targetSession, messages);
|
||||
|
||||
// notify the Observables
|
||||
m_interceptor.notifyTopicSubscribed(newSubscription, username);
|
||||
}
|
||||
|
||||
public void notifyChannelWritable(Channel channel) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
boolean emptyQueue = false;
|
||||
while (channel.isWritable() && !emptyQueue) {
|
||||
EnqueuedMessage msg = clientSession.poll();
|
||||
if (msg == null) {
|
||||
emptyQueue = true;
|
||||
} else {
|
||||
// recreate a publish from stored publish in queue
|
||||
MqttPublishMessage pubMsg = createPublishForQos(msg.msg.getTopic(), msg.msg.getQos(),
|
||||
msg.msg.getPayload(), msg.msg.isRetained(),
|
||||
msg.messageId);
|
||||
channel.write(pubMsg).addListener(FIRE_EXCEPTION_ON_FAILURE);
|
||||
}
|
||||
}
|
||||
channel.flush();
|
||||
}
|
||||
|
||||
public void addInterceptHandler(InterceptHandler interceptHandler) {
|
||||
this.m_interceptor.addInterceptHandler(interceptHandler);
|
||||
}
|
||||
|
||||
public void removeInterceptHandler(InterceptHandler interceptHandler) {
|
||||
this.m_interceptor.removeInterceptHandler(interceptHandler);
|
||||
}
|
||||
|
||||
public IMessagesStore getMessagesStore() {
|
||||
return m_messagesStore;
|
||||
}
|
||||
|
||||
public ISessionsStore getSessionsStore() {
|
||||
return m_sessionsStore;
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
if (m_interceptor != null)
|
||||
m_interceptor.stop();
|
||||
}
|
||||
}
|
||||
@@ -1,265 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.BrokerConstants;
|
||||
import io.moquette.interception.InterceptHandler;
|
||||
import io.moquette.server.ConnectionDescriptorStore;
|
||||
import io.moquette.server.Server;
|
||||
import io.moquette.server.config.IConfig;
|
||||
import io.moquette.server.config.IResourceLoader;
|
||||
import io.moquette.spi.IMessagesStore;
|
||||
import io.moquette.spi.ISessionsStore;
|
||||
import io.moquette.spi.IStore;
|
||||
import io.moquette.spi.ISubscriptionsStore;
|
||||
import io.moquette.spi.impl.security.*;
|
||||
import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
|
||||
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
|
||||
import io.moquette.spi.impl.subscriptions.Subscription;
|
||||
import io.moquette.spi.security.IAuthenticator;
|
||||
import io.moquette.spi.security.IAuthorizatorPolicy;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.text.ParseException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
|
||||
/**
|
||||
* It's main responsibility is bootstrap the ProtocolProcessor.
|
||||
*/
|
||||
public class ProtocolProcessorBootstrapper {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessorBootstrapper.class);
|
||||
public static final String INMEMDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService";
|
||||
|
||||
private ISessionsStore m_sessionsStore;
|
||||
|
||||
private ISubscriptionsStore subscriptionsStore;
|
||||
|
||||
private Runnable storeShutdown;
|
||||
|
||||
private final ProtocolProcessor m_processor = new ProtocolProcessor();
|
||||
private ConnectionDescriptorStore connectionDescriptors;
|
||||
|
||||
public ProtocolProcessorBootstrapper() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the processing part of the broker.
|
||||
*
|
||||
* @param props
|
||||
* the properties carrier where some props like port end host could be loaded. For
|
||||
* the full list check of configurable properties check moquette.conf file.
|
||||
* @param embeddedObservers
|
||||
* a list of callbacks to be notified of certain events inside the broker. Could be
|
||||
* empty list of null.
|
||||
* @param authenticator
|
||||
* an implementation of the authenticator to be used, if null load that specified in
|
||||
* config and fallback on the default one (permit all).
|
||||
* @param authorizator
|
||||
* an implementation of the authorizator to be used, if null load that specified in
|
||||
* config and fallback on the default one (permit all).
|
||||
* @param server
|
||||
* the server to init.
|
||||
* @return the processor created for the broker.
|
||||
*/
|
||||
public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> embeddedObservers,
|
||||
IAuthenticator authenticator, IAuthorizatorPolicy authorizator, Server server) {
|
||||
IMessagesStore messagesStore;
|
||||
LOG.info("Initializing messages and sessions stores...");
|
||||
String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, INMEMDB_STORE_CLASS);
|
||||
if (storageClassName == null || storageClassName.isEmpty()) {
|
||||
LOG.error("storage_class property not defined");
|
||||
throw new IllegalArgumentException("Can't find a valid persistence layer");
|
||||
}
|
||||
final IStore store = instantiateConfiguredStore(storageClassName, props, server.getScheduler());
|
||||
if (store == null) {
|
||||
throw new IllegalArgumentException("Can't start the persistence layer");
|
||||
}
|
||||
store.initStore();
|
||||
messagesStore = store.messagesStore();
|
||||
m_sessionsStore = store.sessionsStore();
|
||||
SessionsRepository sessionsRepository = new SessionsRepository(this.m_sessionsStore, server.getScheduler());
|
||||
sessionsRepository.init();
|
||||
this.subscriptionsStore = m_sessionsStore.subscriptionStore();
|
||||
storeShutdown = new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
store.close();
|
||||
}
|
||||
};
|
||||
|
||||
LOG.info("Configuring message interceptors...");
|
||||
|
||||
List<InterceptHandler> observers = new ArrayList<>(embeddedObservers);
|
||||
String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
|
||||
if (interceptorClassName != null && !interceptorClassName.isEmpty()) {
|
||||
InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, Server.class, server);
|
||||
if (handler != null) {
|
||||
observers.add(handler);
|
||||
}
|
||||
}
|
||||
BrokerInterceptor interceptor = new BrokerInterceptor(props, observers);
|
||||
|
||||
LOG.info("Initializing subscriptions store...");
|
||||
ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
|
||||
subscriptions.init(sessionsRepository);
|
||||
|
||||
LOG.info("Configuring MQTT authenticator...");
|
||||
String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
|
||||
|
||||
if (authenticator == null && !authenticatorClassName.isEmpty()) {
|
||||
authenticator = loadClass(authenticatorClassName, IAuthenticator.class, IConfig.class, props);
|
||||
}
|
||||
|
||||
IResourceLoader resourceLoader = props.getResourceLoader();
|
||||
if (authenticator == null) {
|
||||
String passwdPath = props.getProperty(BrokerConstants.PASSWORD_FILE_PROPERTY_NAME, "");
|
||||
if (passwdPath.isEmpty()) {
|
||||
authenticator = new AcceptAllAuthenticator();
|
||||
} else {
|
||||
authenticator = new ResourceAuthenticator(resourceLoader, passwdPath);
|
||||
}
|
||||
LOG.info("An {} authenticator instance will be used", authenticator.getClass().getName());
|
||||
}
|
||||
|
||||
LOG.info("Configuring MQTT authorizator...");
|
||||
String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "");
|
||||
if (authorizator == null && !authorizatorClassName.isEmpty()) {
|
||||
authorizator = loadClass(authorizatorClassName, IAuthorizatorPolicy.class, IConfig.class, props);
|
||||
}
|
||||
|
||||
if (authorizator == null) {
|
||||
String aclFilePath = props.getProperty(BrokerConstants.ACL_FILE_PROPERTY_NAME, "");
|
||||
if (aclFilePath != null && !aclFilePath.isEmpty()) {
|
||||
authorizator = new DenyAllAuthorizatorPolicy();
|
||||
try {
|
||||
LOG.info("Parsing ACL file. Path = {}", aclFilePath);
|
||||
authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
|
||||
} catch (ParseException pex) {
|
||||
LOG.error("Unable to parse ACL file. path=" + aclFilePath, pex);
|
||||
}
|
||||
} else {
|
||||
authorizator = new PermitAllAuthorizatorPolicy();
|
||||
}
|
||||
LOG.info("An {} authorizator instance will be used", authorizator.getClass().getName());
|
||||
}
|
||||
|
||||
LOG.info("Initializing connection descriptor store...");
|
||||
connectionDescriptors = new ConnectionDescriptorStore();
|
||||
|
||||
LOG.info("Initializing MQTT protocol processor...");
|
||||
boolean allowAnonymous = props.boolProp(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
|
||||
boolean allowZeroByteClientId = props.boolProp(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, false);
|
||||
boolean reauthorizeSubscriptionsOnConnect = props.boolProp(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, false);
|
||||
m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator,
|
||||
allowAnonymous, allowZeroByteClientId, authorizator, interceptor,
|
||||
sessionsRepository, reauthorizeSubscriptionsOnConnect);
|
||||
return m_processor;
|
||||
}
|
||||
|
||||
private IStore instantiateConfiguredStore(String storageClassName, IConfig props,
|
||||
ScheduledExecutorService scheduledExecutor) {
|
||||
LOG.info("Loading storage class {}", storageClassName);
|
||||
Class<? extends IStore> storageClass;
|
||||
try {
|
||||
storageClass = this.getClass().getClassLoader()
|
||||
.loadClass(storageClassName)
|
||||
.asSubclass(IStore.class);
|
||||
} catch (ClassNotFoundException cnfex) {
|
||||
LOG.error("Cannot find storage class " + storageClassName + " in classpath", cnfex);
|
||||
return null;
|
||||
}
|
||||
|
||||
final Constructor<? extends IStore> constructor;
|
||||
try {
|
||||
constructor = storageClass
|
||||
.getConstructor(IConfig.class, ScheduledExecutorService.class);
|
||||
} catch (NoSuchMethodException nsmex) {
|
||||
LOG.error("Cannot find constructor with required params IConfig, ScheduledExecutorService ", nsmex);
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return constructor.newInstance(props, scheduledExecutor);
|
||||
} catch (InstantiationException | IllegalAccessException | InvocationTargetException ex) {
|
||||
LOG.error("Cannot instantiate the " + storageClassName + " instance", ex);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T, U> T loadClass(String className, Class<T> intrface, Class<U> constructorArgClass, U props) {
|
||||
T instance = null;
|
||||
try {
|
||||
// check if constructor with constructor arg class parameter
|
||||
// exists
|
||||
LOG.info("Invoking constructor with {} argument. ClassName={}, interfaceName={}",
|
||||
constructorArgClass.getName(), className, intrface.getName());
|
||||
instance = this.getClass().getClassLoader()
|
||||
.loadClass(className)
|
||||
.asSubclass(intrface)
|
||||
.getConstructor(constructorArgClass)
|
||||
.newInstance(props);
|
||||
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
|
||||
LOG.warn("Unable to invoke constructor with {} argument. ClassName={}, interfaceName={}, cause={}, " +
|
||||
"errorMessage={}", constructorArgClass.getName(), className, intrface.getName(), ex.getCause(),
|
||||
ex.getMessage());
|
||||
return null;
|
||||
} catch (NoSuchMethodException | InvocationTargetException e) {
|
||||
try {
|
||||
LOG.info("Invoking default constructor. ClassName={}, interfaceName={}", className, intrface.getName());
|
||||
// fallback to default constructor
|
||||
instance = this.getClass().getClassLoader()
|
||||
.loadClass(className)
|
||||
.asSubclass(intrface)
|
||||
.getDeclaredConstructor().newInstance();
|
||||
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException |
|
||||
NoSuchMethodException | InvocationTargetException ex) {
|
||||
LOG.error("Unable to invoke default constructor. ClassName={}, interfaceName={}, cause={}, " +
|
||||
"errorMessage={}", className, intrface.getName(), ex.getCause(), ex.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
return instance;
|
||||
}
|
||||
|
||||
public ISessionsStore getSessionsStore() {
|
||||
return m_sessionsStore;
|
||||
}
|
||||
|
||||
public List<Subscription> getSubscriptions() {
|
||||
return this.subscriptionsStore.listAllSubscriptions();
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
if (storeShutdown != null)
|
||||
storeShutdown.run();
|
||||
if (m_processor != null)
|
||||
m_processor.shutdown();
|
||||
}
|
||||
|
||||
public ConnectionDescriptorStore getConnectionDescriptors() {
|
||||
return connectionDescriptors;
|
||||
}
|
||||
}
|
||||
@@ -1,68 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.server.netty.NettyUtils;
|
||||
import io.moquette.spi.IMessagesStore;
|
||||
import io.moquette.spi.impl.subscriptions.Topic;
|
||||
import io.moquette.spi.security.IAuthorizatorPolicy;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.handler.codec.mqtt.MqttPublishMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
|
||||
|
||||
class Qos0PublishHandler extends QosPublishHandler {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Qos0PublishHandler.class);
|
||||
|
||||
private final IMessagesStore m_messagesStore;
|
||||
private final BrokerInterceptor m_interceptor;
|
||||
private final MessagesPublisher publisher;
|
||||
|
||||
public Qos0PublishHandler(IAuthorizatorPolicy authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
|
||||
MessagesPublisher messagesPublisher) {
|
||||
super(authorizator);
|
||||
this.m_messagesStore = messagesStore;
|
||||
this.m_interceptor = interceptor;
|
||||
this.publisher = messagesPublisher;
|
||||
}
|
||||
|
||||
void receivedPublishQos0(Channel channel, MqttPublishMessage msg) {
|
||||
// verify if topic can be write
|
||||
final Topic topic = new Topic(msg.variableHeader().topicName());
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
String username = NettyUtils.userName(channel);
|
||||
if (!m_authorizator.canWrite(topic, username, clientID)) {
|
||||
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
|
||||
return;
|
||||
}
|
||||
|
||||
// route message to subscribers
|
||||
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
|
||||
toStoreMsg.setClientID(clientID);
|
||||
|
||||
this.publisher.publish2Subscribers(toStoreMsg, topic);
|
||||
|
||||
if (msg.fixedHeader().isRetain()) {
|
||||
// QoS == 0 && retain => clean old retained
|
||||
m_messagesStore.cleanRetained(topic);
|
||||
}
|
||||
|
||||
m_interceptor.notifyTopicPublished(msg, clientID, username);
|
||||
}
|
||||
}
|
||||
@@ -1,118 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
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.IMessagesStore;
|
||||
import io.moquette.spi.impl.subscriptions.Topic;
|
||||
import io.moquette.spi.security.IAuthorizatorPolicy;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.handler.codec.mqtt.MqttFixedHeader;
|
||||
import io.netty.handler.codec.mqtt.MqttMessageType;
|
||||
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
|
||||
import io.netty.handler.codec.mqtt.MqttPublishMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
|
||||
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
|
||||
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
|
||||
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
|
||||
|
||||
class Qos1PublishHandler extends QosPublishHandler {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class);
|
||||
|
||||
private final IMessagesStore m_messagesStore;
|
||||
private final BrokerInterceptor m_interceptor;
|
||||
private final IConnectionsManager connectionDescriptors;
|
||||
private final MessagesPublisher publisher;
|
||||
|
||||
Qos1PublishHandler(IAuthorizatorPolicy authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
|
||||
IConnectionsManager connectionDescriptors, MessagesPublisher messagesPublisher) {
|
||||
super(authorizator);
|
||||
this.m_messagesStore = messagesStore;
|
||||
this.m_interceptor = interceptor;
|
||||
this.connectionDescriptors = connectionDescriptors;
|
||||
this.publisher = messagesPublisher;
|
||||
}
|
||||
|
||||
void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {
|
||||
// verify if topic can be write
|
||||
final Topic topic = new Topic(msg.variableHeader().topicName());
|
||||
topic.getTokens();
|
||||
if (!topic.isValid()) {
|
||||
LOG.warn("Invalid topic format, force close the connection");
|
||||
channel.close().addListener(CLOSE_ON_FAILURE);
|
||||
return;
|
||||
}
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
String username = NettyUtils.userName(channel);
|
||||
if (!m_authorizator.canWrite(topic, username, clientID)) {
|
||||
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
|
||||
return;
|
||||
}
|
||||
|
||||
final int messageID = msg.variableHeader().messageId();
|
||||
|
||||
// route message to subscribers
|
||||
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
|
||||
toStoreMsg.setClientID(clientID);
|
||||
|
||||
this.publisher.publish2Subscribers(toStoreMsg, topic, messageID);
|
||||
|
||||
sendPubAck(clientID, messageID);
|
||||
|
||||
if (msg.fixedHeader().isRetain()) {
|
||||
if (!msg.payload().isReadable()) {
|
||||
m_messagesStore.cleanRetained(topic);
|
||||
} else {
|
||||
// before wasn't stored
|
||||
m_messagesStore.storeRetained(topic, toStoreMsg);
|
||||
}
|
||||
}
|
||||
|
||||
m_interceptor.notifyTopicPublished(msg, clientID, username);
|
||||
}
|
||||
|
||||
private void sendPubAck(String clientId, int messageID) {
|
||||
LOG.trace("sendPubAck invoked");
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE, false, 0);
|
||||
MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(fixedHeader, from(messageID));
|
||||
|
||||
try {
|
||||
LOG.trace("connected clientIDs are {}", connectionDescriptors.getConnectedClientIds());
|
||||
if (!connectionDescriptors.isConnected(clientId)) {
|
||||
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s",
|
||||
clientId, connectionDescriptors));
|
||||
}
|
||||
LOG.info("Sending {} message CId=<{}>", MqttMessageType.PUBACK, clientId);
|
||||
final Optional<ConnectionDescriptor> optDescriptor = connectionDescriptors.lookupDescriptor(clientId);
|
||||
if (optDescriptor.isPresent()) {
|
||||
final ConnectionDescriptor descriptor = optDescriptor.get();
|
||||
descriptor.writeAndFlush(pubAckMessage);
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
LOG.error(null, t);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,164 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
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.IMessagesStore;
|
||||
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
|
||||
import io.moquette.spi.impl.subscriptions.Topic;
|
||||
import io.moquette.spi.security.IAuthorizatorPolicy;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.handler.codec.mqtt.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
import static io.moquette.spi.impl.DebugUtils.payload2Str;
|
||||
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
|
||||
import static io.moquette.spi.impl.Utils.messageId;
|
||||
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
|
||||
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
|
||||
|
||||
class Qos2PublishHandler extends QosPublishHandler {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Qos2PublishHandler.class);
|
||||
|
||||
private final ISubscriptionsDirectory subscriptions;
|
||||
private final IMessagesStore m_messagesStore;
|
||||
private final BrokerInterceptor m_interceptor;
|
||||
private final IConnectionsManager connectionDescriptors;
|
||||
private final MessagesPublisher publisher;
|
||||
private final SessionsRepository sessionsRepository;
|
||||
|
||||
Qos2PublishHandler(IAuthorizatorPolicy authorizator, ISubscriptionsDirectory subscriptions,
|
||||
IMessagesStore messagesStore, BrokerInterceptor interceptor,
|
||||
IConnectionsManager connectionDescriptors,
|
||||
MessagesPublisher messagesPublisher, SessionsRepository sessionsRepository) {
|
||||
super(authorizator);
|
||||
this.subscriptions = subscriptions;
|
||||
this.m_messagesStore = messagesStore;
|
||||
this.m_interceptor = interceptor;
|
||||
this.connectionDescriptors = connectionDescriptors;
|
||||
this.publisher = messagesPublisher;
|
||||
this.sessionsRepository = sessionsRepository;
|
||||
}
|
||||
|
||||
void receivedPublishQos2(Channel channel, MqttPublishMessage msg) {
|
||||
final Topic topic = new Topic(msg.variableHeader().topicName());
|
||||
// check if the topic can be wrote
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
String username = NettyUtils.userName(channel);
|
||||
if (!m_authorizator.canWrite(topic, username, clientID)) {
|
||||
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
|
||||
return;
|
||||
}
|
||||
final int messageID = msg.variableHeader().messageId();
|
||||
|
||||
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
|
||||
toStoreMsg.setClientID(clientID);
|
||||
|
||||
LOG.info("Sending publish message to subscribers CId={}, topic={}, messageId={}", clientID, topic, messageID);
|
||||
if (LOG.isTraceEnabled()) {
|
||||
LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getPayload()), subscriptions.dumpTree());
|
||||
}
|
||||
|
||||
this.sessionsRepository.sessionForClient(clientID).markAsInboundInflight(messageID, toStoreMsg);
|
||||
|
||||
sendPubRec(clientID, messageID);
|
||||
|
||||
// Next the client will send us a pub rel
|
||||
// NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
|
||||
|
||||
// if (msg.fixedHeader().isRetain()) {
|
||||
// if (msg.payload().readableBytes() == 0) {
|
||||
// m_messagesStore.cleanRetained(topic);
|
||||
// } else {
|
||||
// m_messagesStore.storeRetained(topic, toStoreMsg);
|
||||
// }
|
||||
// }
|
||||
//TODO this should happen on PUB_REL, else we notify false positive
|
||||
m_interceptor.notifyTopicPublished(msg, clientID, username);
|
||||
}
|
||||
|
||||
/**
|
||||
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored
|
||||
* message and publish to all interested subscribers.
|
||||
*/
|
||||
void processPubRel(Channel channel, MqttMessage msg) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
int messageID = messageId(msg);
|
||||
LOG.info("Processing PUBREL message. CId={}, messageId={}", clientID, messageID);
|
||||
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
|
||||
IMessagesStore.StoredMessage evt = targetSession.inboundInflight(messageID);
|
||||
if (evt == null) {
|
||||
LOG.warn("Can't find inbound inflight message for CId={}, messageId={}", clientID, messageID);
|
||||
throw new IllegalArgumentException("Can't find inbound inflight message");
|
||||
}
|
||||
final Topic topic = new Topic(evt.getTopic());
|
||||
|
||||
this.publisher.publish2Subscribers(evt, topic, messageID);
|
||||
|
||||
if (evt.isRetained()) {
|
||||
if (evt.getPayload().readableBytes() == 0) {
|
||||
m_messagesStore.cleanRetained(topic);
|
||||
} else {
|
||||
m_messagesStore.storeRetained(topic, evt);
|
||||
}
|
||||
}
|
||||
|
||||
//TODO here we should notify to the listeners
|
||||
//m_interceptor.notifyTopicPublished(msg, clientID, username);
|
||||
|
||||
sendPubComp(clientID, messageID);
|
||||
}
|
||||
|
||||
private void sendPubRec(String clientID, int messageID) {
|
||||
LOG.trace("Sending PUBREC message. CId={}, messageId={}", clientID, messageID);
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE, false, 0);
|
||||
MqttMessage pubRecMessage = new MqttMessage(fixedHeader, from(messageID));
|
||||
final Optional<ConnectionDescriptor> connectionDescriptor = connectionDescriptors.lookupDescriptor(clientID);
|
||||
if (connectionDescriptor.isPresent()) {
|
||||
final ConnectionDescriptor descriptor = connectionDescriptor.get();
|
||||
try {
|
||||
descriptor.writeAndFlush(pubRecMessage);
|
||||
} catch (Throwable e) {
|
||||
LOG.error("Unable to send {} message. CId=<{}>, messageId={}", MqttMessageType.PUBREC, clientID,
|
||||
messageID, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void sendPubComp(String clientID, int messageID) {
|
||||
LOG.trace("Sending PUBCOMP message. CId={}, messageId={}", clientID, messageID);
|
||||
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0);
|
||||
MqttMessage pubCompMessage = new MqttMessage(fixedHeader, from(messageID));
|
||||
final Optional<ConnectionDescriptor> connectionDescriptor = connectionDescriptors.lookupDescriptor(clientID);
|
||||
if (connectionDescriptor.isPresent()) {
|
||||
final ConnectionDescriptor descriptor = connectionDescriptor.get();
|
||||
try {
|
||||
descriptor.writeAndFlush(pubCompMessage);
|
||||
} catch (Throwable e) {
|
||||
LOG.error("Unable to send {} message. CId=<{}>, messageId={}", MqttMessageType.PUBCOMP, clientID,
|
||||
messageID, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,45 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import io.moquette.server.netty.NettyUtils;
|
||||
import io.moquette.spi.impl.subscriptions.Topic;
|
||||
import io.moquette.spi.security.IAuthorizatorPolicy;
|
||||
import io.netty.channel.Channel;
|
||||
|
||||
abstract class QosPublishHandler {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(QosPublishHandler.class);
|
||||
|
||||
protected final IAuthorizatorPolicy m_authorizator;
|
||||
|
||||
protected QosPublishHandler(IAuthorizatorPolicy m_authorizator) {
|
||||
this.m_authorizator = m_authorizator;
|
||||
}
|
||||
|
||||
public boolean checkWriteOnTopic(Topic topic, Channel channel) {
|
||||
String clientID = NettyUtils.clientID(channel);
|
||||
String username = NettyUtils.userName(channel);
|
||||
if (!m_authorizator.canWrite(topic, username, clientID)) {
|
||||
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -1,48 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
class RunningSubscription {
|
||||
|
||||
final String clientID;
|
||||
final long packetId;
|
||||
|
||||
RunningSubscription(String clientID, long packetId) {
|
||||
this.clientID = clientID;
|
||||
this.packetId = packetId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o)
|
||||
return true;
|
||||
if (o == null || getClass() != o.getClass())
|
||||
return false;
|
||||
|
||||
RunningSubscription that = (RunningSubscription) o;
|
||||
|
||||
return packetId == that.packetId
|
||||
&& (clientID != null ? clientID.equals(that.clientID) : that.clientID == null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = clientID != null ? clientID.hashCode() : 0;
|
||||
result = 31 * result + (int) (packetId ^ (packetId >>> 32));
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@@ -1,158 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2018 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
package io.moquette.spi.impl;
|
||||
|
||||
import io.moquette.persistence.PersistentSession;
|
||||
import io.moquette.spi.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
public class SessionsRepository {
|
||||
|
||||
/**
|
||||
* Task to be scheduled to execute the cleaning of persisted sessions (clean flag=false) older than a
|
||||
* defined period.
|
||||
*/
|
||||
private class SessionCleanerTask implements Runnable {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
wipeExpiredSessions();
|
||||
}
|
||||
}
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(SessionsRepository.class);
|
||||
private final ISessionsStore sessions;
|
||||
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;
|
||||
this.subscriptionsStore = sessionsStore.subscriptionStore();
|
||||
this.scheduler = scheduler;
|
||||
}
|
||||
|
||||
@SuppressWarnings("FutureReturnValueIgnored")
|
||||
public void init() {
|
||||
SessionCleanerTask cleanerTask = new SessionCleanerTask();
|
||||
this.scheduler.schedule(cleanerTask, 1, TimeUnit.HOURS);
|
||||
}
|
||||
|
||||
public ClientSession sessionForClient(String clientID) {
|
||||
LOG.debug("Retrieving session. CId={}", clientID);
|
||||
if (this.sessionsCache.containsKey(clientID)) {
|
||||
return this.sessionsCache.get(clientID);
|
||||
}
|
||||
|
||||
if (!this.sessions.contains(clientID)) {
|
||||
LOG.warn("Session does not exist. CId={}", clientID);
|
||||
return null;
|
||||
}
|
||||
PersistentSession storedSession = this.sessions.loadSessionByKey(clientID);
|
||||
return newClientSessionAndCacheIt(storedSession.clientID, storedSession.cleanSession);
|
||||
}
|
||||
|
||||
private ClientSession newClientSessionAndCacheIt(String clientID, boolean cleanSession) {
|
||||
ClientSession session;
|
||||
if (cleanSession) {
|
||||
session = new TransientSession(clientID, this.willMessages);
|
||||
} else {
|
||||
DurableSession durableSession = new DurableSession(clientID, this.sessions, this.subscriptionsStore, willMessages);
|
||||
durableSession.reloadAllSubscriptionsFromStore();
|
||||
session = durableSession;
|
||||
}
|
||||
this.sessionsCache.put(clientID, session);
|
||||
return session;
|
||||
}
|
||||
|
||||
public ClientSession createNewSession(String clientID, boolean cleanSession) {
|
||||
if (sessions.contains(clientID)) {
|
||||
LOG.error("Unable to create a new session: the client ID is already in use. CId={}, cleanSession={}",
|
||||
clientID, cleanSession);
|
||||
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
|
||||
}
|
||||
LOG.debug("Creating new session. CId={}, cleanSession={}", clientID, cleanSession);
|
||||
if (!cleanSession) {
|
||||
sessions.createNewDurableSession(clientID);
|
||||
}
|
||||
return newClientSessionAndCacheIt(clientID, cleanSession);
|
||||
}
|
||||
|
||||
public Collection<ClientSession> getAllSessions() {
|
||||
Collection<ClientSession> result = new ArrayList<>();
|
||||
for (PersistentSession persistentSession : this.sessions.listAllSessions()) {
|
||||
result.add(sessionForClient(persistentSession.clientID));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
ClientSession createOrLoadClientSession(String clientId, boolean cleanSession) {
|
||||
ClientSession clientSession = this.sessionForClient(clientId);
|
||||
if (clientSession == null) {
|
||||
clientSession = this.createNewSession(clientId, cleanSession);
|
||||
} else {
|
||||
//session was already present
|
||||
if (!clientSession.isCleanSession() && cleanSession) {
|
||||
//remove from storage & create new transient session
|
||||
//TODO existing subscription from durable session has to be copied into the new transient?
|
||||
sessions.removeDurableSession(clientId);
|
||||
clientSession = this.createNewSession(clientId, true);
|
||||
} /*else {
|
||||
clientSession.cleanSession(cleanSession);
|
||||
this.updateCleanStatus(clientId, cleanSession);
|
||||
}*/
|
||||
}
|
||||
if (cleanSession) {
|
||||
LOG.info("Cleaning session. CId={}", clientId);
|
||||
clientSession.cleanSession();
|
||||
}
|
||||
return clientSession;
|
||||
}
|
||||
|
||||
public void disconnect(String clientId) {
|
||||
LOG.debug("Removing session from repository's cache");
|
||||
ClientSession clientSession = this.sessionForClient(clientId);
|
||||
if (clientSession == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
sessionsCache.remove(clientId);
|
||||
this.sessions.trackSessionClose(LocalDateTime.now(), clientId);
|
||||
}
|
||||
|
||||
private void wipeExpiredSessions() {
|
||||
final LocalDateTime pin = LocalDateTime.now().minus(6, ChronoUnit.DAYS);
|
||||
final Set<String> expiredSessionsIds = this.sessions.sessionOlderThan(pin);
|
||||
for (String expiredSession : expiredSessionsIds) {
|
||||
this.sessions.removeDurableSession(expiredSession);
|
||||
this.subscriptionsStore.wipeSubscriptions(expiredSession);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,36 +0,0 @@
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -82,7 +82,18 @@ class CNode {
|
||||
}
|
||||
|
||||
CNode addSubscription(Subscription newSubscription) {
|
||||
this.subscriptions.add(new Subscription(newSubscription));
|
||||
// if already contains one with same topic and same client, keep that with higher QoS
|
||||
if (subscriptions.contains(newSubscription)) {
|
||||
final Subscription existing = subscriptions.stream()
|
||||
.filter(s -> s.equals(newSubscription))
|
||||
.findFirst().get();
|
||||
if (existing.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
|
||||
subscriptions.remove(existing);
|
||||
subscriptions.add(new Subscription(newSubscription));
|
||||
}
|
||||
} else {
|
||||
this.subscriptions.add(new Subscription(newSubscription));
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
+9
-54
@@ -15,15 +15,12 @@
|
||||
*/
|
||||
package io.moquette.spi.impl.subscriptions;
|
||||
|
||||
import io.moquette.spi.ClientSession;
|
||||
import io.moquette.spi.impl.SessionsRepository;
|
||||
import io.moquette.broker.ISubscriptionsRepository;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
import static java.lang.String.format;
|
||||
|
||||
public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(CTrieSubscriptionDirectory.class);
|
||||
@@ -32,7 +29,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
private static final INode NO_PARENT = null;
|
||||
|
||||
INode root;
|
||||
private volatile SessionsRepository sessionsRepository;
|
||||
private volatile ISubscriptionsRepository sessionsRepository;
|
||||
|
||||
interface IVisitor<T> {
|
||||
|
||||
@@ -46,7 +43,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void init(SessionsRepository sessionsRepository) {
|
||||
public void init(ISubscriptionsRepository sessionsRepository) {
|
||||
LOG.info("Initializing CTrie");
|
||||
final CNode mainNode = new CNode();
|
||||
mainNode.token = ROOT;
|
||||
@@ -58,12 +55,11 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
if (LOG.isTraceEnabled()) {
|
||||
LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}", dumpTree());
|
||||
}
|
||||
for (ClientSession session : this.sessionsRepository.getAllSessions()) {
|
||||
for (Subscription subscription : session.getSubscriptions()) {
|
||||
LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", subscription.clientId,
|
||||
subscription.topicFilter);
|
||||
add(subscription);
|
||||
}
|
||||
|
||||
for (Subscription subscription : this.sessionsRepository.listAllSubscriptions()) {
|
||||
LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", subscription.clientId,
|
||||
subscription.topicFilter);
|
||||
add(subscription);
|
||||
}
|
||||
if (LOG.isTraceEnabled()) {
|
||||
LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree());
|
||||
@@ -86,7 +82,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
|
||||
@Override
|
||||
public List<Subscription> matches(Topic topic) {
|
||||
return new ArrayList<>(match(topic));
|
||||
return new ArrayList<>(matchQosSharpening(topic));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -118,47 +114,6 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
|
||||
return new HashSet<>(subsGroupedByClient.values());
|
||||
}
|
||||
|
||||
/**
|
||||
* Given a topic string return the clients subscriptions that matches it. Topic string can't
|
||||
* contain character # and + because they are reserved to listeners subscriptions, and not topic
|
||||
* publishing.
|
||||
*
|
||||
* @param topic
|
||||
* to use fo searching matching subscriptions.
|
||||
* @return the list of matching subscriptions, or empty if not matching.
|
||||
*/
|
||||
@Deprecated
|
||||
Set<Subscription> match(Topic topic) {
|
||||
final Set<Subscription> matchingSubs = matchWithoutQosSharpening(topic);
|
||||
// remove the overlapping subscriptions, selecting ones with greatest qos
|
||||
Map<String, Subscription> subsForClient = loadSubscriptionsFromStorage(matchingSubs);
|
||||
return new HashSet<>(subsForClient.values());
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
private Map<String, Subscription> loadSubscriptionsFromStorage(Set<Subscription> matchingSubs) {
|
||||
Map<String, Subscription> subsForClient = new HashMap<>();
|
||||
for (Subscription matchingSub : matchingSubs) {
|
||||
Subscription existingSub = subsForClient.get(matchingSub.clientId);
|
||||
final ClientSession subscribedSession = this.sessionsRepository.sessionForClient(matchingSub.clientId);
|
||||
if (subscribedSession == null) {
|
||||
//clean session disconnected
|
||||
continue;
|
||||
}
|
||||
Subscription sub = subscribedSession.findSubscriptionByTopicFilter(matchingSub);
|
||||
if (sub == null) {
|
||||
final String excpMesg = format("Target session %s is connected but doesn't anymore subscribed to %s",
|
||||
matchingSub.clientId, matchingSub);
|
||||
throw new IllegalStateException(excpMesg);
|
||||
}
|
||||
// update the selected subscriptions if not present or if has a greater qos
|
||||
if (existingSub == null || existingSub.qosLessThan(sub)) {
|
||||
subsForClient.put(sub.clientId, sub);
|
||||
}
|
||||
}
|
||||
return subsForClient;
|
||||
}
|
||||
|
||||
Set<Subscription> recursiveMatch(Topic topic, INode inode) {
|
||||
CNode cnode = inode.mainNode();
|
||||
if (Token.MULTI.equals(cnode.token)) {
|
||||
|
||||
+2
-2
@@ -15,14 +15,14 @@
|
||||
*/
|
||||
package io.moquette.spi.impl.subscriptions;
|
||||
|
||||
import io.moquette.spi.impl.SessionsRepository;
|
||||
import io.moquette.broker.ISubscriptionsRepository;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
public interface ISubscriptionsDirectory {
|
||||
|
||||
void init(SessionsRepository sessionsRepository);
|
||||
void init(ISubscriptionsRepository sessionsRepository);
|
||||
|
||||
Set<Subscription> matchWithoutQosSharpening(Topic topic);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user