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

Minor, fixed logging level to be more usefull for distribution

Browse Source
This commit is contained in:
andsel
2018-04-27 15:56:26 +02:00
parent d8e5998f21
commit 99f0612209
8 changed files with 80 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/ConnectionDescriptor.java
+5 -12
View File
@@ -87,7 +87,7 @@ public class ConnectionDescriptor {
}
public boolean close() {
LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID);
LOG.debug("Closing connection descriptor. MqttClientId = {}.", clientID);
final boolean success = assignState(ConnectionState.INTERCEPTORS_NOTIFIED, ConnectionState.DISCONNECTED);
if (!success) {
return false;
@@ -111,19 +111,12 @@ public class ConnectionDescriptor {
}
public boolean assignState(ConnectionState expected, ConnectionState newState) {
LOG.debug(
"Updating state of connection descriptor. MqttClientId = {}, expectedState = {}, newState = {}.",
clientID,
expected,
newState);
LOG.debug("Updating state of connection descriptor. MqttClientId = {}, expectedState = {}, newState = {}.",
clientID, expected, newState);
boolean retval = channelState.compareAndSet(expected, newState);
if (!retval) {
LOG.error(
"Unable to update state of connection descriptor."
+ " MqttclientId = {}, expectedState = {}, newState = {}.",
clientID,
expected,
newState);
LOG.error("Unable to update state of connection descriptor. CId = {}, expectedState = {}, newState = {}.",
clientID, expected, newState);
}
return retval;
}
broker/src/main/java/io/moquette/server/Server.java
+10 -9
View File
@@ -124,7 +124,7 @@ public class Server {
* @throws IOException in case of any IO Error.
*/
public void startServer(Properties configProps) throws IOException {
LOG.info("Starting Moquette server using properties object");
LOG.debug("Starting Moquette server using properties object");
final IConfig config = new MemoryConfig(configProps);
startServer(config);
}
@@ -136,7 +136,7 @@ public class Server {
* @throws IOException in case of any IO Error.
*/
public void startServer(IConfig config) throws IOException {
LOG.info("Starting Moquette server using IConfig instance...");
LOG.debug("Starting Moquette server using IConfig instance");
startServer(config, null);
}
@@ -149,16 +149,17 @@ public class Server {
* @throws IOException in case of any IO Error.
*/
public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
LOG.info("Starting moquette server using IConfig instance and intercept handlers");
LOG.debug("Starting moquette server using IConfig instance and intercept handlers");
startServer(config, handlers, null, null, null);
}
public void startServer(IConfig config, List<? extends InterceptHandler> handlers, ISslContextCreator sslCtxCreator,
IAuthenticator authenticator, IAuthorizator authorizator) throws IOException {
final long start = System.currentTimeMillis();
if (handlers == null) {
handlers = Collections.emptyList();
}
LOG.info("Starting Moquette Server. MQTT message interceptors={}", getInterceptorIds(handlers));
LOG.trace("Starting Moquette Server. MQTT message interceptors={}", getInterceptorIds(handlers));
scheduler = Executors.newScheduledThreadPool(1);
@@ -168,22 +169,22 @@ public class Server {
}
configureCluster(config);
final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME);
LOG.info("Configuring Using persistent store file, path={}", persistencePath);
LOG.debug("Configuring Using persistent store file, path={}", persistencePath);
m_processorBootstrapper = new ProtocolProcessorBootstrapper();
final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator,
this);
LOG.info("Initialized MQTT protocol processor");
LOG.debug("Initialized MQTT protocol processor");
if (sslCtxCreator == null) {
LOG.warn("Using default SSL context creator");
LOG.info("Using default SSL context creator");
sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
}
LOG.info("Binding server to the configured ports");
m_acceptor = new NettyAcceptor();
m_acceptor.initialize(processor, config, sslCtxCreator);
m_processor = processor;
LOG.info("Moquette server has been initialized successfully");
final long startTime = System.currentTimeMillis() - start;
LOG.info("Moquette server has been started successfully in {} ms", startTime);
m_initialized = true;
}
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+17 -26
View File
@@ -110,9 +110,8 @@ public class NettyAcceptor implements ServerAcceptor {
private Class<? extends ServerSocketChannel> channelClass;
@Override
public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator)
throws IOException {
LOG.info("Initializing Netty acceptor");
public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator) {
LOG.debug("Initializing Netty acceptor");
nettySoBacklog = props.intProp(BrokerConstants.NETTY_SO_BACKLOG_PROPERTY_NAME, 128);
nettySoReuseaddr = props.boolProp(BrokerConstants.NETTY_SO_REUSEADDR_PROPERTY_NAME, true);
@@ -170,7 +169,7 @@ public class NettyAcceptor implements ServerAcceptor {
}
private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
LOG.info("Initializing server. Protocol={}", protocol);
LOG.debug("Initializing server. Protocol={}", protocol);
ServerBootstrap b = new ServerBootstrap();
b.group(m_bossGroup, m_workerGroup).channel(channelClass)
.childHandler(new ChannelInitializer<SocketChannel>() {
@@ -191,19 +190,18 @@ public class NettyAcceptor implements ServerAcceptor {
.childOption(ChannelOption.TCP_NODELAY, nettyTcpNodelay)
.childOption(ChannelOption.SO_KEEPALIVE, nettySoKeepalive);
try {
LOG.info("Binding server. host={}, port={}", host, port);
LOG.debug("Binding server. host={}, port={}", host, port);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(host, port);
LOG.info("Server has been bound. host={}, port={}", host, port);
LOG.info("Server bound to host={}, port={}, protocol={}", host, port, protocol);
f.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);
} catch (InterruptedException ex) {
LOG.error("An interruptedException was caught while initializing server. Protocol={}", protocol, ex);
}
}
private void initializePlainTCPTransport(final NettyMQTTHandler handler,
IConfig props) throws IOException {
LOG.info("Configuring TCP MQTT transport");
private void initializePlainTCPTransport(NettyMQTTHandler handler, IConfig props) {
LOG.debug("Configuring TCP MQTT transport");
final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
String tcpPortProp = props.getProperty(PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
@@ -236,8 +234,8 @@ public class NettyAcceptor implements ServerAcceptor {
});
}
private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
LOG.info("Configuring Websocket MQTT transport");
private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) {
LOG.debug("Configuring Websocket MQTT transport");
String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
// Do nothing no WebSocket configured
@@ -272,9 +270,8 @@ public class NettyAcceptor implements ServerAcceptor {
});
}
private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext)
throws IOException {
LOG.info("Configuring SSL MQTT transport");
private void initializeSSLTCPTransport(NettyMQTTHandler handler, IConfig props, SSLContext sslContext) {
LOG.debug("Configuring SSL MQTT transport");
String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(sslPortProp)) {
// Do nothing no SSL configured
@@ -284,7 +281,7 @@ public class NettyAcceptor implements ServerAcceptor {
}
int sslPort = Integer.parseInt(sslPortProp);
LOG.info("Starting SSL on port {}", sslPort);
LOG.debug("Starting SSL on port {}", sslPort);
final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
@@ -308,9 +305,8 @@ public class NettyAcceptor implements ServerAcceptor {
});
}
private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext)
throws IOException {
LOG.info("Configuring secure websocket MQTT transport");
private void initializeWSSTransport(NettyMQTTHandler handler, IConfig props, SSLContext sslContext) {
LOG.debug("Configuring secure websocket MQTT transport");
String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(sslPortProp)) {
// Do nothing no SSL configured
@@ -350,7 +346,7 @@ public class NettyAcceptor implements ServerAcceptor {
@SuppressWarnings("FutureReturnValueIgnored")
@Override
public void close() {
LOG.info("Closing Netty acceptor...");
LOG.debug("Closing Netty acceptor...");
if (m_workerGroup == null || m_bossGroup == null) {
LOG.error("Netty acceptor is not initialized");
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
@@ -380,15 +376,10 @@ public class NettyAcceptor implements ServerAcceptor {
m_bossGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
}
LOG.info("Collecting message metrics...");
MessageMetrics metrics = m_metricsCollector.computeMetrics();
LOG.info("Metrics have been collected. Read messages={}, written messages={}", metrics.messagesRead(),
metrics.messagesWrote());
LOG.info("Collecting bytes metrics...");
BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics();
LOG.info("Bytes metrics have been collected. Read bytes={}, written bytes={}", bytesMetrics.readBytes(),
bytesMetrics.wroteBytes());
LOG.info("Metrics messages[read={}, write={}] bytes[read={}, write={}]", metrics.messagesRead(),
metrics.messagesWrote(), bytesMetrics.readBytes(), bytesMetrics.wroteBytes());
}
private ChannelHandler createSslHandler(SSLContext sslContext, boolean needsClientAuth) {
broker/src/main/java/io/moquette/spi/TransientSession.java
+9 -10
View File
@@ -52,11 +52,11 @@ public class TransientSession extends ClientSession {
@Override
public boolean subscribe(Subscription newSubscription) {
LOG.info("Adding new subscription. CId={}, topics={}, qos={}", newSubscription.getClientId(),
LOG.trace("Adding new subscription. CId={}, topics={}, qos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), newSubscription.getRequestedQos());
boolean validTopic = newSubscription.getTopicFilter().isValid();
if (!validTopic) {
LOG.warn("The topic filter is not valid. CId={}, topics={}", newSubscription.getClientId(),
LOG.warn("Invalid topic filter. CId={}, topicFilter={}", newSubscription.getClientId(),
newSubscription.getTopicFilter());
// send SUBACK with 0x80 for this topic filter
return false;
@@ -67,7 +67,7 @@ public class TransientSession extends ClientSession {
if (mathingExisting.isPresent()) {
Subscription existingSub = mathingExisting.get();
if (existingSub.qosLessThan(newSubscription)) {
LOG.info("Subscription already existed with a lower QoS value. It will be updated. CId={}, " +
LOG.debug("Subscription already existed with a lower QoS value. It will be updated. CId={}, " +
"topics={}, existingQos={}, newQos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos());
subscriptions.remove(newSubscription);
@@ -80,7 +80,7 @@ public class TransientSession extends ClientSession {
@Override
public void unsubscribeFrom(Topic topicFilter) {
LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter);
LOG.trace("Removing subscription. CId={}, topics={}", clientID, topicFilter);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
for (Subscription sub : this.subscriptions) {
if (sub.getTopicFilter().equals(topicFilter)) {
@@ -112,14 +112,13 @@ public class TransientSession extends ClientSession {
@Override
public void dropQueue() {
LOG.debug("Removing messages of session. CId={}", this.clientID);
this.messagesQueue = null;
LOG.debug("Messages of the session have been removed. CId={}", this.clientID);
LOG.trace("Messages of the session have been removed. CId={}", this.clientID);
}
@Override
public void cleanSession() {
LOG.info("Transient sessiom, wiping existing subscriptions. ClientId={}", clientID);
LOG.trace("Transient session, wiping existing subscriptions. ClientId={}", clientID);
this.subscriptions.clear();
//TODO
@@ -134,7 +133,7 @@ public class TransientSession extends ClientSession {
@Override
public int inFlightAckWaiting(StoredMessage msg) {
LOG.debug("Adding message to inflight zone. CId={}", clientID);
LOG.trace("Adding message to inflight zone. CId={}", clientID);
int messageId = this.nextPacketId();
this.outboundInflightMap.put(messageId, msg);
return messageId;
@@ -142,7 +141,7 @@ public class TransientSession extends ClientSession {
@Override
public IMessagesStore.StoredMessage inFlightAcknowledged(int messageID) {
LOG.debug("Removing message to inflight zone. CId={}, messageID={}", clientID, messageID);
LOG.trace("Removing message to inflight zone. CId={}, messageID={}", clientID, messageID);
return this.outboundInflightMap.remove(messageID);
}
@@ -164,7 +163,7 @@ public class TransientSession extends ClientSession {
@Override
public IMessagesStore.StoredMessage completeReleasedPublish(int messageID) {
LOG.info("Acknowledged message in second phase, clientID <{}> messageID {}", clientID, messageID);
LOG.trace("Acknowledged message in second phase, clientID <{}> messageID {}", clientID, messageID);
return this.secondPhaseStore.remove(messageID);
}
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+1
View File
@@ -88,6 +88,7 @@ final class BrokerInterceptor implements Interceptor {
LOG.warn("Forcing shutdown of interceptor thread pool...");
executor.shutdownNow();
}
LOG.info("interceptors stopped");
}
@Override
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+34 -32
View File
@@ -200,7 +200,7 @@ public class ProtocolProcessor {
public void processConnect(Channel channel, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientId = payload.clientIdentifier();
LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName());
LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, payload.userName());
if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel()
&& msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) {
@@ -268,7 +268,7 @@ public class ProtocolProcessor {
channel.close().addListener(CLOSE_ON_FAILURE);
}
LOG.info("CONNECT message processed CId={}, username={}", clientId, payload.userName());
LOG.info("Connected client <{}> with login <{}>", clientId, payload.userName());
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
@@ -312,8 +312,8 @@ public class ProtocolProcessor {
return true;
}
private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) {
LOG.info("Sending CONNACK. CId={}", clientId);
private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, String clientId) {
LOG.debug("Sending CONNACK. CId={}", clientId);
final boolean success = descriptor.assignState(DISCONNECTED, SENDACK);
if (!success) {
return false;
@@ -330,7 +330,7 @@ public class ProtocolProcessor {
}
descriptor.writeAndFlush(okResp);
LOG.info("CONNACK has been sent. CId={}", clientId);
LOG.debug("CONNACK has been sent. CId={}", clientId);
if (isSessionAlreadyStored && msgCleanSessionFlag) {
for (Subscription existingSub : clientSession.getSubscriptions()) {
@@ -340,24 +340,23 @@ public class ProtocolProcessor {
return true;
}
private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, final String clientId) {
private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) {
int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
LOG.info("Configuring connection. CId={}", clientId);
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("The connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}",
LOG.debug("Connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}",
clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
}
private void storeWillMessage(MqttConnectMessage msg, final String clientId) {
private void storeWillMessage(MqttConnectMessage msg, String clientId) {
// Handle will flag
if (msg.variableHeader().isWillFlag()) {
MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos());
LOG.info("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}",
LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}",
clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain());
byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8);
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
@@ -365,7 +364,7 @@ public class ProtocolProcessor {
WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(),
willQos);
m_willStore.put(clientId, will);
LOG.info("MQTT last will and testament has been configured. CId={}", clientId);
LOG.debug("MQTT last will and testament has been configured. CId={}", clientId);
}
}
@@ -556,9 +555,9 @@ public class ProtocolProcessor {
m_interceptor.notifyMessageAcknowledged(interceptAckMsg);
}
public void processDisconnect(Channel channel) throws InterruptedException {
public void processDisconnect(Channel channel) {
final String clientID = NettyUtils.clientID(channel);
LOG.info("Processing DISCONNECT message. CId={}", clientID);
LOG.debug("Processing DISCONNECT message. CId={}", clientID);
channel.flush();
final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID);
if (existingDescriptor == null) {
@@ -593,7 +592,7 @@ public class ProtocolProcessor {
}
if (!existingDescriptor.close()) {
LOG.info("Connection has been closed. CId={}", clientID);
LOG.debug("Connection has been closed. CId={}", clientID);
return;
}
@@ -605,7 +604,7 @@ public class ProtocolProcessor {
}
this.sessionsRepository.disconnect(clientID);
LOG.info("DISCONNECT message has been processed. CId={}", clientID);
LOG.info("Client <{}> disconnected", clientID);
}
private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) {
@@ -615,10 +614,10 @@ public class ProtocolProcessor {
}
if (descriptor.cleanSession) {
LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID);
LOG.trace("Removing saved subscriptions. CId={}", descriptor.clientID);
final ClientSession session = this.sessionsRepository.sessionForClient(clientID);
session.wipeSubscriptions();
LOG.info("Saved subscriptions have been removed. CId={}", descriptor.clientID);
LOG.trace("Saved subscriptions have been removed. CId={}", descriptor.clientID);
}
return true;
}
@@ -645,7 +644,7 @@ public class ProtocolProcessor {
return false;
}
LOG.info("Removing will message. ClientId={}", descriptor.clientID);
LOG.trace("Removing will message. ClientId={}", descriptor.clientID);
// cleanup the will store
m_willStore.remove(clientID);
String username = descriptor.getUsername();
@@ -654,7 +653,7 @@ public class ProtocolProcessor {
}
public void processConnectionLost(String clientID, Channel channel) {
LOG.info("Processing connection lost event. CId={}", clientID);
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
@@ -681,7 +680,7 @@ public class ProtocolProcessor {
List<String> topics = msg.payload().topics();
String clientID = NettyUtils.clientID(channel);
LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics);
LOG.debug("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics);
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
for (String t : topics) {
@@ -694,7 +693,7 @@ public class ProtocolProcessor {
return;
}
LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic);
LOG.trace("Removing subscription. CId={}, topic={}", clientID, topic);
subscriptions.removeSubscription(topic, clientID);
clientSession.unsubscribeFrom(topic);
String username = NettyUtils.userName(channel);
@@ -706,14 +705,15 @@ public class ProtocolProcessor {
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0);
MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID));
LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, 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.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID);
LOG.debug("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID);
RunningSubscription executionKey = new RunningSubscription(clientID, messageID);
SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED);
@@ -731,7 +731,7 @@ public class ProtocolProcessor {
return;
}
LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics);
LOG.debug("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics);
List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID);
@@ -740,7 +740,7 @@ public class ProtocolProcessor {
subscriptions.add(subscription);
}
LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID);
LOG.debug("Sending SUBACK response CId={}, messageId={}", clientID, messageID);
channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE);
// fire the persisted messages in session
@@ -751,6 +751,8 @@ public class ProtocolProcessor {
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);
}
}
@@ -763,8 +765,8 @@ public class ProtocolProcessor {
if (req.qualityOfService() == FAILURE) {
continue;
}
Subscription newSubscription = new Subscription(clientID, new Topic(req.topicName()),
req.qualityOfService());
final Topic topic = new Topic(req.topicName());
Subscription newSubscription = new Subscription(clientID, topic, req.qualityOfService());
clientSession.subscribe(newSubscription);
newSubscriptions.add(newSubscription);
@@ -790,17 +792,17 @@ public class ProtocolProcessor {
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.error("Client does not have read permissions on the topic CId={}, username={}, messageId={}, " +
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.info("Client will be subscribed to the topic CId={}, username={}, messageId={}, topic={}",
LOG.debug("Client will be subscribed to the topic CId={}, username={}, messageId={}, topic={}",
clientID, username, messageId, topic);
qos = req.qualityOfService();
} else {
LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", clientID,
LOG.warn("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", clientID,
username, messageId, topic);
qos = FAILURE;
}
@@ -824,8 +826,8 @@ public class ProtocolProcessor {
return new MqttSubAckMessage(fixedHeader, from(messageId), payload);
}
private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) {
LOG.info("Retrieving retained messages CId={}, topics={}", newSubscription.getClientId(),
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