From 99f06122091f576feb23d9723af57a1febb8e1e2 Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 27 Apr 2018 15:56:26 +0200 Subject: [PATCH] Minor, fixed logging level to be more usefull for distribution --- .../moquette/server/ConnectionDescriptor.java | 17 ++--- .../main/java/io/moquette/server/Server.java | 19 +++--- .../moquette/server/netty/NettyAcceptor.java | 43 +++++------- .../io/moquette/spi/TransientSession.java | 19 +++--- .../moquette/spi/impl/BrokerInterceptor.java | 1 + .../moquette/spi/impl/ProtocolProcessor.java | 66 ++++++++++--------- .../src/main/resources/log4j.properties | 2 + distribution/src/main/scripts/moquette.sh | 2 + 8 files changed, 80 insertions(+), 89 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 40ad54c8..755cddb0 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 5ecb4bf8..54dc21df 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -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 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 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; } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index ad132f24..c829858b 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -110,9 +110,8 @@ public class NettyAcceptor implements ServerAcceptor { private Class 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() { @@ -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) { diff --git a/broker/src/main/java/io/moquette/spi/TransientSession.java b/broker/src/main/java/io/moquette/spi/TransientSession.java index 288ae9d6..ff4d1018 100644 --- a/broker/src/main/java/io/moquette/spi/TransientSession.java +++ b/broker/src/main/java/io/moquette/spi/TransientSession.java @@ -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 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); } diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index c59ad4d2..57f98261 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -88,6 +88,7 @@ final class BrokerInterceptor implements Interceptor { LOG.warn("Forcing shutdown of interceptor thread pool..."); executor.shutdownNow(); } + LOG.info("interceptors stopped"); } @Override diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 4088ccd3..63aac13e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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 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 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 diff --git a/distribution/src/main/resources/log4j.properties b/distribution/src/main/resources/log4j.properties index 96d40879..f587da28 100644 --- a/distribution/src/main/resources/log4j.properties +++ b/distribution/src/main/resources/log4j.properties @@ -19,6 +19,7 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender #log4j.appender.stdout.Threshold=DEBUG log4j.appender.stdout.Threshold=TRACE log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +#log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n log4j.appender.stdout.layout.ConversionPattern=%d{dd/MM/yyyy HH:mm:ss,SSS} [%t] %-5p %c{1} %M %L %x - %m%n #file appender @@ -28,6 +29,7 @@ log4j.appender.file.File=moquette.log log4j.appender.file.MaxFileSize=1MB log4j.appender.file.MaxBackupIndex=1 log4j.appender.file.layout=org.apache.log4j.PatternLayout +#log4j.appender.file.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n log4j.appender.file.layout.ConversionPattern=%d{dd/MM/yyyy HH:mm:ss,SSS} [%t] %-5p %c{1} %M %L %x - %m%n diff --git a/distribution/src/main/scripts/moquette.sh b/distribution/src/main/scripts/moquette.sh index 7431e11d..3391cbaa 100644 --- a/distribution/src/main/scripts/moquette.sh +++ b/distribution/src/main/scripts/moquette.sh @@ -13,6 +13,7 @@ echo " \_| |_/\___/ \__, |\__,_|\___|\__|\__\___| \_| |_/\_/\_\ \_/ \_/ " echo " | | " echo " |_| " echo " " +echo " version: 0.11-SNAPSHOT " cd "$(dirname "$0")" @@ -88,6 +89,7 @@ JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCApplicationStoppedTime" JAVA_OPTS="$JAVA_OPTS -XX:+PrintPromotionFailure" #JAVA_OPTS="$JAVA_OPTS -XX:PrintFLSStatistics=1" #JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/moquette/gc.log" +JAVA_OPTS="$JAVA_OPTS -Xloggc:$MOQUETTE_HOME/gc.log" JAVA_OPTS="$JAVA_OPTS -XX:+UseGCLogFileRotation" JAVA_OPTS="$JAVA_OPTS -XX:NumberOfGCLogFiles=10" JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M"