From 033a428184ce1e8a134fda082c04346ac6858c61 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 25 Apr 2018 19:39:12 +0200 Subject: [PATCH] Fixed checkStyle and ErrorProne warns --- .../HazelcastInterceptHandler.java | 4 +- .../messages/InterceptConnectMessage.java | 6 +- .../persistence/MemoryMessagesStore.java | 2 +- .../persistence/MemorySessionStore.java | 3 +- .../moquette/server/ConnectionDescriptor.java | 10 +- .../config/ClasspathResourceLoader.java | 4 +- .../server/config/ConfigurationParser.java | 13 +- .../server/config/FileResourceLoader.java | 9 +- .../netty/MoquetteIdleTimeoutHandler.java | 4 +- .../moquette/server/netty/NettyAcceptor.java | 8 +- .../server/netty/NettyMQTTHandler.java | 14 ++- .../netty/metrics/BytesMetricsHandler.java | 4 +- .../netty/metrics/MQTTMessageLogger.java | 3 +- .../netty/metrics/MessageMetricsHandler.java | 4 +- .../java/io/moquette/spi/DurableSession.java | 16 --- .../io/moquette/spi/ISubscriptionsStore.java | 16 +++ .../io/moquette/spi/InboundFlightZone.java | 32 +++++ .../io/moquette/spi/TransientSession.java | 4 +- .../java/io/moquette/spi/impl/DebugUtils.java | 4 +- .../moquette/spi/impl/MessagesPublisher.java | 10 +- .../impl/PersistentQueueMessageSender.java | 8 +- .../moquette/spi/impl/ProtocolProcessor.java | 115 +++++++----------- .../impl/ProtocolProcessorBootstrapper.java | 14 ++- .../spi/impl/RunningSubscription.java | 48 ++++++++ .../moquette/spi/impl/SessionsRepository.java | 4 +- .../spi/impl/security/ACLFileParser.java | 16 ++- .../impl/security/AcceptAllAuthenticator.java | 1 + .../spi/impl/subscriptions/CNode.java | 18 +-- .../CTrieSubscriptionDirectory.java | 85 ++----------- .../impl/subscriptions/DumpTreeVisitor.java | 67 ++++++++++ .../SubscriptionCounterVisitor.java | 33 +++++ .../spi/impl/subscriptions/Token.java | 4 +- .../spi/impl/subscriptions/Topic.java | 14 +-- broker/src/test/java/io/moquette/Utils.java | 8 +- .../moquette/persistence/MessageStoreTCK.java | 29 ++++- .../io/moquette/server/MessageCollector.java | 2 +- .../server/ServerIntegrationFuseTest.java | 4 +- ...rationHazelcastHandlerInterceptorTest.java | 7 +- .../server/ServerIntegrationPahoTest.java | 20 +-- .../ServerIntegrationQoSValidationTest.java | 20 +-- .../server/ServerIntegrationRestartTest.java | 14 ++- .../server/ServerIntegrationSSLTest.java | 3 +- .../ServerIntegrationWebSocketTest.java | 1 + ...erverLowlevelMessagesIntegrationTests.java | 5 +- .../io/moquette/spi/ClientSessionTest.java | 9 +- .../AbstractProtocolProcessorCommonUtils.java | 32 ++--- .../spi/impl/BrokerInterceptorTest.java | 3 +- .../moquette/spi/impl/MockAuthenticator.java | 5 +- .../spi/impl/ProtocolProcessorTest.java | 6 +- .../impl/ProtocolProcessor_CONNECT_Test.java | 7 +- ...rotocolProcessor_VariousUseCases_Test.java | 4 +- .../impl/security/DBAuthenticatorTest.java | 8 +- .../impl/security/FileAuthenticatorTest.java | 8 +- ...TrieSubscriptionDirectoryMatchingTest.java | 9 +- .../CTrieSubscriptionDirectoryTest.java | 21 ++-- .../spi/impl/subscriptions/TopicTest.java | 1 - .../java/io/moquette/testclient/Client.java | 10 +- .../testclient/ClientNettyMQTTHandler.java | 6 +- .../io/moquette/testclient/RawClient.java | 14 ++- checkstyle.xml | 4 +- distribution/build.gradle | 20 ++- .../testembedded/EmbeddedLauncher.java | 8 +- .../persistence/h2/H2PersistentQueue.java | 15 +++ .../persistence/h2/H2PersistentStore.java | 16 +-- .../persistence/h2/H2SessionsStore.java | 28 +++-- .../persistence/h2/H2PersistentQueueTest.java | 19 ++- .../persistence/h2/H2SessionsStoreTest.java | 3 +- .../persistence/mapdb/MapDBMessagesStore.java | 2 +- .../mapdb/MapDBPersistentStore.java | 1 + .../persistence/mapdb/MapDBSessionsStore.java | 14 ++- .../mapdb/MapDBPersistentStoreTest.java | 4 +- .../mapdb/MapDBSessionsStoreTest.java | 3 +- .../netty/performance/BenchmarkPublisher.java | 23 +++- .../performance/BenchmarkSubscriber.java | 15 ++- .../netty/performance/LoopMQTTHandler.java | 17 +-- .../NettyPublishReceiverHandler.java | 9 +- .../performance/ProtocolDecodingServer.java | 15 ++- .../ProtocolPublishDecodingServer.java | 11 +- .../netty/performance/PublishBomber.java | 14 ++- .../performance/PublishReceiverHandler.java | 7 +- .../integration/bunch_publish_issue12.groovy | 3 +- 81 files changed, 680 insertions(+), 424 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/InboundFlightZone.java create mode 100644 broker/src/main/java/io/moquette/spi/impl/RunningSubscription.java create mode 100644 broker/src/main/java/io/moquette/spi/impl/subscriptions/DumpTreeVisitor.java create mode 100644 broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionCounterVisitor.java diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java index b5f9155e..d59c78b7 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java @@ -24,6 +24,7 @@ import io.netty.buffer.ByteBuf; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import static io.moquette.spi.impl.Utils.readBytesAndRewind; +import static java.nio.charset.StandardCharsets.UTF_8; public class HazelcastInterceptHandler extends AbstractInterceptHandler { @@ -45,7 +46,8 @@ public class HazelcastInterceptHandler extends AbstractInterceptHandler { ByteBuf payload = msg.getPayload(); byte[] payloadContent = readBytesAndRewind(payload); - LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(payloadContent)); + LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), + new String(payloadContent, UTF_8)); ITopic topic = hz.getTopic("moquette"); HazelcastMsg hazelcastMsg = new HazelcastMsg(msg); topic.publish(hazelcastMsg); diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptConnectMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptConnectMessage.java index 07041b1c..2ac6e2f9 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptConnectMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptConnectMessage.java @@ -18,6 +18,8 @@ package io.moquette.interception.messages; import io.netty.handler.codec.mqtt.MqttConnectMessage; +import java.nio.charset.StandardCharsets; + public class InterceptConnectMessage extends InterceptAbstractMessage { private final MqttConnectMessage msg; @@ -72,7 +74,7 @@ public class InterceptConnectMessage extends InterceptAbstractMessage { } public byte[] getPassword() { - return msg.payload().password().getBytes(); + return msg.payload().password().getBytes(StandardCharsets.UTF_8); } public String getWillTopic() { @@ -80,6 +82,6 @@ public class InterceptConnectMessage extends InterceptAbstractMessage { } public byte[] getWillMessage() { - return msg.payload().willMessage().getBytes(); + return msg.payload().willMessage().getBytes(StandardCharsets.UTF_8); } } diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 1d3ef0fb..cf9838fb 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -41,7 +41,7 @@ public class MemoryMessagesStore implements IMessagesStore { public void storeRetained(Topic topic, StoredMessage storedMessage) { LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); if (storedMessage.getClientID() == null) { - throw new IllegalArgumentException( "Message to be persisted must have a not null client ID"); + throw new IllegalArgumentException("Message to be persisted must have a not null client ID"); } m_retainedStore.put(topic, storedMessage); } diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 05c0b9e0..c09a7651 100644 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -33,7 +33,7 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore { private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); - class Session { + static class Session { final String clientID; final Map subscriptions = new ConcurrentHashMap<>(); boolean cleanSession; @@ -170,7 +170,6 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore { return subscriptions.get(subcription.getTopicFilter()); } - @Override public StoredMessage inFlightAck(String clientID, int messageID) { return getSession(clientID).outboundFlightMessages.remove(messageID); diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 33b20328..40ad54c8 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -29,6 +29,10 @@ import io.moquette.server.netty.metrics.MessageMetrics; import io.moquette.server.netty.metrics.MessageMetricsHandler; import io.netty.channel.Channel; +import static io.netty.channel.ChannelFutureListener.CLOSE; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; +import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE; + /** * Value object to maintain the information of single connection, like ClientID, Channel, and clean * session flag. @@ -62,7 +66,7 @@ public class ConnectionDescriptor { } public void writeAndFlush(Object payload) { - this.channel.writeAndFlush(payload); + this.channel.writeAndFlush(payload).addListener(FIRE_EXCEPTION_ON_FAILURE); } public void setupAutoFlusher(int flushIntervalMs) { @@ -88,7 +92,7 @@ public class ConnectionDescriptor { if (!success) { return false; } - this.channel.close(); + this.channel.close().addListener(CLOSE_ON_FAILURE); return true; } @@ -100,7 +104,7 @@ public class ConnectionDescriptor { LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID); // try { // this.channel.disconnect().sync(); - this.channel.close(); // .sync(); + this.channel.close().addListener(CLOSE_ON_FAILURE); // .sync(); // } catch (InterruptedException e) { // e.printStackTrace(); // } diff --git a/broker/src/main/java/io/moquette/server/config/ClasspathResourceLoader.java b/broker/src/main/java/io/moquette/server/config/ClasspathResourceLoader.java index b8582c4f..c88dc78e 100644 --- a/broker/src/main/java/io/moquette/server/config/ClasspathResourceLoader.java +++ b/broker/src/main/java/io/moquette/server/config/ClasspathResourceLoader.java @@ -19,6 +19,8 @@ package io.moquette.server.config; import java.io.InputStream; import java.io.InputStreamReader; import java.io.Reader; +import java.nio.charset.StandardCharsets; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,7 +53,7 @@ public class ClasspathResourceLoader implements IResourceLoader { public Reader loadResource(String relativePath) { LOG.info("Loading resource. RelativePath = {}.", relativePath); InputStream is = this.classLoader.getResourceAsStream(relativePath); - return is != null ? new InputStreamReader(is) : null; + return is != null ? new InputStreamReader(is, StandardCharsets.UTF_8) : null; } @Override diff --git a/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java b/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java index a91802c3..ec6ec554 100644 --- a/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java +++ b/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java @@ -24,6 +24,8 @@ import java.io.FileNotFoundException; import java.io.FileReader; import java.io.IOException; import java.io.Reader; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.text.ParseException; import java.util.Properties; @@ -55,15 +57,10 @@ class ConfigurationParser { return; } try { - FileReader reader = new FileReader(file); + Reader reader = Files.newBufferedReader(file.toPath(), StandardCharsets.UTF_8); parse(reader); - } catch (FileNotFoundException fex) { - LOG.warn( - String.format( - "parsing not existing file %s, so fallback on default configuration!", - file.getAbsolutePath()), - fex); - return; + } catch (IOException fex) { + LOG.warn("parsing not existing file {}, fallback on default configuration!", file.getAbsolutePath(), fex); } } diff --git a/broker/src/main/java/io/moquette/server/config/FileResourceLoader.java b/broker/src/main/java/io/moquette/server/config/FileResourceLoader.java index 4c27dada..bcc9bb0a 100644 --- a/broker/src/main/java/io/moquette/server/config/FileResourceLoader.java +++ b/broker/src/main/java/io/moquette/server/config/FileResourceLoader.java @@ -19,10 +19,15 @@ package io.moquette.server.config; import java.io.File; import java.io.FileNotFoundException; import java.io.FileReader; +import java.io.IOException; import java.io.Reader; +import java.nio.file.Files; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static java.nio.charset.StandardCharsets.UTF_8; + public class FileResourceLoader implements IResourceLoader { private static final Logger LOG = LoggerFactory.getLogger(FileResourceLoader.class); @@ -68,8 +73,8 @@ public class FileResourceLoader implements IResourceLoader { throw new ResourceIsDirectoryException("File \"" + f + "\" is a directory!"); } try { - return new FileReader(f); - } catch (FileNotFoundException e) { + return Files.newBufferedReader(f.toPath(), UTF_8); + } catch (IOException e) { LOG.error("The file does not exist. Path = {}.", f.getAbsolutePath()); return null; } diff --git a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java index 63d67151..a08faba5 100644 --- a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java @@ -24,6 +24,8 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.handler.timeout.IdleState; import io.netty.handler.timeout.IdleStateEvent; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + @Sharable class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { @@ -37,7 +39,7 @@ class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { LOG.info("Firing channel inactive event. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); // fire a channelInactive to trigger publish of Will ctx.fireChannelInactive(); - ctx.close(); + ctx.close().addListener(CLOSE_ON_FAILURE); } } else { if (LOG.isDebugEnabled()) { 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 4db0de12..96f8c66d 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -44,6 +44,7 @@ import io.netty.handler.codec.mqtt.MqttEncoder; import io.netty.handler.ssl.SslHandler; import io.netty.handler.timeout.IdleStateHandler; import io.netty.util.concurrent.Future; +import io.netty.util.concurrent.GenericFutureListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.net.ssl.SSLContext; @@ -53,6 +54,7 @@ import java.util.List; import java.util.Optional; import java.util.concurrent.TimeUnit; import static io.moquette.BrokerConstants.*; +import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE; public class NettyAcceptor implements ServerAcceptor { @@ -85,7 +87,7 @@ public class NettyAcceptor implements ServerAcceptor { } } - abstract class PipelineInitializer { + abstract static class PipelineInitializer { abstract void init(ChannelPipeline pipeline) throws Exception; } @@ -197,7 +199,7 @@ public class NettyAcceptor implements ServerAcceptor { // Bind and start to accept incoming connections. ChannelFuture f = b.bind(host, port); LOG.info("Server has been bound. host={}, port={}", host, port); - f.sync(); + f.sync().addListener(FIRE_EXCEPTION_ON_FAILURE); } catch (InterruptedException ex) { LOG.error("An interruptedException was caught while initializing server. Protocol={}", protocol, ex); } @@ -349,6 +351,8 @@ public class NettyAcceptor implements ServerAcceptor { }); } + @SuppressWarnings("FutureReturnValueIgnored") + @Override public void close() { LOG.info("Closing Netty acceptor..."); if (m_workerGroup == null || m_bossGroup == null) { diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 28419b8d..43d97857 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -24,6 +24,8 @@ import io.netty.handler.codec.mqtt.*; import io.netty.util.ReferenceCountUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; @Sharable @@ -78,16 +80,16 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { false, 0); MqttMessage pingResp = new MqttMessage(pingHeader); - ctx.writeAndFlush(pingResp); + ctx.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE); break; default: LOG.error("Unkonwn MessageType:{}", messageType); break; } } catch (Throwable ex) { - LOG.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex); + LOG.error("Exception was caught while processing MQTT message, {}", ex.getCause(), ex); ctx.fireExceptionCaught(ex); - ctx.close(); + ctx.close().addListener(CLOSE_ON_FAILURE); } finally { ReferenceCountUtil.release(msg); } @@ -97,17 +99,17 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { public void channelInactive(ChannelHandlerContext ctx) throws Exception { String clientID = NettyUtils.clientID(ctx.channel()); if (clientID != null && !clientID.isEmpty()) { - LOG.info("Notifying connection lost event. MqttClientId = {}.", clientID); + LOG.info("Notifying connection lost event. MqttClientId = {}", clientID); m_processor.processConnectionLost(clientID, ctx.channel()); } - ctx.close(); + ctx.close().addListener(CLOSE_ON_FAILURE); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { LOG.error("An unexpected exception was caught while processing MQTT message. Closing Netty channel. CId={}, " + "cause={}, errorMessage={}", NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage()); - ctx.close(); + ctx.close().addListener(CLOSE_ON_FAILURE); } @Override diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java index d93c0e17..8f426a95 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java @@ -24,6 +24,8 @@ import io.netty.channel.ChannelPromise; import io.netty.util.Attribute; import io.netty.util.AttributeKey; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + public class BytesMetricsHandler extends ChannelDuplexHandler { private static final AttributeKey ATTR_KEY_METRICS = AttributeKey.valueOf("BytesMetrics"); @@ -53,7 +55,7 @@ public class BytesMetricsHandler extends ChannelDuplexHandler { public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get(); metrics.incrementWrote(((ByteBuf) msg).writableBytes()); - ctx.write(msg, promise); + ctx.write(msg, promise).addListener(CLOSE_ON_FAILURE); } @Override diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java index 791a681b..3564e84c 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java @@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory; import java.util.List; import static io.moquette.spi.impl.Utils.messageId; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; /** * @@ -100,6 +101,6 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { logMQTTMessage(ctx, msg, "C<-B"); - ctx.write(msg, promise); + ctx.write(msg, promise).addListener(CLOSE_ON_FAILURE); } } diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java index c4894d12..9e7c0212 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java @@ -23,6 +23,8 @@ import io.netty.channel.ChannelPromise; import io.netty.util.Attribute; import io.netty.util.AttributeKey; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + public class MessageMetricsHandler extends ChannelDuplexHandler { private static final AttributeKey ATTR_KEY_METRICS = AttributeKey.valueOf("MessageMetrics"); @@ -52,7 +54,7 @@ public class MessageMetricsHandler extends ChannelDuplexHandler { public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get(); metrics.incrementWrote(1); - ctx.write(msg, promise); + ctx.write(msg, promise).addListener(CLOSE_ON_FAILURE); } @Override diff --git a/broker/src/main/java/io/moquette/spi/DurableSession.java b/broker/src/main/java/io/moquette/spi/DurableSession.java index fe1558be..60d8b74e 100644 --- a/broker/src/main/java/io/moquette/spi/DurableSession.java +++ b/broker/src/main/java/io/moquette/spi/DurableSession.java @@ -22,9 +22,7 @@ import org.slf4j.LoggerFactory; import java.util.Collection; import java.util.HashSet; -import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; public class DurableSession extends ClientSession { @@ -51,20 +49,6 @@ public class DurableSession extends ClientSession { } } - class InboundFlightZone { - - final Map inboundFlightMessages = new ConcurrentHashMap<>(); - - IMessagesStore.StoredMessage lookup(int messageID) { - return inboundFlightMessages.get(messageID); - } - - void waitingRel(int messageID, IMessagesStore.StoredMessage msg) { - inboundFlightMessages.put(messageID, msg); - } - } - - private static final Logger LOG = LoggerFactory.getLogger(DurableSession.class); private final ISessionsStore sessionsStore; diff --git a/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java index 5dc0597f..9bab4e25 100644 --- a/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java @@ -1,3 +1,19 @@ +/* + * Copyright (c) 2012-2017 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; import io.moquette.spi.impl.subscriptions.Subscription; diff --git a/broker/src/main/java/io/moquette/spi/InboundFlightZone.java b/broker/src/main/java/io/moquette/spi/InboundFlightZone.java new file mode 100644 index 00000000..f63eeae4 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/InboundFlightZone.java @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2012-2017 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; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +class InboundFlightZone { + + private final Map inboundFlightMessages = new ConcurrentHashMap<>(); + + IMessagesStore.StoredMessage lookup(int messageID) { + return inboundFlightMessages.get(messageID); + } + + void waitingRel(int messageID, IMessagesStore.StoredMessage msg) { + inboundFlightMessages.put(messageID, msg); + } +} diff --git a/broker/src/main/java/io/moquette/spi/TransientSession.java b/broker/src/main/java/io/moquette/spi/TransientSession.java index 9a064c2c..de83035e 100644 --- a/broker/src/main/java/io/moquette/spi/TransientSession.java +++ b/broker/src/main/java/io/moquette/spi/TransientSession.java @@ -30,7 +30,7 @@ public class TransientSession extends ClientSession { private static final Logger LOG = LoggerFactory.getLogger(TransientSession.class); - private Queue messagesQueue = new LinkedList<>(); + private Queue messagesQueue = new ArrayDeque<>(); private final AtomicInteger packetGenerator = new AtomicInteger(1); @@ -43,7 +43,7 @@ public class TransientSession extends ClientSession { } @Override - public void disconnect() {} + public void disconnect() { } @Override public boolean isCleanSession() { diff --git a/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java b/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java index 25d07214..ba8be7a8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java +++ b/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java @@ -18,10 +18,12 @@ package io.moquette.spi.impl; import io.netty.buffer.ByteBuf; +import java.nio.charset.StandardCharsets; + final class DebugUtils { static String payload2Str(ByteBuf content) { - return new String(content.copy().array()); + return new String(content.copy().array(), StandardCharsets.UTF_8); } private DebugUtils() { diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 15721d11..efaf2ec1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -62,11 +62,12 @@ class MessagesPublisher { 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()), + "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); + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", + pubMsg.getClientID(), topic, messageID); } publish2Subscribers(pubMsg, topic); } @@ -82,7 +83,8 @@ class MessagesPublisher { 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. + // 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); diff --git a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java index 5f0c2ecc..f469d52f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -50,12 +50,12 @@ class PersistentQueueMessageSender { 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); + 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); + LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, " + + "topic={}, qos={}, removeTemporaryQoS2={}", messageId, clientId, topicName, qos, true); } } } 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 7b71cbb3..cc5e76d1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -37,6 +37,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -45,6 +46,8 @@ import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; 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.MqttConnectReturnCode.*; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.*; @@ -92,37 +95,6 @@ public class ProtocolProcessor { STORED, VERIFIED } - private 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; - } - } - private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); private ConnectionDescriptorStore connectionDescriptors; @@ -157,15 +129,15 @@ public class ProtocolProcessor { public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository) { - init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, - interceptor, sessionsRepository); + init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, + authorizator, interceptor, sessionsRepository); } public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository) { - init(new ConnectionDescriptorStore(sessionsRepository), subscriptions, storageService, sessionsStore, authenticator, - allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository); + init(new ConnectionDescriptorStore(sessionsRepository), subscriptions, storageService, sessionsStore, + authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository); } /** @@ -235,8 +207,8 @@ public class ProtocolProcessor { MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION); LOG.error("MQTT protocol version is not valid. CId={}", clientId); - channel.writeAndFlush(badProto); - channel.close(); + channel.writeAndFlush(badProto).addListener(FIRE_EXCEPTION_ON_FAILURE); + channel.close().addListener(CLOSE_ON_FAILURE); return; } @@ -245,8 +217,8 @@ public class ProtocolProcessor { if (!cleanSession || !this.allowZeroByteClientId) { MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED); - channel.writeAndFlush(badId); - channel.close(); + channel.writeAndFlush(badId).addListener(FIRE_EXCEPTION_ON_FAILURE); + channel.close().addListener(CLOSE_ON_FAILURE); LOG.error("The MQTT client ID cannot be empty. Username={}", payload.userName()); return; } @@ -258,7 +230,7 @@ public class ProtocolProcessor { } if (!login(channel, msg, clientId)) { - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); return; } @@ -275,25 +247,25 @@ public class ProtocolProcessor { initializeKeepAliveTimeout(channel, msg, clientId); storeWillMessage(msg, clientId); if (!sendAck(descriptor, msg, clientId)) { - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); return; } m_interceptor.notifyClientConnected(msg); if (!descriptor.assignState(SENDACK, SESSION_CREATED)) { - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); return; } final ClientSession clientSession = this.sessionsRepository.createOrLoadClientSession(clientId, cleanSession); if (!republish(descriptor, msg, clientSession)) { - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); return; } final boolean success = descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED); if (!success) { - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); } LOG.info("CONNECT message processed CId={}, username={}", clientId, payload.userName()); @@ -319,7 +291,7 @@ public class ProtocolProcessor { if (msg.variableHeader().hasUserName()) { byte[] pwd = null; if (msg.variableHeader().hasPassword()) { - pwd = msg.payload().password().getBytes(); + 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); @@ -386,12 +358,12 @@ public class ProtocolProcessor { if (msg.variableHeader().isWillFlag()) { MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos()); LOG.info("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}", - clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain()); - byte[] willPayload = msg.payload().willMessage().getBytes(); + clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain()); + byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); // save the will testament in the clientID store WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), - willQos); + willQos); m_willStore.put(clientId, will); LOG.info("MQTT last will and testament has been configured. CId={}", clientId); } @@ -414,7 +386,8 @@ public class ProtocolProcessor { } private void failedCredentials(Channel session) { - session.writeAndFlush(connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)); + 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); } @@ -435,7 +408,8 @@ public class ProtocolProcessor { StoredMessage inflightMsg = targetSession.inFlightAcknowledged(messageID); String topic = inflightMsg.getTopic(); - InterceptAcknowledgedMessage wrapped = new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID); + InterceptAcknowledgedMessage wrapped = new InterceptAcknowledgedMessage(inflightMsg, topic, username, + messageID); m_interceptor.notifyMessageAcknowledged(wrapped); } @@ -517,19 +491,19 @@ public class ProtocolProcessor { /** * 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); + 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); - } + //Stores retained message to the topic + if (will.isRetained()) { + m_messagesStore.storeRetained(topic, tobeStored); + } } static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) { @@ -565,7 +539,7 @@ public class ProtocolProcessor { MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0); MqttMessage pubRelMessage = new MqttMessage(pubRelHeader, from(messageID)); - channel.writeAndFlush(pubRelMessage); + channel.writeAndFlush(pubRelMessage).addListener(FIRE_EXCEPTION_ON_FAILURE); } public void processPubComp(Channel channel, MqttMessage msg) { @@ -589,7 +563,7 @@ public class ProtocolProcessor { final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); if (existingDescriptor == null) { // another client with same ID removed the descriptor, we must exit - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); return; } @@ -714,7 +688,7 @@ public class ProtocolProcessor { boolean validTopic = topic.isValid(); if (!validTopic) { // close the connection, not valid topicFilter is a protocol violation - channel.close(); + channel.close().addListener(CLOSE_ON_FAILURE); LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } @@ -732,7 +706,7 @@ public class ProtocolProcessor { MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); - channel.writeAndFlush(ackMessage); + channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE); } public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { @@ -766,7 +740,7 @@ public class ProtocolProcessor { } LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); - channel.writeAndFlush(ackMessage); + channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE); // fire the persisted messages in session for (Subscription subscription : newSubscriptions) { @@ -879,9 +853,10 @@ public class ProtocolProcessor { 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); + 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(); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 577564c0..13d20871 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -222,8 +222,9 @@ public class ProtocolProcessorBootstrapper { .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()); + 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 { @@ -232,10 +233,11 @@ public class ProtocolProcessorBootstrapper { instance = this.getClass().getClassLoader() .loadClass(className) .asSubclass(intrface) - .newInstance(); - } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { - LOG.error("Unable to invoke default constructor. ClassName={}, interfaceName={}, cause={}, errorMessage={}", - className, intrface.getName(), ex.getCause(), ex.getMessage()); + .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; } } diff --git a/broker/src/main/java/io/moquette/spi/impl/RunningSubscription.java b/broker/src/main/java/io/moquette/spi/impl/RunningSubscription.java new file mode 100644 index 00000000..d2697d34 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/RunningSubscription.java @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2012-2017 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; + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java index 04c49c43..5a7666aa 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java +++ b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java @@ -33,7 +33,8 @@ 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. + * Task to be scheduled to execute the cleaning of persisted sessions (clean flag=false) older than a + * defined period. */ private class SessionCleanerTask implements Runnable { @@ -55,6 +56,7 @@ public class SessionsRepository { this.scheduler = scheduler; } + @SuppressWarnings("FutureReturnValueIgnored") public void init() { SessionCleanerTask cleanerTask = new SessionCleanerTask(); this.scheduler.schedule(cleanerTask, 1, TimeUnit.HOURS); diff --git a/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java b/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java index 5081af54..798c327a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java @@ -18,11 +18,18 @@ package io.moquette.spi.impl.security; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.*; + +import java.io.BufferedReader; +import java.io.File; +import java.io.IOException; +import java.io.Reader; +import java.nio.file.Files; import java.text.ParseException; import java.util.regex.Matcher; import java.util.regex.Pattern; +import static java.nio.charset.StandardCharsets.UTF_8; + /** * Parses the acl configuration file. If a line starts with # it's comment. Blank lines are skipped. * The format is "topic [read|write|readwrite] {topic name}" @@ -53,9 +60,9 @@ public final class ACLFileParser { return AuthorizationsCollector.emptyImmutableCollector(); } try { - FileReader reader = new FileReader(file); + Reader reader = Files.newBufferedReader(file.toPath(), UTF_8); return parse(reader); - } catch (FileNotFoundException fex) { + } catch (IOException fex) { LOG.warn( String.format( "parsing not existing file %s, so fallback on default configuration!", @@ -88,7 +95,8 @@ public final class ACLFileParser { Pattern emptyLine = Pattern.compile("^\\s*$"); Pattern commentLine = Pattern.compile("^#.*"); // As spec, comment lines should start with '#' Pattern invalidCommentLine = Pattern.compile("^\\s*#.*"); - Pattern endLineComment = Pattern.compile("^([\\w\\s\\/\\+]+#?)(\\s*#.*)$"); // This pattern has a dependency on filtering `commentLine`. + // This pattern has a dependency on filtering `commentLine`. + Pattern endLineComment = Pattern.compile("^([\\w\\s\\/\\+]+#?)(\\s*#.*)$"); Matcher endLineCommentMatcher; try { diff --git a/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java index 3bc0195a..c9f91820 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java @@ -20,6 +20,7 @@ import io.moquette.spi.security.IAuthenticator; public class AcceptAllAuthenticator implements IAuthenticator { + @Override public boolean checkValid(String clientId, String username, byte[] password) { return true; } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java index 4c722dcc..59be06a2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java @@ -15,10 +15,7 @@ */ package io.moquette.spi.impl.subscriptions; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Set; +import java.util.*; class CNode { @@ -41,7 +38,7 @@ class CNode { boolean anyChildrenMatch(Token token) { for (INode iNode : children) { final CNode child = iNode.mainNode(); - if (child.equals(token)) { + if (child.equalsToken(token)) { return true; } } @@ -55,15 +52,20 @@ class CNode { INode childOf(Token token) { for (INode iNode : children) { final CNode child = iNode.mainNode(); - if (child.equals(token)) { + if (child.equalsToken(token)) { return iNode; } } throw new IllegalArgumentException("Asked for a token that doesn't exists in any child [" + token + "]"); } - private boolean equals(Token token) { - return this.token != null && this.token.equals(token); + private boolean equalsToken(Token token) { + return token != null && this.token != null && this.token.equals(token); + } + + @Override + public int hashCode() { + return Objects.hash(token); } CNode copy() { diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java index 47e3b77f..a7e91062 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java @@ -17,12 +17,10 @@ package io.moquette.spi.impl.subscriptions; import io.moquette.spi.ClientSession; import io.moquette.spi.impl.SessionsRepository; -import io.netty.util.internal.StringUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; -import java.util.concurrent.atomic.AtomicInteger; import static java.lang.String.format; @@ -36,66 +34,18 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { INode root; private volatile SessionsRepository sessionsRepository; - private interface IVisitor { + interface IVisitor { void visit(CNode node, int deep); T getResult(); } - private class DumpTreeVisitor implements IVisitor { - - String s = ""; - - @Override - public void visit(CNode node, int deep) { - String indentTabs = indentTabs(deep); - s += indentTabs + (node.token == null ? "''" : node.token.toString()) + prettySubscriptions(node) + "\n"; - } - - private String prettySubscriptions(CNode node) { - if (node instanceof TNode) { - return "TNode"; - } - if (node.subscriptions.isEmpty()) { - return StringUtil.EMPTY_STRING; - } - StringBuilder subScriptionsStr = new StringBuilder(" ~~["); - int counter = 0; - for (Subscription couple : node.subscriptions) { - subScriptionsStr - .append("{filter=").append(couple.topicFilter).append(", ") - .append("client='").append(couple.clientId).append("'}"); - counter++; - if (counter < node.subscriptions.size()) { - subScriptionsStr.append(";"); - } - } - return subScriptionsStr.append("]").toString(); - } - - private String indentTabs(int deep) { - StringBuilder s = new StringBuilder(); - if (deep > 0) { - s.append(" "); - for (int i = 0; i < deep - 1; i++) { - s.append("| "); - } - s.append("|-"); - } - return s.toString(); - } - - @Override - public String getResult() { - return s; - } - } - private enum Action { OK, REPEAT } + @Override public void init(SessionsRepository sessionsRepository) { LOG.info("Initializing CTrie"); final CNode mainNode = new CNode(); @@ -134,13 +84,11 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { return Optional.of(inode.mainNode()); } - @Override public List matches(Topic topic) { return new ArrayList<>(match(topic)); } - /** * 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 @@ -177,7 +125,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { Set recursiveMatch(Topic topic, INode inode) { CNode cnode = inode.mainNode(); - if (cnode.token == Token.MULTI) { + if (Token.MULTI.equals(cnode.token)) { return cnode.subscriptions; } if (topic.isEmpty()) { @@ -187,10 +135,10 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { return Collections.emptySet(); } final Token token = topic.headToken(); - if (!(cnode.token == Token.SINGLE || cnode.token.equals(token) || cnode.token == ROOT)) { + if (!(Token.SINGLE.equals(cnode.token) || cnode.token.equals(token) || ROOT.equals(cnode.token))) { return Collections.emptySet(); } - Topic remainingTopic = (cnode.token == ROOT) ? topic : topic.exceptHeadToken(); + Topic remainingTopic = (ROOT.equals(cnode.token)) ? topic : topic.exceptHeadToken(); Set subscriptions = new HashSet<>(); if (remainingTopic.isEmpty()) { subscriptions.addAll(cnode.subscriptions); @@ -219,6 +167,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { return iParent.compareAndSet(iParent.mainNode(), updatedCnode) ? Action.OK : Action.REPEAT; } + @Override public void add(Subscription newSubscription) { Action res; do { @@ -282,14 +231,13 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { } /** - * * Removes subscription from CTrie, adds TNode when the last client unsubscribes, then calls for cleanTomb in a - * seperate atomic CAS operation. - * + * separate atomic CAS operation. * * @param topic * @param clientID */ + @Override public void removeSubscription(Topic topic, String clientID) { Action res; do { @@ -329,27 +277,14 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { } } - private class SubscriptionCounterVisitor implements IVisitor { - - AtomicInteger accumulator = new AtomicInteger(0); - - @Override - public void visit(CNode node, int deep) { - accumulator.addAndGet(node.subscriptions.size()); - } - - @Override - public Integer getResult() { - return accumulator.get(); - } - } - + @Override public int size() { SubscriptionCounterVisitor visitor = new SubscriptionCounterVisitor(); dfsVisit(this.root, visitor, 0); return visitor.getResult(); } + @Override public String dumpTree() { DumpTreeVisitor visitor = new DumpTreeVisitor(); dfsVisit(this.root, visitor, 0); diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/DumpTreeVisitor.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/DumpTreeVisitor.java new file mode 100644 index 00000000..084641e0 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/DumpTreeVisitor.java @@ -0,0 +1,67 @@ +/* + * Copyright (c) 2012-2017 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.subscriptions; + +import io.netty.util.internal.StringUtil; + +class DumpTreeVisitor implements CTrieSubscriptionDirectory.IVisitor { + + String s = ""; + + @Override + public void visit(CNode node, int deep) { + String indentTabs = indentTabs(deep); + s += indentTabs + (node.token == null ? "''" : node.token.toString()) + prettySubscriptions(node) + "\n"; + } + + private String prettySubscriptions(CNode node) { + if (node instanceof TNode) { + return "TNode"; + } + if (node.subscriptions.isEmpty()) { + return StringUtil.EMPTY_STRING; + } + StringBuilder subScriptionsStr = new StringBuilder(" ~~["); + int counter = 0; + for (Subscription couple : node.subscriptions) { + subScriptionsStr + .append("{filter=").append(couple.topicFilter).append(", ") + .append("client='").append(couple.clientId).append("'}"); + counter++; + if (counter < node.subscriptions.size()) { + subScriptionsStr.append(";"); + } + } + return subScriptionsStr.append("]").toString(); + } + + private String indentTabs(int deep) { + StringBuilder s = new StringBuilder(); + if (deep > 0) { + s.append(" "); + for (int i = 0; i < deep - 1; i++) { + s.append("| "); + } + s.append("|-"); + } + return s.toString(); + } + + @Override + public String getResult() { + return s; + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionCounterVisitor.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionCounterVisitor.java new file mode 100644 index 00000000..6398b2bb --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionCounterVisitor.java @@ -0,0 +1,33 @@ +/* + * Copyright (c) 2012-2017 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.subscriptions; + +import java.util.concurrent.atomic.AtomicInteger; + +class SubscriptionCounterVisitor implements CTrieSubscriptionDirectory.IVisitor { + + private AtomicInteger accumulator = new AtomicInteger(0); + + @Override + public void visit(CNode node, int deep) { + accumulator.addAndGet(node.subscriptions.size()); + } + + @Override + public Integer getResult() { + return accumulator.get(); + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java index 110e1ebf..cbd4b3e2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java @@ -35,11 +35,11 @@ public class Token { } protected boolean match(Token t) { - if (t == MULTI || t == SINGLE) { + if (MULTI.equals(t) || SINGLE.equals(t)) { return false; } - if (this == MULTI || this == SINGLE) { + if (MULTI.equals(this) || SINGLE.equals(this)) { return true; } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java index 8ed70a95..33dfbc67 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java @@ -165,7 +165,7 @@ public class Topic implements Serializable { int i = 0; for (; i < subscriptionTokens.size(); i++) { Token subToken = subscriptionTokens.get(i); - if (subToken != Token.MULTI && subToken != Token.SINGLE) { + if (!Token.MULTI.equals(subToken) && !Token.SINGLE.equals(subToken)) { if (i >= msgTokens.size()) { return false; } @@ -174,18 +174,14 @@ public class Topic implements Serializable { return false; } } else { - if (subToken == Token.MULTI) { + if (Token.MULTI.equals(subToken)) { return true; } - if (subToken == Token.SINGLE) { - // skip a step forward - } +// if (Token.SINGLE.equals(subToken)) { +// // skip a step forward +// } } } - // if last token was a SINGLE then treat it as an empty - // if (subToken == Token.SINGLE && (i - msgTokens.size() == 1)) { - // i--; - // } return i == msgTokens.size(); } diff --git a/broker/src/test/java/io/moquette/Utils.java b/broker/src/test/java/io/moquette/Utils.java index 24ca0886..1bf80c14 100644 --- a/broker/src/test/java/io/moquette/Utils.java +++ b/broker/src/test/java/io/moquette/Utils.java @@ -169,19 +169,19 @@ public final class Utils { return 3; if (2097152 <= len && len <= 268435455) return 4; - throw new IllegalArgumentException("value shoul be in the range [0..268435455]"); + throw new IllegalArgumentException("value should be in the range [0..268435455]"); } static byte encodeFlags(MqttMessage message) { byte flags = 0; if (message.fixedHeader().isDup()) { - flags |= 0x08; + flags |= (byte) 0x08; } if (message.fixedHeader().isRetain()) { - flags |= 0x01; + flags |= (byte) 0x01; } - flags |= (message.fixedHeader().qosLevel().value() & 0x03) << 1; + flags |= (byte) ((message.fixedHeader().qosLevel().value() & 0x03) << 1); return flags; } diff --git a/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java b/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java index 882440d9..577a3246 100644 --- a/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java +++ b/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java @@ -1,3 +1,19 @@ +/* + * Copyright (c) 2012-2017 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.persistence; import io.moquette.spi.*; @@ -9,11 +25,14 @@ import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.Test; +import java.io.UnsupportedEncodingException; +import java.nio.charset.StandardCharsets; import java.util.List; import static io.moquette.spi.impl.subscriptions.Topic.asTopic; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.EXACTLY_ONCE; +import static java.nio.charset.StandardCharsets.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -28,9 +47,9 @@ public abstract class MessageStoreTCK { protected SessionsRepository sessionsRepository; @Test - public void testDropMessagesInSessionDoesntCleanAnyRetainedStoredMessages() { + public void testDropMessagesInSessionDoesntCleanAnyRetainedStoredMessages() throws UnsupportedEncodingException { final ClientSession session = sessionsRepository.createNewSession(TEST_CLIENT, true); - StoredMessage publishToStore = new StoredMessage("Hello".getBytes(), EXACTLY_ONCE, "/topic"); + StoredMessage publishToStore = new StoredMessage("Hello".getBytes(UTF_8), EXACTLY_ONCE, "/topic"); publishToStore.setClientID(TEST_CLIENT); publishToStore.setRetained(true); messagesStore.storeRetained(new Topic("/topic"), publishToStore); @@ -50,8 +69,8 @@ public abstract class MessageStoreTCK { } @Test - public void testStoreRetained() { - StoredMessage msgStored = new StoredMessage("Hello".getBytes(), MqttQoS.AT_LEAST_ONCE, "/topic"); + public void testStoreRetained() throws UnsupportedEncodingException { + StoredMessage msgStored = new StoredMessage("Hello".getBytes(UTF_8), MqttQoS.AT_LEAST_ONCE, "/topic"); msgStored.setClientID(TEST_CLIENT); messagesStore.storeRetained(asTopic("/topic"), msgStored); @@ -68,7 +87,7 @@ public abstract class MessageStoreTCK { final ByteBuf payload = msgRetrieved.getPayload(); byte[] content = new byte[payload.readableBytes()]; payload.readBytes(content); - assertEquals("Hello", new String(content)); + assertEquals("Hello", new String(content, UTF_8)); } @Test diff --git a/broker/src/test/java/io/moquette/server/MessageCollector.java b/broker/src/test/java/io/moquette/server/MessageCollector.java index d8103634..04b235c7 100644 --- a/broker/src/test/java/io/moquette/server/MessageCollector.java +++ b/broker/src/test/java/io/moquette/server/MessageCollector.java @@ -26,7 +26,7 @@ import java.util.concurrent.*; */ public class MessageCollector implements MqttCallback { - private final class ReceivedMessage { + private static final class ReceivedMessage { private final MqttMessage message; private final String topic; diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java index a791d989..b1e96bdf 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java @@ -27,6 +27,8 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.Properties; import java.util.concurrent.TimeUnit; + +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.*; public class ServerIntegrationFuseTest { @@ -98,6 +100,6 @@ public class ServerIntegrationFuseTest { Message msg = m_subscriber.receive(500, TimeUnit.MILLISECONDS); assertNotNull("We should get notified with 'Will' message", msg); msg.ack(); - assertEquals(willTestamentMsg, new String(msg.getPayload())); + assertEquals(willTestamentMsg, new String(msg.getPayload(), UTF_8)); } } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java index 753794c3..2be5ed20 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -35,6 +35,7 @@ import java.sql.SQLException; import java.util.Properties; import static io.moquette.BrokerConstants.*; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; public class ServerIntegrationHazelcastHandlerInterceptorTest { @@ -122,7 +123,7 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest { LOG.info("*** checkPublishPassThroughCluster_Qos0 ***"); m_listener.subscribe("/topic", 1); - m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); + m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(UTF_8), 0, false); MqttMessage messageQos0 = m_messagesCollector.waitMessage(1); assertEquals("Hello world MQTT QoS0", messageQos0.toString()); assertEquals(0, messageQos0.getQos()); @@ -133,7 +134,7 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest { LOG.info("*** checkPublishPassThroughCluster_Qos1 ***"); m_listener.subscribe("/topic", 1); - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); + m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(UTF_8), 1, false); MqttMessage messageQos1 = m_messagesCollector.waitMessage(1); assertEquals("Hello world MQTT QoS1", messageQos1.toString()); assertEquals(1, messageQos1.getQos()); @@ -144,7 +145,7 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest { LOG.info("*** checkPublishPassThroughCluster_Qos2 ***"); m_listener.subscribe("/topic", 2); - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 2, false); MqttMessage messageQos2 = m_messagesCollector.waitMessage(1); assertEquals("Hello world MQTT QoS2", messageQos2.toString()); assertEquals(2, messageQos2.getQos()); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 73a2f70c..0d31f9ae 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -26,6 +26,8 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.util.Properties; + +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.*; public class ServerIntegrationPahoTest { @@ -83,7 +85,6 @@ public class ServerIntegrationPahoTest { m_server.stopServer(); } - @Ignore("This test hasn't any meaning using in memory storage service") @Test public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception { @@ -100,12 +101,11 @@ public class ServerIntegrationPahoTest { // reconnect and publish m_client.connect(options); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); + m_client.publish("/topic", "Test my payload".getBytes(UTF_8), 0, false); assertEquals("/topic", m_messagesCollector.getTopic()); } - /** * subscriber A connect and subscribe on "a/b" QoS 1 subscriber B connect and subscribe on "a/+" * BUT with QoS 2 publisher connects and send a message "hello" on "a/b" subscriber A must @@ -133,15 +133,15 @@ public class ServerIntegrationPahoTest { subscriberB.subscribe("a/+", 2); m_client.connect(); - m_client.publish("a/b", "Hello world MQTT!!".getBytes(), 2, false); + m_client.publish("a/b", "Hello world MQTT!!".getBytes(UTF_8), 2, false); MqttMessage messageOnA = cbSubscriberA.waitMessage(1); - assertEquals("Hello world MQTT!!", new String(messageOnA.getPayload())); + assertEquals("Hello world MQTT!!", new String(messageOnA.getPayload(), UTF_8)); assertEquals(1, messageOnA.getQos()); subscriberA.disconnect(); MqttMessage messageOnB = cbSubscriberB.waitMessage(1); - assertEquals("Hello world MQTT!!", new String(messageOnB.getPayload())); + assertEquals("Hello world MQTT!!", new String(messageOnB.getPayload(), UTF_8)); assertEquals(2, messageOnB.getQos()); subscriberB.disconnect(); } @@ -180,15 +180,15 @@ public class ServerIntegrationPahoTest { tmpDir + File.separator + "clientForPublish"); MqttClient clientForPublish = new MqttClient("tcp://localhost:1883", "clientForPublish", dsSubscriberPUB); clientForPublish.connect(); - clientForPublish.publish("topic", "Hello".getBytes(), 2, true); + clientForPublish.publish("topic", "Hello".getBytes(UTF_8), 2, true); // verify clientForSubscribe1 doesn't receive a notification but clientForSubscribe2 yes - LOG.info("Before waiting to receive 1 sec from " + clientForSubscribe1.getClientId()); + LOG.info("Before waiting to receive 1 sec from {}", clientForSubscribe1.getClientId()); assertFalse(clientForSubscribe1.isConnected()); assertTrue(clientForSubscribe2.isConnected()); - LOG.info("Waiting to receive 1 sec from " + clientForSubscribe2.getClientId()); + LOG.info("Waiting to receive 1 sec from {}", clientForSubscribe2.getClientId()); MqttMessage messageOnB = cbSubscriber2.waitMessage(1); - assertEquals("Hello", new String(messageOnB.getPayload())); + assertEquals("Hello", new String(messageOnB.getPayload(), UTF_8)); } // @Test diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java index ed81b441..b8f21729 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java @@ -32,6 +32,8 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.util.Properties; + +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; public class ServerIntegrationQoSValidationTest { @@ -92,7 +94,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS0ReceiveQoS0publishes ***"); m_subscriber.subscribe("/topic", 0); - m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); + m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(UTF_8), 0, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS0", message.toString()); assertEquals(0, message.getQos()); @@ -103,7 +105,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS0ReceiveQoS1publishes_downgrade ***"); m_subscriber.subscribe("/topic", 0); - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); + m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(UTF_8), 1, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS1", message.toString()); assertEquals(0, message.getQos()); @@ -114,7 +116,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS0ReceiveQoS2publishes_downgrade ***"); m_subscriber.subscribe("/topic", 0); - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 2, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS2", message.toString()); assertEquals(0, message.getQos()); @@ -125,7 +127,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS1ReceiveQoS0publishes ***"); m_subscriber.subscribe("/topic", 1); - m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); + m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(UTF_8), 0, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS0", message.toString()); assertEquals(0, message.getQos()); @@ -136,7 +138,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS1ReceiveQoS1publishes ***"); m_subscriber.subscribe("/topic", 1); - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); + m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(UTF_8), 1, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS1", message.toString()); assertEquals(1, message.getQos()); @@ -147,7 +149,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS1ReceiveQoS2publishes_downgrade ***"); m_subscriber.subscribe("/topic", 1); - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 2, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS2", message.toString()); assertEquals(1, message.getQos()); @@ -158,7 +160,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS2ReceiveQoS0publishes ***"); m_subscriber.subscribe("/topic", 2); - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 0, false); + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 0, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS2", message.toString()); assertEquals(0, message.getQos()); @@ -169,7 +171,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS2ReceiveQoS1publishes ***"); m_subscriber.subscribe("/topic", 2); - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 1, false); + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 1, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS2", message.toString()); assertEquals(1, message.getQos()); @@ -180,7 +182,7 @@ public class ServerIntegrationQoSValidationTest { LOG.info("*** checkSubscriberQoS2ReceiveQoS2publishes ***"); m_subscriber.subscribe("/topic", 2); - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 2, false); MqttMessage message = m_callback.waitMessage(1); assertEquals("Hello world MQTT QoS2", message.toString()); assertEquals(2, message.getQos()); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java index df6a5c9b..a5776d47 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java @@ -27,6 +27,8 @@ import org.junit.Test; import java.io.File; import java.io.IOException; import java.util.Properties; + +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -100,11 +102,11 @@ public class ServerIntegrationRestartTest { // should be just one registration so a publisher receive one notification m_publisher.connect(CLEAN_SESSION_OPT); - m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 1, false); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(UTF_8), 1, false); // read the messages MqttMessage msg = m_messageCollector.waitMessage(1); - assertEquals("Hello world MQTT!!", new String(msg.getPayload())); + assertEquals("Hello world MQTT!!", new String(msg.getPayload(), UTF_8)); // no more messages on the same topic will be received assertNull(m_messageCollector.waitMessage(1)); } @@ -121,7 +123,7 @@ public class ServerIntegrationRestartTest { m_server.startServer(IntegrationUtils.prepareTestProperties()); m_publisher.connect(); - m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 0, false); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(UTF_8), 0, false); } @Test @@ -147,7 +149,7 @@ public class ServerIntegrationRestartTest { // publisher publishes on /topic m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); m_publisher.connect(); - m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 1, false); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(UTF_8), 1, false); // Expected // the subscriber doesn't get notified (it's fully unsubscribed) @@ -163,9 +165,9 @@ public class ServerIntegrationRestartTest { client.setCallback(collector); client.connect(); client.subscribe(topic, 1); - client.publish(topic, "Hello world MQTT!!".getBytes(), 0, false); + client.publish(topic, "Hello world MQTT!!".getBytes(UTF_8), 0, false); MqttMessage msg = collector.waitMessage(1); - assertEquals("Hello world MQTT!!", new String(msg.getPayload())); + assertEquals("Hello world MQTT!!", new String(msg.getPayload(), UTF_8)); return client; } } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java index 2a19ee51..af2d4b7f 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java @@ -16,6 +16,7 @@ package io.moquette.server; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertFalse; import java.io.File; import java.io.IOException; @@ -150,7 +151,7 @@ public class ServerIntegrationSSLTest { MqttConnectOptions secondClientOptions = new MqttConnectOptions(); secondClientOptions.setSocketFactory(ssf); secondClient.connect(secondClientOptions); - secondClient.publish("/topic", new MqttMessage("message".getBytes())); + secondClient.publish("/topic", new MqttMessage("message".getBytes(UTF_8))); secondClient.disconnect(); m_client.disconnect(); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java index 29461264..9d7c270a 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java @@ -65,6 +65,7 @@ public class ServerIntegrationWebSocketTest { m_server.stopServer(); } + @SuppressWarnings("FutureReturnValueIgnored") @Test public void checkPlainConnect() throws Exception { LOG.info("*** checkPlainConnect ***"); diff --git a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 1fe8de61..a1242ed1 100644 --- a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -33,6 +33,7 @@ import java.nio.charset.StandardCharsets; import java.util.Properties; import java.util.concurrent.TimeUnit; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.*; public class ServerLowlevelMessagesIntegrationTests { @@ -121,7 +122,7 @@ public class ServerLowlevelMessagesIntegrationTests { assertNotNull("the will message should be fired after keep alive!", msg); // the will message hasn't to be received before the elapsing of Keep Alive timeout assertTrue(System.currentTimeMillis() - connectTime > 3000); - assertEquals(willTestamentMsg, new String(msg.getPayload(), StandardCharsets.UTF_8)); + assertEquals(willTestamentMsg, new String(msg.getPayload(), UTF_8)); }); m_willSubscriber.disconnect(); @@ -153,7 +154,7 @@ public class ServerLowlevelMessagesIntegrationTests { // Verify will testament is published org.eclipse.paho.client.mqttv3.MqttMessage receivedTestament = m_messageCollector.waitMessage(1); - assertEquals(willTestamentMsg, new String(receivedTestament.getPayload())); + assertEquals(willTestamentMsg, new String(receivedTestament.getPayload(), UTF_8)); m_willSubscriber.disconnect(); } diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index 005fb795..982ec40f 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -64,15 +64,16 @@ public class ClientSessionTest { subscriptionsDirectory.add(oldSubscription); // Subscribe on /topic again that overrides the previous subscription. - Subscription overrindingSubscription = new Subscription(sessionClean.clientID, new Topic("/topic"), EXACTLY_ONCE); - sessionClean.subscribe(overrindingSubscription); - subscriptionsDirectory.add(overrindingSubscription); + Subscription overridingSubscription = new Subscription(sessionClean.clientID, new Topic("/topic"), + EXACTLY_ONCE); + sessionClean.subscribe(overridingSubscription); + subscriptionsDirectory.add(overridingSubscription); // Verify List subscriptions = subscriptionsDirectory.matches(new Topic("/topic")); assertEquals("Only one subscription must match /topic", 1, subscriptions.size()); Subscription sub = subscriptions.get(0); assertEquals("Matching subscription MUST have higher QoS", - overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); + overridingSubscription.getRequestedQos(), sub.getRequestedQos()); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index a5d83f1a..7fe3ecbd 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -23,7 +23,6 @@ import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.*; import io.netty.buffer.Unpooled; -import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.*; @@ -33,6 +32,7 @@ import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEP import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -104,11 +104,10 @@ abstract class AbstractProtocolProcessorCommonUtils { final String clientId = NettyUtils.clientID(channel); Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos); - verifySubscriptionExists(channel, m_sessionStore, expectedSubscription); + verifySubscriptionExists(m_sessionStore, expectedSubscription); } - protected static void verifySubscriptionExists(Channel channel, ISessionsStore sessionsStore, Subscription expectedSubscription) { - final String clientId = NettyUtils.clientID(channel); + protected static void verifySubscriptionExists(ISessionsStore sessionsStore, Subscription expectedSubscription) { final Subscription subscription = sessionsStore.subscriptionStore().reload(expectedSubscription); assertEquals(expectedSubscription, subscription); } @@ -166,7 +165,7 @@ abstract class AbstractProtocolProcessorCommonUtils { .topicName(topic) .retained(retained) .qos(qos) - .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes())).build(); + .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes(UTF_8))).build(); this.m_processor.internalPublish(publish, "INTRPUBL"); } @@ -180,7 +179,7 @@ abstract class AbstractProtocolProcessorCommonUtils { .topicName(topic) .retained(retained) .qos(qos) - .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes())).build(); + .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes(UTF_8))).build(); this.m_processor.processPublish(m_channel, publish); } @@ -188,33 +187,37 @@ abstract class AbstractProtocolProcessorCommonUtils { publishToAs(this.m_channel, clientId, topic, qos, messageId, retained); } - protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, MqttQoS qos, int messageId, boolean retained) { + protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, MqttQoS qos, int messageId, + boolean retained) { publishToAs(channel, clientId, topic, HELLO_WORLD_MQTT, qos, messageId, retained); } - protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, String payload, MqttQoS qos, int messageId, boolean retained) { + protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, String payload, MqttQoS qos, + int messageId, boolean retained) { NettyUtils.userName(channel, clientId); MqttPublishMessage publish = MqttMessageBuilders.publish() .topicName(topic) .retained(retained) .messageId(messageId) .qos(qos) - .payload(Unpooled.copiedBuffer(payload.getBytes())).build(); + .payload(Unpooled.copiedBuffer(payload.getBytes(UTF_8))).build(); this.m_processor.processPublish(channel, publish); } - protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, int messageId, boolean retained) { + protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, int messageId, + boolean retained) { publishQoS2ToAs(channel, clientId, topic, HELLO_WORLD_MQTT, messageId, retained); } - protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, String payload, int messageId, boolean retained) { + protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, String payload, + int messageId, boolean retained) { NettyUtils.userName(channel, clientId); MqttPublishMessage publish = MqttMessageBuilders.publish() .topicName(topic) .retained(retained) .messageId(messageId) .qos(MqttQoS.EXACTLY_ONCE) - .payload(Unpooled.copiedBuffer(payload.getBytes())).build(); + .payload(Unpooled.copiedBuffer(payload.getBytes(UTF_8))).build(); this.m_processor.processPublish(channel, publish); verifyPubrecIsReceived(channel, messageId); @@ -236,7 +239,6 @@ abstract class AbstractProtocolProcessorCommonUtils { assertEquals(messageId, Utils.messageId(pubComp)); } - protected void connect() { connectAsClient(FAKE_CLIENT_ID); } @@ -307,7 +309,7 @@ abstract class AbstractProtocolProcessorCommonUtils { protected void verifyPublishIsReceived(EmbeddedChannel channel) { final MqttPublishMessage publishReceived = channel.readOutbound(); - String payloadMessage = new String(publishReceived.payload().array()); + String payloadMessage = new String(publishReceived.payload().array(), UTF_8); assertEquals("Sent and received payload must be identical", HELLO_WORLD_MQTT, payloadMessage); } @@ -321,7 +323,7 @@ abstract class AbstractProtocolProcessorCommonUtils { protected void verifyPublishIsReceived(EmbeddedChannel channel, String expectedPayload, MqttQoS expectedQoS) { final MqttPublishMessage publishReceived = channel.readOutbound(); - String payloadMessage = new String(publishReceived.payload().array()); + String payloadMessage = new String(publishReceived.payload().array(), UTF_8); assertEquals("Sent and received payload must be identical", expectedPayload, payloadMessage); assertEquals("Expected QoS don't match", expectedQoS, publishReceived.fixedHeader().qosLevel()); } diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index 01bb15d4..073697f3 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -30,6 +30,7 @@ import org.junit.Test; import java.util.Collections; import java.util.concurrent.atomic.AtomicInteger; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; import static org.mockito.Matchers.refEq; import static org.mockito.Mockito.mock; @@ -126,7 +127,7 @@ public class BrokerInterceptorTest { public void testNotifyTopicPublished() throws Exception { interceptor.notifyTopicPublished( MqttMessageBuilders.publish().qos(MqttQoS.AT_MOST_ONCE) - .payload(Unpooled.copiedBuffer("Hello".getBytes())).build(), + .payload(Unpooled.copiedBuffer("Hello".getBytes(UTF_8))).build(), "cli1234", "cli1234"); interval(); diff --git a/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java b/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java index cc03205c..1575b7ed 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java +++ b/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java @@ -20,6 +20,8 @@ import java.util.Map; import java.util.Set; import io.moquette.spi.security.IAuthenticator; +import static java.nio.charset.StandardCharsets.UTF_8; + /** * Test utility to implements authenticator instance. */ @@ -33,6 +35,7 @@ class MockAuthenticator implements IAuthenticator { m_userPwds = userPwds; } + @Override public boolean checkValid(String clientId, String username, byte[] password) { if (!m_clientIds.contains(clientId)) { return false; @@ -43,7 +46,7 @@ class MockAuthenticator implements IAuthenticator { if (password == null) { return false; } - return m_userPwds.get(username).equals(new String(password)); + return m_userPwds.get(username).equals(new String(password, UTF_8)); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 8d530d5a..457e20ea 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -42,6 +42,7 @@ import static io.moquette.spi.impl.NettyChannelAssertions.assertConnAckAccepted; import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.*; import static org.mockito.Mockito.*; @@ -61,7 +62,6 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils static final List EMPTY_OBSERVERS = Collections.emptyList(); static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); - @Before public void setUp() throws InterruptedException { initializeProcessorAndSubsystems(); @@ -143,7 +143,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils // Exercise MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(AT_MOST_ONCE) - .retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes())).build(); + .retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes(UTF_8))).build(); NettyUtils.userName(m_channel, "FakeCLI"); m_processor.processPublish(m_channel, msg); @@ -269,7 +269,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils List emptySubs = Collections.emptyList(); when(subs.matches(any(Topic.class))).thenReturn(emptySubs); - StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(), MqttQoS.EXACTLY_ONCE, "/topic"); + StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(UTF_8), MqttQoS.EXACTLY_ONCE, "/topic"); retainedMessage.setRetained(true); retainedMessage.setClientID(FAKE_PUBLISHER_ID); m_messagesStore.storeRetained(new Topic("/topic"), retainedMessage); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 949ce1de..6cf32a67 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -69,7 +69,8 @@ public class ProtocolProcessor_CONNECT_Test { subscriptions.init(sessionsRepository); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null)); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, + new SessionsRepository(this.m_sessionStore, null)); } @Test @@ -286,7 +287,7 @@ public class ProtocolProcessor_CONNECT_Test { FAKE_CLIENT_ID, new Topic(ProtocolProcessorTest.FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); - verifySubscriptionExists(m_session, m_sessionStore, expectedSubscription); + verifySubscriptionExists(m_sessionStore, expectedSubscription); assertEqualsSubAck(m_session.readOutbound()); // disconnect @@ -300,7 +301,7 @@ public class ProtocolProcessor_CONNECT_Test { assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); // verify that the first subscription is still preserved - verifySubscriptionExists(m_session, m_sessionStore, expectedSubscription); + verifySubscriptionExists(m_sessionStore, expectedSubscription); } @Test diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java index 5f173502..35f2e53d 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java @@ -201,7 +201,8 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws InterruptedException { + public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() + throws InterruptedException { connect(); subscribe("/topic", AT_LEAST_ONCE); disconnect(); @@ -221,7 +222,6 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc verifyPublishIsReceived("Hello MQTT 2", AT_LEAST_ONCE); } - @Test public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws InterruptedException { connect(); diff --git a/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java index 3ed70cc0..b94ac3aa 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java @@ -24,6 +24,8 @@ import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.sql.*; + +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -70,7 +72,7 @@ public class DBAuthenticatorTest { JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256); - assertTrue(dbAuthenticator.checkValid(null, "dbuser", "password".getBytes())); + assertTrue(dbAuthenticator.checkValid(null, "dbuser", "password".getBytes(UTF_8))); } @Test @@ -80,7 +82,7 @@ public class DBAuthenticatorTest { JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256); - assertFalse(dbAuthenticator.checkValid(null, "dbuser2", "password".getBytes())); + assertFalse(dbAuthenticator.checkValid(null, "dbuser2", "password".getBytes(UTF_8))); } @Test @@ -90,7 +92,7 @@ public class DBAuthenticatorTest { JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256); - assertFalse(dbAuthenticator.checkValid(null, "dbuser", "wrongPassword".getBytes())); + assertFalse(dbAuthenticator.checkValid(null, "dbuser", "wrongPassword".getBytes(UTF_8))); } @After diff --git a/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java index 7b344653..b756cbbb 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java @@ -18,6 +18,8 @@ package io.moquette.spi.impl.security; import io.moquette.spi.security.IAuthenticator; import org.junit.Test; + +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -29,7 +31,7 @@ public class FileAuthenticatorTest { String file = getClass().getResource("/password_file.conf").getPath(); IAuthenticator auth = new FileAuthenticator(null, file); - assertTrue(auth.checkValid(null, "testuser", "passwd".getBytes())); + assertTrue(auth.checkValid(null, "testuser", "passwd".getBytes(UTF_8))); } @Test @@ -37,14 +39,14 @@ public class FileAuthenticatorTest { String file = getClass().getResource("/password_file.conf").getPath(); IAuthenticator auth = new FileAuthenticator(null, file); - assertFalse(auth.checkValid(null, "testuser2", "passwd".getBytes())); + assertFalse(auth.checkValid(null, "testuser2", "passwd".getBytes(UTF_8))); } @Test public void loadPasswordFile_verifyDirectoryRef() { IAuthenticator auth = new FileAuthenticator("", ""); - assertFalse(auth.checkValid(null, "testuser2", "passwd".getBytes())); + assertFalse(auth.checkValid(null, "testuser2", "passwd".getBytes(UTF_8))); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java index 84b53fb9..230812ea 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java @@ -83,10 +83,11 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.add(financeAnySub); // Verify - assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub); - assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub); - - System.out.println(sut.dumpTree()); + assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)) + .containsExactlyInAnyOrder(financeAnySub, anySub); + assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)) + .containsExactlyInAnyOrder(financeAnySub, anySub); +// System.out.println(sut.dumpTree()); } @Test diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java index 9ce84dc4..0423a7cd 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java @@ -144,8 +144,8 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithSomeNodeWhenRemoveMultipleTimes() { sut.add(clientSubOnTopic("TempSensor1", "test")); - - // make sure no TNode exceptions + + // make sure no TNode exceptions sut.removeSubscription(asTopic("test"), "TempSensor1"); sut.removeSubscription(asTopic("test"), "TempSensor1"); sut.removeSubscription(asTopic("test"), "TempSensor1"); @@ -192,35 +192,30 @@ public class CTrieSubscriptionDirectoryTest { sut.add(clientSubOnTopic("TempSensor1", "/temp/2")); //Exercise - sut.removeSubscription(asTopic("/temp"), "TempSensor1"); final Set matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1"), sut.root); final Set matchingSubs2 = sut.recursiveMatch(asTopic("/temp/2"), sut.root); //Verify - // not clear to me, but I believe /temp unsubscribe should not unsub you from downstream /temp/1 or /temp/2 - final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("/temp/1")); assertThat(matchingSubs1).contains(expectedMatchingsub1); final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("/temp/2")); assertThat(matchingSubs2).contains(expectedMatchingsub2); } - @Test public void givenTreeWithDeepNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() { sut.add(clientSubOnTopic("TempSensor1", "/bah/bin/bash")); - sut.removeSubscription(asTopic("/bah/bin/bash"),"TempSensor1"); + sut.removeSubscription(asTopic("/bah/bin/bash"), "TempSensor1"); //Verify final Optional matchedNode = sut.lookup(asTopic("/bah/bin/bash")); assertFalse("Node on path /temp can't be present", matchedNode.isPresent()); } - @Test public void testMatchSubscriptionNoWildcards() { sut.add(clientSubOnTopic("TempSensor1", "/temp")); @@ -237,7 +232,7 @@ public class CTrieSubscriptionDirectoryTest { public void testRemovalInnerTopicOffRootSameClient() { sut.add(clientSubOnTopic("TempSensor1", "temp")); sut.add(clientSubOnTopic("TempSensor1", "temp/1")); - + //Exercise final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); @@ -249,7 +244,7 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs1).contains(expectedMatchingsub1); assertThat(matchingSubs2).contains(expectedMatchingsub2); - sut.removeSubscription(asTopic("temp"),"TempSensor1"); + sut.removeSubscription(asTopic("temp"), "TempSensor1"); //Exercise final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); @@ -275,7 +270,7 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs1).contains(expectedMatchingsub1); assertThat(matchingSubs2).contains(expectedMatchingsub2); - sut.removeSubscription(asTopic("temp"),"TempSensor1"); + sut.removeSubscription(asTopic("temp"), "TempSensor1"); //Exercise final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); @@ -301,7 +296,7 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs1).contains(expectedMatchingsub1); assertThat(matchingSubs2).contains(expectedMatchingsub2); - sut.removeSubscription(asTopic("temp/1"),"TempSensor2"); + sut.removeSubscription(asTopic("temp/1"), "TempSensor2"); //Exercise final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); @@ -311,5 +306,3 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs4).doesNotContain(expectedMatchingsub2); } } - - diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java index 95cfe5c2..bb32cc95 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java @@ -122,7 +122,6 @@ public class TopicTest { assertEquals(Topic.asTopic("a/b"), Topic.asTopic("/a/b").exceptHeadToken()); } - public static TopicAssert assertThat(Topic topic) { return new TopicAssert(topic); } diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java index 52c35d4f..05722440 100644 --- a/broker/src/test/java/io/moquette/testclient/Client.java +++ b/broker/src/test/java/io/moquette/testclient/Client.java @@ -25,9 +25,13 @@ import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; +import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE; + /** * Class used just to send and receive MQTT messages without any protocol login in action, just use * the encoder/decoder part. @@ -52,6 +56,7 @@ public class Client { this(host, BrokerConstants.PORT); } + @SuppressWarnings("FutureReturnValueIgnored") public Client(String host, int port) { handler.setClient(this); workerGroup = new NioEventLoopGroup(); @@ -157,7 +162,7 @@ public class Client { } public void sendMessage(MqttMessage msg) { - m_channel.writeAndFlush(msg); + m_channel.writeAndFlush(msg).addListener(FIRE_EXCEPTION_ON_FAILURE); } public MqttMessage lastReceivedMessage() { @@ -179,9 +184,10 @@ public class Client { return m_connectionLost; } + @SuppressWarnings("FutureReturnValueIgnored") public void close() throws InterruptedException { // Wait until the connection is closed. - m_channel.closeFuture().sync(); + m_channel.closeFuture().sync().addListener(CLOSE_ON_FAILURE); if (workerGroup == null) { throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); } diff --git a/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java b/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java index 76d28424..09629b48 100644 --- a/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java +++ b/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java @@ -24,6 +24,8 @@ import io.netty.handler.codec.mqtt.MqttMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + @ChannelHandler.Sharable class ClientNettyMQTTHandler extends ChannelInboundHandlerAdapter { @@ -38,9 +40,9 @@ class ClientNettyMQTTHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { + public void channelInactive(ChannelHandlerContext ctx) { m_client.setConnectionLost(true); - ctx.close(/* false */); + ctx.close().addListener(CLOSE_ON_FAILURE); } void setClient(Client client) { diff --git a/broker/src/test/java/io/moquette/testclient/RawClient.java b/broker/src/test/java/io/moquette/testclient/RawClient.java index 81ad8348..a8e24dc7 100644 --- a/broker/src/test/java/io/moquette/testclient/RawClient.java +++ b/broker/src/test/java/io/moquette/testclient/RawClient.java @@ -32,6 +32,8 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + /** * This class is used to have a fluent interface to interact with a server. Inspired by * org.kaazing.robot @@ -55,10 +57,9 @@ public final class RawClient { } @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - // m_client.setConnectionLost(true); + public void channelInactive(ChannelHandlerContext ctx) { disconnectLatch.countDown(); - ctx.close(/* false */); + ctx.close().addListener(CLOSE_ON_FAILURE); } } @@ -72,6 +73,7 @@ public final class RawClient { private CountDownLatch disconnectLatch; private final Semaphore readableBytesSem; + @SuppressWarnings("FutureReturnValueIgnored") private RawClient(String host, int port) { handler = new RawMessageHandler(); heapBuffer = Unpooled.buffer(128); @@ -120,14 +122,14 @@ public final class RawClient { for (int b : bytes) { buff.writeByte((byte) b); } - m_channel.write(buff); + m_channel.write(buff).addListener(CLOSE_ON_FAILURE); return this; } public RawClient writeWithSize(String str) { ByteBuf buff = Unpooled.buffer(str.length() + 2); buff.writeBytes(Utils.encodeString(str)); - m_channel.write(buff); + m_channel.write(buff).addListener(CLOSE_ON_FAILURE); return this; } @@ -145,7 +147,7 @@ public final class RawClient { throw new IllegalStateException(ex); } out.writeBytes(raw); - m_channel.write(out); + m_channel.write(out).addListener(CLOSE_ON_FAILURE); return this; } diff --git a/checkstyle.xml b/checkstyle.xml index eca8d2ed..b4c27646 100644 --- a/checkstyle.xml +++ b/checkstyle.xml @@ -43,11 +43,11 @@ - + diff --git a/distribution/build.gradle b/distribution/build.gradle index d91bb071..1a3a6251 100644 --- a/distribution/build.gradle +++ b/distribution/build.gradle @@ -1,11 +1,9 @@ - -description = 'Moquette - distribution' - apply plugin: 'application' - // adds task shadowJar - Create a combined JAR of project and runtime dependencies +description = 'Moquette - distribution' apply plugin: 'com.github.johnrengelman.shadow' + mainClassName = "io.moquette.server.Server" dependencies { @@ -17,35 +15,35 @@ dependencies { def packageSet = project.copySpec { from 'src/main/resources/README.txt' - into ('bin'){ - from 'src/main/scripts/' + into ('bin') { + from 'src/ main/scripts/' fileMode=0755 } - into ('lib'){ + into ('lib') { from "build/libs/${project.name}-${version}.jar" rename { "moquette-${version}.jar" } } - into ('lib'){ + into ('lib') { from configurations.compile } - into ('config'){ + into ('config') { from 'src/main/resources/moquette.conf' from 'src/main/resources/password_file.conf' from 'src/main/resources/hazelcast.xml' // from 'src/main/resources/acl.conf' } - into ('config'){ + into ('config') { from 'src/main/resources/log4j.properties' rename { 'moquette-log.properties' } } } task distMoquetteTar(type: Tar) { - archiveName = "moquette-${version}.tar" + archiveName = "moquette-${version}.tar.gz" destinationDir file('build') extension 'tar' compression = Compression.GZIP diff --git a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java index 6a4c2a30..60a54a44 100644 --- a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java +++ b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java @@ -32,6 +32,8 @@ import java.io.IOException; import java.util.Collections; import java.util.List; +import static java.nio.charset.StandardCharsets.UTF_8; + final class EmbeddedLauncher { @@ -44,8 +46,8 @@ final class EmbeddedLauncher { @Override public void onPublish(InterceptPublishMessage msg) { - System.out.println( - "Received on topic: " + msg.getTopicName() + " content: " + new String(msg.getPayload().array())); + final String decodedPayload = new String(msg.getPayload().array(), UTF_8); + System.out.println("Received on topic: " + msg.getTopicName() + " content: " + decodedPayload); } } @@ -73,7 +75,7 @@ final class EmbeddedLauncher { // qos(MqttQoS.AT_MOST_ONCE); // qQos(MqttQoS.AT_LEAST_ONCE); .qos(MqttQoS.EXACTLY_ONCE) - .payload(Unpooled.copiedBuffer("Hello World!!".getBytes())) + .payload(Unpooled.copiedBuffer("Hello World!!".getBytes(UTF_8))) .build(); mqttBroker.internalPublish(message, "INTRLPUB"); diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java index 3b1fc895..70065697 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2017 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.persistence.h2; import org.h2.mvstore.MVMap; diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java index 43b54a19..c2feb4ed 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java @@ -41,17 +41,18 @@ public class H2PersistentStore implements IStore { public H2PersistentStore(IConfig props, ScheduledExecutorService scheduler) { this.storePath = props.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, ""); - this.autosaveInterval = Integer.parseInt(props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30")); + final String autosaveProp = props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30"); + this.autosaveInterval = Integer.parseInt(autosaveProp); this.scheduler = scheduler; } + @SuppressWarnings("FutureReturnValueIgnored") @Override public void initStore() { LOG.info("Initializing H2 store"); if (storePath == null || storePath.isEmpty()) { LOG.warn("H2 store file path is empty, using in-memory store"); mvStore = MVStore.open(null); - } else { mvStore = new MVStore.Builder() .fileName(storePath) @@ -60,13 +61,9 @@ public class H2PersistentStore implements IStore { } LOG.info("Scheduling H2 commit task"); - scheduler.scheduleWithFixedDelay(new Runnable() { - - @Override - public void run() { - LOG.debug("Committing to H2"); - mvStore.commit(); - } + scheduler.scheduleWithFixedDelay(() -> { + LOG.debug("Committing to H2"); + mvStore.commit(); }, this.autosaveInterval, this.autosaveInterval, TimeUnit.SECONDS); messageStore = new H2MessagesStore(mvStore); @@ -78,7 +75,6 @@ public class H2PersistentStore implements IStore { @Override public void close() { - } @Override diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java index 9134eecb..fb563dfe 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java @@ -71,7 +71,7 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { LOG.info("Adding new subscription CId={}, topics={}", newSubscription.getClientId(), newSubscription.getTopicFilter()); final String clientID = newSubscription.getClientId(); - final MVMap sessionSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + final MVMap sessionSubscriptions = this.mvStore.openMap(subscriptionsMapName(clientID)); sessionSubscriptions.put(newSubscription.getTopicFilter(), newSubscription); if (LOG.isTraceEnabled()) { @@ -84,10 +84,10 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { @Override public void removeSubscription(Topic topicFilter, String clientID) { LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter); - if (!this.mvStore.hasMap("subscriptions_" + clientID)) { + if (!this.mvStore.hasMap(subscriptionsMapName(clientID))) { return; } - final MVMap sessionsSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + final MVMap sessionsSubscriptions = this.mvStore.openMap(subscriptionsMapName(clientID)); sessionsSubscriptions.remove(topicFilter); if (LOG.isDebugEnabled()) { LOG.debug("Subscription has been removed. CId={}, topics={}, clientSubscriptions={}", clientID, @@ -98,10 +98,10 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { @Override public void wipeSubscriptions(String sessionID) { LOG.info("Wiping subscriptions. CId={}", sessionID); - if (!this.mvStore.hasMap("subscriptions_" + sessionID)) { + if (!this.mvStore.hasMap(subscriptionsMapName(sessionID))) { return; } - final MVMap subscriptions = this.mvStore.openMap("subscriptions_" + sessionID); + final MVMap subscriptions = this.mvStore.openMap(subscriptionsMapName(sessionID)); this.mvStore.removeMap(subscriptions); if (LOG.isDebugEnabled()) { LOG.debug("Subscriptions have been removed. CId={}, clientSubscriptions={}", sessionID, subscriptions); @@ -113,7 +113,8 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { LOG.debug("Retrieving existing subscriptions..."); List subscriptions = new ArrayList<>(); for (String clientID : this.sessions.keySet()) { - ConcurrentMap clientSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + final String mapName = subscriptionsMapName(clientID); + ConcurrentMap clientSubscriptions = this.mvStore.openMap(mapName); subscriptions.addAll(clientSubscriptions.values()); } LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions); @@ -122,7 +123,7 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { @Override public Collection listClientSubscriptions(String clientID) { - ConcurrentMap clientSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + ConcurrentMap clientSubscriptions = this.mvStore.openMap(subscriptionsMapName(clientID)); if (clientSubscriptions == null) { throw new IllegalStateException("Asking for subscriptions of not persisted client: " + clientID); } @@ -130,10 +131,15 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { } @Override - public Subscription reload(Subscription subcription) { - ConcurrentMap clientSubscriptions = this.mvStore.openMap("subscriptions_" + subcription.getClientId()); - LOG.debug("Retrieving subscriptions CId={}, subscriptions={}", subcription.getClientId(), clientSubscriptions); - return clientSubscriptions.get(subcription.getTopicFilter()); + public Subscription reload(Subscription subscription) { + final String subscriptionsMapName = subscriptionsMapName(subscription.getClientId()); + ConcurrentMap clientSubscriptions = this.mvStore.openMap(subscriptionsMapName); + LOG.debug("Retrieving subscriptions CId={}, subscriptions={}", subscription.getClientId(), clientSubscriptions); + return clientSubscriptions.get(subscription.getTopicFilter()); + } + + private String subscriptionsMapName(String clientId) { + return "subscriptions_" + clientId; } @Override diff --git a/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java b/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java index 1e4b911b..2e51a9ca 100644 --- a/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java +++ b/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2017 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.persistence.h2; import io.moquette.BrokerConstants; @@ -61,12 +76,12 @@ public class H2PersistentQueueTest { H2PersistentQueue sut = new H2PersistentQueue<>(this.mvStore, "test"); int numIterations = 10000000; - for (int i=0; i < numIterations; i++) { + for (int i = 0; i < numIterations; i++) { sut.add("Hello"); } mvStore.commit(); - for (int i=0; i < numIterations; i++) { + for (int i = 0; i < numIterations; i++) { assertEquals("Hello", sut.poll()); } diff --git a/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java b/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java index 6970dc25..c8017469 100644 --- a/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java +++ b/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java @@ -97,7 +97,8 @@ public class H2SessionsStoreTest { } @Test - public void testRetrieve1000ExpiredSessionPerformance() throws InterruptedException, ExecutionException, TimeoutException { + public void testRetrieve1000ExpiredSessionPerformance() throws InterruptedException, ExecutionException, + TimeoutException { H2SessionsStore sessionsStore = (H2SessionsStore) storageService.sessionsStore(); final ExecutorService executorService = Executors.newFixedThreadPool(2); diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java index 548ced06..e3ba23d3 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java @@ -78,7 +78,7 @@ class MapDBMessagesStore implements IMessagesStore { public void storeRetained(Topic topic, StoredMessage storedMessage) { LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); if (storedMessage.getClientID() == null) { - throw new IllegalArgumentException( "Message to be persisted must have a not null client ID"); + throw new IllegalArgumentException("Message to be persisted must have a not null client ID"); } m_retainedStore.put(topic, storedMessage); } diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java index 55116ae9..cabf286e 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java @@ -63,6 +63,7 @@ public class MapDBPersistentStore implements IStore { return m_sessionsStore; } + @SuppressWarnings("FutureReturnValueIgnored") @Override public void initStore() { LOG.info("Initializing MapDB store"); diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index 71ffe2f1..4e40080f 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -127,10 +127,12 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { } @Override - public Subscription reload(Subscription subcription) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + subcription.getClientId()); - LOG.debug("Retrieving subscriptions. CId={}, subscriptions={}", subcription.getClientId(), clientSubscriptions); - return clientSubscriptions.get(subcription.getTopicFilter()); + public Subscription reload(Subscription subscription) { + final String clientId = subscription.getClientId(); + final String clientSubcMapName = "subscriptions_" + clientId; + ConcurrentMap clientSubscriptions = m_db.getHashMap(clientSubcMapName); + LOG.debug("Retrieving subscriptions. CId={}, subscriptions={}", clientId, clientSubscriptions); + return clientSubscriptions.get(subscription.getTopicFilter()); } @Override @@ -178,7 +180,6 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { inFlightForClient.add(nextPacketId); this.m_inFlightIds.put(clientID, inFlightForClient); return nextPacketId; - } int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); @@ -264,7 +265,8 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { @Override public int getInflightMessagesNo(String clientID) { int totalInflight = 0; - ConcurrentMap inflightPerClient = m_db.getHashMap(inboundMessageId2MessagesMapName(clientID)); + final String infligthMessageMapName = inboundMessageId2MessagesMapName(clientID); + ConcurrentMap inflightPerClient = m_db.getHashMap(infligthMessageMapName); if (inflightPerClient != null) { totalInflight += inflightPerClient.size(); } diff --git a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java index 0643813a..96d19c6a 100644 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java @@ -28,11 +28,13 @@ import org.junit.Before; import org.junit.Test; import java.io.File; +import java.nio.charset.StandardCharsets; import java.util.Properties; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.*; /** @@ -99,7 +101,7 @@ public class MapDBPersistentStoreTest extends MessageStoreTCK { @Test public void testNextPacketID() { - StoredMessage msgStored = new StoredMessage("Hello".getBytes(), MqttQoS.AT_LEAST_ONCE, "/topic"); + StoredMessage msgStored = new StoredMessage("Hello".getBytes(UTF_8), MqttQoS.AT_LEAST_ONCE, "/topic"); msgStored.setClientID(TEST_CLIENT); // request a first ID diff --git a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java index b52d0b4f..a5ca3b67 100644 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java @@ -98,7 +98,8 @@ public class MapDBSessionsStoreTest { } @Test - public void testRetrieve1000ExpiredSessionPerformance() throws InterruptedException, ExecutionException, TimeoutException { + public void testRetrieve1000ExpiredSessionPerformance() throws InterruptedException, ExecutionException, + TimeoutException { MapDBSessionsStore sessionsStore = (MapDBSessionsStore) storageService.sessionsStore(); final ExecutorService executorService = Executors.newFixedThreadPool(2); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java index e1bf7ba7..52967171 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java @@ -21,19 +21,25 @@ import org.eclipse.paho.client.mqttv3.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.UnsupportedEncodingException; import java.util.concurrent.CountDownLatch; +import static java.nio.charset.StandardCharsets.UTF_8; + class BenchmarkPublisher { private static final Logger LOG = LoggerFactory.getLogger(BenchmarkPublisher.class); static class PublishCallback implements IMqttActionListener { + + @Override public void onFailure(IMqttToken asyncActionToken, Throwable exception) { - LOG.error("PUB: Publish failed: " + exception); + LOG.error("PUB: Publish failed", exception); exception.printStackTrace(); System.exit(2); } + @Override public void onSuccess(IMqttToken asyncActionToken) { } } @@ -44,12 +50,13 @@ class BenchmarkPublisher { private final long numToSend; private final CountDownLatch stopLatch; - ExitTopicCallback(long startedTime, long numToSend, IMqttAsyncClient connection, CountDownLatch stopLatch) { + ExitTopicCallback(long startedTime, long numToSend, CountDownLatch stopLatch) { this.startedTime = startedTime; this.numToSend = numToSend; this.stopLatch = stopLatch; } + @Override public void onSuccess(IMqttToken asyncActionToken) { long stopTime = System.currentTimeMillis(); long spentTime = stopTime - startedTime; @@ -65,6 +72,7 @@ class BenchmarkPublisher { this.stopLatch.countDown(); } + @Override public void onFailure(IMqttToken asyncActionToken, Throwable th) { LOG.error("PUB: Publish failed: " + th); System.exit(2); @@ -90,11 +98,14 @@ class BenchmarkPublisher { MqttConnectOptions connectOptions = new MqttConnectOptions(); connectOptions.setCleanSession(true); this.client.connect(connectOptions, null, new IMqttActionListener() { + + @Override public void onFailure(IMqttToken asyncActionToken, Throwable exception) { LOG.error("PUB: connect fail", exception); System.exit(2); } + @Override public void onSuccess(IMqttToken asyncActionToken) { LOG.info("PUB: Successfully connected to server"); } @@ -103,7 +114,7 @@ class BenchmarkPublisher { this.m_latch = new CountDownLatch(1); } - public void firePublishes() throws MqttException { + public void firePublishes() throws MqttException, UnsupportedEncodingException { long pauseMicroseconds = (int) ((1.0 / messagesPerSecond) * 1000 * 1000); LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds); @@ -116,14 +127,14 @@ class BenchmarkPublisher { IMqttActionListener pubCallback = new PublishCallback(); for (int i = 0; i < numToSend; i++) { long nanos = System.nanoTime(); - byte[] message = ("Hello world!!-" + nanos).getBytes(); + byte[] message = ("Hello world!!-" + nanos).getBytes(UTF_8); this.client.publish("/topic" + dialog_id, message, qos, retain, null, pubCallback); timer.sleep(pauseMicroseconds); } - IMqttActionListener exitCallback = new ExitTopicCallback(startTime, numToSend, this.client, m_latch); + IMqttActionListener exitCallback = new ExitTopicCallback(startTime, numToSend, m_latch); long nanosExit = System.nanoTime(); - byte[] exitMessage = ("Hello world!!-" + nanosExit).getBytes(); + byte[] exitMessage = ("Hello world!!-" + nanosExit).getBytes(UTF_8); this.client.publish("/exit" + dialog_id, exitMessage, qos, retain, null, exitCallback); } diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java index 25e35e41..dcbef3d7 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java @@ -23,12 +23,15 @@ import org.slf4j.LoggerFactory; import java.util.concurrent.CountDownLatch; +import static java.nio.charset.StandardCharsets.UTF_8; + class BenchmarkSubscriber { private static final Logger LOG = LoggerFactory.getLogger(BenchmarkSubscriber.class); class SubscriberCallback implements MqttCallback { + @Override public void messageArrived(String topic, MqttMessage message) throws Exception { //used to catch the first message published if (!alreadyStarted) { @@ -54,7 +57,7 @@ class BenchmarkSubscriber { histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000 m_latch.countDown(); } else { - String payload = new String(message.getPayload()); + String payload = new String(message.getPayload(), UTF_8); //long sentTime = Long.parseLong(payload.split("-")[1]); long sentTime = Long.parseLong(payload.split("-")[1]); long delay = System.nanoTime() - sentTime; @@ -67,6 +70,7 @@ class BenchmarkSubscriber { } } + @Override public void connectionLost(Throwable cause) { LOG.error("Client Subscriber lost connection, caused by ", cause); try { @@ -78,6 +82,7 @@ class BenchmarkSubscriber { m_latch.countDown(); } + @Override public void deliveryComplete(IMqttDeliveryToken token) { } } @@ -102,14 +107,18 @@ class BenchmarkSubscriber { this.client.setCallback(new SubscriberCallback()); this.client.connect(connectOptions, null, new IMqttActionListener() { + @Override public void onFailure(IMqttToken asyncActionToken, Throwable exception) { LOG.error("SUB: connect fail", exception); System.exit(2); } + @Override public void onSuccess(IMqttToken asyncActionToken) { try { client.subscribe("/topic" + dialog_id, 1, null, new IMqttActionListener() { + + @Override public void onFailure(IMqttToken asyncActionToken2, Throwable exception) { try { client.disconnect(); @@ -118,6 +127,7 @@ class BenchmarkSubscriber { } } + @Override public void onSuccess(IMqttToken asyncActionToken2) { LOG.info("SUB: subscribed to /topic{} qos: 1", dialog_id); } @@ -128,6 +138,8 @@ class BenchmarkSubscriber { try { client.subscribe("/exit" + dialog_id, 1, null, new IMqttActionListener() { + + @Override public void onFailure(IMqttToken asyncActionToken2, Throwable exception) { try { client.disconnect(); @@ -137,6 +149,7 @@ class BenchmarkSubscriber { m_latch.countDown(); } + @Override public void onSuccess(IMqttToken asyncActionToken2) { LOG.info("SUB: subscribed to /exit{} qos: 1", dialog_id); } diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java index 9896c79f..449d5dab 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java @@ -27,6 +27,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str; +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_LEAST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; @@ -65,7 +66,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { return; case DISCONNECT: LOG.info("Received a message of type {} from <{}>", messageType, clientID); - ctx.close(); + ctx.close().addListener(CLOSE_ON_FAILURE); // case PUBACK: // NettyChannel channel; // synchronized (m_channelMapper) { @@ -85,7 +86,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { false, 0); MqttMessage pingResp = new MqttMessage(pingHeader); - ctx.writeAndFlush(pingResp); + ctx.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE); break; default: LOG.info("Received a message of type {} from <{}>", messageType, clientID); @@ -114,7 +115,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(msg.variableHeader().topicName(), 0); MqttPublishMessage pubMessage = new MqttPublishMessage(fixedHeader, varHeader, msg.payload()); - m_state.getSubscriberCh().writeAndFlush(pubMessage); + m_state.getSubscriberCh().writeAndFlush(pubMessage).addListener(CLOSE_ON_FAILURE); /*Channel subscriberCh = m_state.getSubscriberCh(); if (subscriberCh.isWritable()) { subscriberCh.write(pubMessage); @@ -135,7 +136,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { fixedHeader, from(msg.variableHeader().messageId()), payload); - ctx.writeAndFlush(ackMessage); + ctx.writeAndFlush(ackMessage).addListener(CLOSE_ON_FAILURE); LOG.debug("subscribed client to {}", msg.payload().topicSubscriptions()); } @@ -156,8 +157,8 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader( MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false); MqttConnAckMessage koResp = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); - ctx.writeAndFlush(koResp); - ctx.close(); + ctx.writeAndFlush(koResp).addListener(CLOSE_ON_FAILURE); + ctx.close().addListener(CLOSE_ON_FAILURE); return; } @@ -166,7 +167,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader( MqttConnectReturnCode.CONNECTION_ACCEPTED, false); MqttConnAckMessage okResp = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); - ctx.writeAndFlush(okResp); + ctx.writeAndFlush(okResp).addListener(CLOSE_ON_FAILURE); } @Override @@ -175,7 +176,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { // NettyChannel channel = m_channelMapper.get(ctx); // String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); // m_messaging.lostConnection(channel, clientID); - ctx.channel().close(/*false*/); + ctx.channel().close().addListener(CLOSE_ON_FAILURE); System.out.println("Processing time histogram (microsecs)"); this.processingTime.outputPercentileDistribution(System.out, 1000.0); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java index 9424074e..0a39f5e7 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java @@ -27,6 +27,9 @@ import org.HdrHistogram.Histogram; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; +import static java.nio.charset.StandardCharsets.UTF_8; + @ChannelHandler.Sharable class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter { @@ -76,13 +79,13 @@ class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter { rawBytes = new byte[size]; content.getBytes(content.readerIndex(), rawBytes); } - return new String(rawBytes); + return new String(rawBytes, UTF_8); } @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { + public void channelInactive(ChannelHandlerContext ctx) { LOG.info("Received channel inactive"); - ctx.channel().close(); + ctx.channel().close().addListener(CLOSE_ON_FAILURE); System.out.println("Network time histogram (microsecs)"); this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java index d0371c4a..7860c534 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java @@ -32,6 +32,10 @@ import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.UnsupportedEncodingException; + +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + /** * Emulates a broker, but doesn't apply any protocol logic, just forward the qos0 publishes * to the lone subscriber, it's used just to measure the protocol decoding/encoding overhead. @@ -39,7 +43,7 @@ import org.slf4j.LoggerFactory; */ public class ProtocolDecodingServer { - class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { + static class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { @@ -47,7 +51,7 @@ public class ProtocolDecodingServer { if (e == IdleState.READER_IDLE) { //fire a channelInactive to trigger publish of Will ctx.fireChannelInactive(); - ctx.close(); + ctx.close().addListener(CLOSE_ON_FAILURE); } /*else if (e.getState() == IdleState.WRITER_IDLE) { ctx.writeAndFlush(new PingMessage()); }*/ @@ -125,12 +129,13 @@ public class ProtocolDecodingServer { // Bind and start to accept incoming connections. ChannelFuture f = b.bind(host, port); LOG.info("Server binded host: {}, port: {}", host, port); - f.sync(); + f.sync().addListener(CLOSE_ON_FAILURE); } catch (InterruptedException ex) { LOG.error(null, ex); } } + @SuppressWarnings("FutureReturnValueIgnored") public void stop() { if (m_workerGroup == null) { throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); @@ -145,7 +150,7 @@ public class ProtocolDecodingServer { System.out.println("Server stopped"); } - public static void main(String[] args) throws MqttException, InterruptedException { + public static void main(String[] args) throws MqttException, InterruptedException, UnsupportedEncodingException { final ProtocolDecodingServer server = new ProtocolDecodingServer(); server.init(); @@ -155,7 +160,7 @@ public class ProtocolDecodingServer { Runtime.getRuntime().addShutdownHook(new Thread(server::stop)); } - private static void startClientsTesting() throws MqttException, InterruptedException { + private static void startClientsTesting() throws MqttException, InterruptedException, UnsupportedEncodingException { String host = "localhost"; int numToSend = 10; int messagesPerSecond = 10000; diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java index 6d2ec1eb..228c77b4 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java @@ -27,6 +27,10 @@ import org.eclipse.paho.client.mqttv3.MqttException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.UnsupportedEncodingException; + +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + /** * Emulates a broker, but doesn't apply any protocol logic, just forward the qos0 publishes * to the lone subscriber, it's used just to measure the protocol decoding/encoding overhead. @@ -73,12 +77,13 @@ public class ProtocolPublishDecodingServer { // Bind and start to accept incoming connections. ChannelFuture f = b.bind(host, port); LOG.info("Server binded host: {}, port: {}", host, port); - f.sync(); + f.sync().addListener(CLOSE_ON_FAILURE); } catch (InterruptedException ex) { LOG.error(null, ex); } } + @SuppressWarnings("FutureReturnValueIgnored") public void stop() { if (m_workerGroup == null) { throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); @@ -93,7 +98,7 @@ public class ProtocolPublishDecodingServer { System.out.println("Server stopped"); } - public static void main(String[] args) throws MqttException, InterruptedException { + public static void main(String[] args) throws UnsupportedEncodingException { final ProtocolPublishDecodingServer server = new ProtocolPublishDecodingServer(); server.init(); @@ -107,7 +112,7 @@ public class ProtocolPublishDecodingServer { Runtime.getRuntime().addShutdownHook(new Thread(server::stop)); } - private static void publishBombing() { + private static void publishBombing() throws UnsupportedEncodingException { LOG.info("Started publish loop"); PublishBomber heavyPublisher = new PublishBomber("localhost", 1883); heavyPublisher.publishLoop(25000, 100000); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java b/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java index d703e0bd..59930b5f 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java @@ -24,10 +24,15 @@ import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.mqtt.*; +import org.apache.commons.codec.CharEncoding; import org.eclipse.jetty.toolchain.perf.PlatformTimer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.UnsupportedEncodingException; + +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + class PublishBomber { private static final Logger LOG = LoggerFactory.getLogger(PublishBomber.class); @@ -35,6 +40,7 @@ class PublishBomber { private final EventLoopGroup workerGroup; private Channel channel; + @SuppressWarnings("FutureReturnValueIgnored") PublishBomber(String host, int port) { workerGroup = new NioEventLoopGroup(); try { @@ -61,13 +67,13 @@ class PublishBomber { private void sendMessage(MqttMessage msg) { try { - channel.writeAndFlush(msg).await(); + channel.writeAndFlush(msg).await().addListener(CLOSE_ON_FAILURE); } catch (InterruptedException e) { e.printStackTrace(); } } - public void publishLoop(int messagesPerSecond, int numToSend) { + public void publishLoop(int messagesPerSecond, int numToSend) throws UnsupportedEncodingException { long pauseMicroseconds = (int) ((1.0 / messagesPerSecond) * 1000 * 1000); LOG.warn("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds); @@ -78,7 +84,7 @@ class PublishBomber { PlatformTimer timer = PlatformTimer.detect(); for (int i = 0; i < numToSend; i++) { long nanos = System.nanoTime(); - byte[] rawContent = ("Hello world!!-" + nanos).getBytes(); + byte[] rawContent = ("Hello world!!-" + nanos).getBytes(CharEncoding.UTF_8); ByteBuf payload = Unpooled.copiedBuffer(rawContent); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, @@ -94,7 +100,7 @@ class PublishBomber { public void disconnect() { try { - this.channel.disconnect().await(); + this.channel.disconnect().await().addListener(CLOSE_ON_FAILURE); } catch (InterruptedException e) { e.printStackTrace(); } diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java index c5feca3b..2cedc63b 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java @@ -27,12 +27,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; @ChannelHandler.Sharable class PublishReceiverHandler extends ChannelInboundHandlerAdapter { private static final Logger LOG = LoggerFactory.getLogger(PublishReceiverHandler.class); - Histogram forthNetworkTime = new Histogram(5); + private Histogram forthNetworkTime = new Histogram(5); PublishReceiverHandler() { } @@ -69,9 +70,9 @@ class PublishReceiverHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { + public void channelInactive(ChannelHandlerContext ctx) { LOG.info("Received channel inactive"); - ctx.channel().close(); + ctx.channel().close().addListener(CLOSE_ON_FAILURE); System.out.println("Network time histogram (microsecs)"); this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0); diff --git a/tools_scripts/integration/bunch_publish_issue12.groovy b/tools_scripts/integration/bunch_publish_issue12.groovy index 4770431b..a7fc365e 100644 --- a/tools_scripts/integration/bunch_publish_issue12.groovy +++ b/tools_scripts/integration/bunch_publish_issue12.groovy @@ -1,6 +1,5 @@ import org.eclipse.paho.client.mqttv3.MqttClient import org.eclipse.paho.client.mqttv3.MqttMessage -import org.eclipse.paho.client.mqttv3.MqttException import org.eclipse.paho.client.mqttv3.MqttDefaultFilePersistence String tmpDir = System.getProperty("java.io.tmpdir") @@ -21,4 +20,4 @@ println "connected" } client.disconnect() -println "disconnected" \ No newline at end of file +println "disconnected"