diff --git a/broker/src/main/java/io/moquette/broker/Authorizator.java b/broker/src/main/java/io/moquette/broker/Authorizator.java index ed7611e4..a832d7b8 100644 --- a/broker/src/main/java/io/moquette/broker/Authorizator.java +++ b/broker/src/main/java/io/moquette/broker/Authorizator.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.subscriptions.Topic; diff --git a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java index d090fa30..cdd442d8 100644 --- a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java +++ b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.BrokerConstants; diff --git a/broker/src/main/java/io/moquette/broker/IRetainedRepository.java b/broker/src/main/java/io/moquette/broker/IRetainedRepository.java index 499f67a5..92844b7e 100644 --- a/broker/src/main/java/io/moquette/broker/IRetainedRepository.java +++ b/broker/src/main/java/io/moquette/broker/IRetainedRepository.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.subscriptions.Topic; diff --git a/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java b/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java index 7257a865..f5948658 100644 --- a/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java +++ b/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.subscriptions.Subscription; diff --git a/broker/src/main/java/io/moquette/broker/InflightResender.java b/broker/src/main/java/io/moquette/broker/InflightResender.java index e7c9995d..6b89060e 100644 --- a/broker/src/main/java/io/moquette/broker/InflightResender.java +++ b/broker/src/main/java/io/moquette/broker/InflightResender.java @@ -155,7 +155,8 @@ public class InflightResender extends ChannelDuplexHandler { private void resendNotAcked(ChannelHandlerContext ctx/* , IdleStateEvent evt */) { if (LOG.isTraceEnabled()) { - LOG.trace("Flushing idle Netty channel {} for clientId: {}", ctx.channel(), NettyUtils.clientID(ctx.channel())); + LOG.trace("Flushing idle Netty channel {} for clientId: {}", ctx.channel(), + NettyUtils.clientID(ctx.channel())); } ctx.fireUserEventTriggered(new ResendNotAckedPublishes()); } diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java index bacd098d..1a053692 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.server.netty.NettyUtils; @@ -77,8 +92,8 @@ final class MQTTConnection { processPubAck(msg); break; case PINGREQ: - MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP,false, AT_MOST_ONCE, - false,0); + MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, AT_MOST_ONCE, + false, 0); MqttMessage pingResp = new MqttMessage(pingHeader); channel.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE); break; @@ -125,20 +140,22 @@ final class MQTTConnection { final boolean cleanSession = msg.variableHeader().isCleanSession(); if (clientId == null || clientId.length() == 0) { if (!brokerConfig.isAllowZeroByteClientId()) { - LOG.warn("Broker doesn't permit MQTT client ID empty. Username={}, channel: {}", username, channel); + LOG.warn("Broker doesn't permit MQTT empty client ID. Username: {}, channel: {}", username, channel); abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED); return; } if (!cleanSession) { - LOG.warn("MQTT client ID cannot be empty for persistent session. Username={}, channel: {}", username, channel); + LOG.warn("MQTT client ID cannot be empty for persistent session. Username: {}, channel: {}", + username, channel); abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED); return; } // Generating client id. clientId = UUID.randomUUID().toString().replace("-", ""); - LOG.debug("Client has connected with server generated id={}, username={}, channel: {}", clientId, username, channel); + LOG.debug("Client has connected with server generated id: {}, username: {}, channel: {}", clientId, + username, channel); } if (!login(msg, clientId)) { @@ -247,7 +264,7 @@ final class MQTTConnection { void sendConnAck(boolean isSessionAlreadyPresent) { connected = true; final MqttConnAckMessage ackMessage = connAck(CONNECTION_ACCEPTED, isSessionAlreadyPresent); - channel.writeAndFlush(ackMessage); + channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE); } boolean isConnected() { @@ -255,7 +272,7 @@ final class MQTTConnection { } void dropConnection() { - channel.close(); + channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE); } void processDisconnect(MqttMessage msg) { @@ -267,7 +284,7 @@ final class MQTTConnection { } sessionRegistry.disconnect(clientID); connected = false; - channel.close(); + channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE); LOG.trace("Processed DISCONNECT CId={}, channel: {}", clientID, channel); } @@ -377,7 +394,7 @@ final class MQTTConnection { LOG.debug("OUT {} on channel {}", msg.fixedHeader().messageType(), channel); } if (channel.isWritable()) { - channel.write(msg); + channel.write(msg).addListener(FIRE_EXCEPTION_ON_FAILURE); } } diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java b/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java index 53576584..d5fdb42e 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.security.IAuthenticator; diff --git a/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java b/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java index bec21ebc..a4b69682 100644 --- a/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java +++ b/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.subscriptions.Topic; diff --git a/broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java b/broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java index ced9d784..48106102 100644 --- a/broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java +++ b/broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.subscriptions.Subscription; diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index 60d4fdea..c4a70fbd 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/broker/src/main/java/io/moquette/broker/PostOffice.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.server.netty.NettyUtils; @@ -143,9 +158,10 @@ class PostOffice { mqttConnection.sendUnsubAckMessage(topics, clientID, messageId); } - void receivedPublishQos0(Topic topic, String username, String clientID, ByteBuf payload, boolean retain, MqttPublishMessage msg) { + void receivedPublishQos0(Topic topic, String username, String clientID, ByteBuf payload, boolean retain, + MqttPublishMessage msg) { if (!authorizator.canWrite(topic, username, clientID)) { - LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic: {}", clientID, topic); + LOG.error("MQTT client: {} is not authorized to publish on topic: {}", clientID, topic); return; } publish2Subscribers(payload, topic, AT_MOST_ONCE); @@ -169,7 +185,7 @@ class PostOffice { } final String clientId = connection.getClientId(); if (!authorizator.canWrite(topic, username, clientId)) { - LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic: {}", clientId, topic); + LOG.error("MQTT client: {} is not authorized to publish on topic: {}", clientId, topic); return; } diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index 68bb2806..c4d372c9 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.BrokerConstants; @@ -233,7 +248,8 @@ public class Server { List observers = new ArrayList<>(embeddedObservers); String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); if (interceptorClassName != null && !interceptorClassName.isEmpty()) { - InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, io.moquette.broker.Server.class, this); + InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, + io.moquette.broker.Server.class, this); if (handler != null) { observers.add(handler); } diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java index ec4e8f5a..52f7b8fb 100644 --- a/broker/src/main/java/io/moquette/broker/Session.java +++ b/broker/src/main/java/io/moquette/broker/Session.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.subscriptions.Subscription; @@ -185,8 +200,8 @@ class Session { drainQueueToConnection(); // TODO notify the interceptor -// final InterceptAcknowledgedMessage interceptAckMsg = new InterceptAcknowledgedMessage(inflightMsg, topic, -// username, messageID); +// final InterceptAcknowledgedMessage interceptAckMsg = new InterceptAcknowledgedMessage(inflightMsg, +// topic, username, messageID); // m_interceptor.notifyMessageAcknowledged(interceptAckMsg); } @@ -219,7 +234,8 @@ class Session { int packetId = mqttConnection.nextPacketId(); inflightWindow.put(packetId, new SessionRegistry.PublishedMessage(topic, qos, payload)); inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS)); - MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos, payload, packetId); + MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos, + payload, packetId); mqttConnection.sendPublish(publishMsg); // TODO drainQueueToConnection();? @@ -235,7 +251,8 @@ class Session { int packetId = mqttConnection.nextPacketId(); inflightWindow.put(packetId, new SessionRegistry.PublishedMessage(topic, qos, payload)); inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS)); - MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos, payload, packetId); + MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos, + payload, packetId); mqttConnection.sendPublish(publishMsg); drainQueueToConnection(); @@ -267,7 +284,8 @@ class Session { for (InFlightPacket notAckPacketId : expired) { if (inflightWindow.containsKey(notAckPacketId.packetId)) { - final SessionRegistry.PublishedMessage msg = (SessionRegistry.PublishedMessage) inflightWindow.get(notAckPacketId.packetId); + final SessionRegistry.PublishedMessage msg = + (SessionRegistry.PublishedMessage) inflightWindow.get(notAckPacketId.packetId); final Topic topic = msg.topic; final MqttQoS qos = msg.publishingQos; final ByteBuf payload = msg.payload; @@ -290,7 +308,8 @@ class Session { LOG.debug("Resending {} in flight packets [{}]", expired.size(), sb); } - private MqttPublishMessage publishNotRetainedDuplicated(InFlightPacket notAckPacketId, Topic topic, MqttQoS qos, ByteBuf payload) { + private MqttPublishMessage publishNotRetainedDuplicated(InFlightPacket notAckPacketId, Topic topic, MqttQoS qos, + ByteBuf payload) { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, qos, false, 0); MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic.toString(), notAckPacketId.packetId); return new MqttPublishMessage(fixedHeader, varHeader, payload); diff --git a/broker/src/main/java/io/moquette/broker/SessionCorruptedException.java b/broker/src/main/java/io/moquette/broker/SessionCorruptedException.java index cbbd877c..d2c1b71e 100644 --- a/broker/src/main/java/io/moquette/broker/SessionCorruptedException.java +++ b/broker/src/main/java/io/moquette/broker/SessionCorruptedException.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; public class SessionCorruptedException extends RuntimeException { diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java index e10a16ab..f4853b4d 100644 --- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java +++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.broker.Session.SessionStatus; @@ -19,7 +34,7 @@ import java.util.concurrent.ConcurrentMap; class SessionRegistry { - static abstract class EnqueuedMessage { + abstract static class EnqueuedMessage { } static class PublishedMessage extends EnqueuedMessage { @@ -75,7 +90,7 @@ class SessionRegistry { isSessionAlreadyStored = true; } final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession(); - boolean isSessionAlreadyPresent = (!msgCleanSessionFlag && isSessionAlreadyStored); + boolean isSessionAlreadyPresent = !msgCleanSessionFlag && isSessionAlreadyStored; mqttConnection.sendConnAck(isSessionAlreadyPresent); if (postConnectAction == PostConnectAction.SEND_STORED_MESSAGES) { @@ -163,7 +178,8 @@ class SessionRegistry { } private Session createNewSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) { - final Queue sessionQueue = queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>()); + final Queue sessionQueue = + queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>()); final boolean clean = msg.variableHeader().isCleanSession(); final Session newSession; if (msg.variableHeader().isWillFlag()) { diff --git a/broker/src/main/java/io/moquette/interception/InterceptHandler.java b/broker/src/main/java/io/moquette/interception/InterceptHandler.java index e06d54de..eec1de11 100644 --- a/broker/src/main/java/io/moquette/interception/InterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/InterceptHandler.java @@ -36,17 +36,13 @@ public interface InterceptHandler { InterceptUnsubscribeMessage.class, InterceptAcknowledgedMessage.class}; /** - * Returns the identifier of this intercept handler. - * - * @return + * @return the identifier of this intercept handler. */ String getID(); /** - * Returns the InterceptMessage subtypes that this handler can process. If the result is null or + * @return the InterceptMessage subtypes that this handler can process. If the result is null or * equal to ALL_MESSAGE_TYPES, all the message types will be processed. - * - * @return */ Class[] getInterceptedMessageTypes(); diff --git a/broker/src/main/java/io/moquette/server/config/IConfig.java b/broker/src/main/java/io/moquette/server/config/IConfig.java index 7fcf4f8a..d661a1f0 100644 --- a/broker/src/main/java/io/moquette/server/config/IConfig.java +++ b/broker/src/main/java/io/moquette/server/config/IConfig.java @@ -29,11 +29,18 @@ public abstract class IConfig { /** * Same semantic of Properties + * + * @param name property name. + * @return property value. * */ public abstract String getProperty(String name); /** * Same semantic of Properties + * + * @param name property name. + * @param defaultValue default value to return in case the property doesn't exists. + * @return property value. * */ public abstract String getProperty(String name, String defaultValue); diff --git a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java index 75af863c..57f1ec49 100644 --- a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java @@ -111,7 +111,8 @@ public class AutoFlushHandler extends ChannelDuplexHandler { // Avoid the case where destroy() is called before scheduling timeouts. // See: https://github.com/netty/netty/issues/143 if (LOG.isDebugEnabled()) { - LOG.debug("Initializing autoflush handler on channel {} Cid: {}", ctx.channel(), NettyUtils.clientID(ctx.channel())); + LOG.debug("Initializing autoflush handler on channel {} Cid: {}", ctx.channel(), + NettyUtils.clientID(ctx.channel())); } switch (state) { case 1: 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 e181ef59..b6d19b0c 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 @@ -60,7 +60,7 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { } MqttMessage msg = (MqttMessage) message; if (msg.fixedHeader() == null) { - throw new IOException("Unknown packet"); + throw new IOException("Unknown packet"); } String clientID = NettyUtils.clientID(ctx.channel()); MqttMessageType messageType = msg.fixedHeader().messageType(); @@ -69,7 +69,7 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { case PINGREQ: case PINGRESP: LOG.debug("{} {} <{}>", direction, messageType, clientID); - break; + break; case CONNECT: case DISCONNECT: LOG.info("{} {} <{}>", direction, messageType, clientID); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 209834a2..6812b62d 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -44,7 +44,6 @@ public interface IMessagesStore { m_topic = topic; } - public void setQos(MqttQoS qos) { this.m_qos = qos; } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index a50d94ee..8960f44c 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -27,6 +27,7 @@ import java.util.Set; /** * Store used to handle the persistence of the subscriptions tree. */ +@Deprecated public interface ISessionsStore { void initStore(); diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 766e43f6..8b4a0196 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -60,7 +60,7 @@ public final class BrokerInterceptor implements Interceptor { /** * Configures a broker interceptor, with a thread pool of one thread. * - * @param handlers + * @param handlers InterceptHandlers listeners. */ public BrokerInterceptor(List handlers) { this(1, handlers); @@ -68,6 +68,9 @@ public final class BrokerInterceptor implements Interceptor { /** * Configures a broker interceptor using the pool size specified in the IConfig argument. + * @param props configuration properties. + * @param handlers InterceptHandlers listeners. + * */ public BrokerInterceptor(IConfig props, List handlers) { this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); 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 e1c9f355..f1df5440 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 @@ -148,9 +148,9 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { * We roughly follow this theory above, but we allow CNode with no Subscriptions to linger (for now). * * - * @param inode - * @param iParent - * @return + * @param inode inode that handle to the tomb node. + * @param iParent inode parent. + * @return REPEAT if the this methods wasn't successful or OK. */ public Action cleanTomb(INode inode, INode iParent) { CNode updatedCnode = iParent.mainNode().copy(); @@ -225,8 +225,8 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { * Removes subscription from CTrie, adds TNode when the last client unsubscribes, then calls for cleanTomb in a * separate atomic CAS operation. * - * @param topic - * @param clientID + * @param topic the subscription's topic to remove. + * @param clientID the Id of client owning the subscription. */ @Override public void removeSubscription(Topic topic, String clientID) { diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java index 9f9d5026..7838d2eb 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java @@ -47,6 +47,9 @@ public final class Subscription implements Serializable { /** * Constructor with undefined maximum QoS + * + * @param clientId id of client owning this subscription. + * @param topicFilter the topic to subscribe to. * */ public Subscription(String clientId, Topic topicFilter) { this.requestedQos = null; 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 f3db6a8a..174ff794 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 @@ -41,6 +41,9 @@ public class Topic implements Serializable { /** * Factory method + * + * @param s the topic string (es "/a/b"). + * @return the created Topic instance. * */ public static Topic asTopic(String s) { return new Topic(s); diff --git a/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java b/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java index dd3dba1b..ff68523c 100644 --- a/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java +++ b/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java @@ -1,3 +1,19 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ + package io.moquette.broker; import io.moquette.interception.InterceptHandler; @@ -12,11 +28,13 @@ import java.util.List; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED; import static org.junit.Assert.*; -public class ConnectionTestUtils { +public final class ConnectionTestUtils { public static final List EMPTY_OBSERVERS = Collections.emptyList(); public static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); + private ConnectionTestUtils() { + } static void assertConnectAccepted(EmbeddedChannel channel) { MqttConnAckMessage connAck = channel.readOutbound(); @@ -29,14 +47,16 @@ public class ConnectionTestUtils { assertPublishIsCorrect(expectedTopic, expectedContent, receivedPublish); } - private static void assertPublishIsCorrect(String expectedTopic, String expectedContent, MqttPublishMessage receivedPublish) { + private static void assertPublishIsCorrect(String expectedTopic, String expectedContent, + MqttPublishMessage receivedPublish) { assertNotNull("Expecting a PUBLISH message", receivedPublish); final String decodedPayload = DebugUtils.payload2Str(receivedPublish.payload()); assertEquals(expectedContent, decodedPayload); assertEquals(expectedTopic, receivedPublish.variableHeader().topicName()); } - static void verifyReceiveRetainedPublish(EmbeddedChannel embeddedChannel, String expectedTopic, String expectedContent) { + static void verifyReceiveRetainedPublish(EmbeddedChannel embeddedChannel, String expectedTopic, + String expectedContent) { MqttPublishMessage receivedPublish = embeddedChannel.readOutbound(); assertPublishIsCorrect(expectedTopic, expectedContent, receivedPublish); assertTrue("MUST be retained publish", receivedPublish.fixedHeader().isRetain()); diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java index 3638c9f5..d7931bee 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.MockAuthenticator; diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java index 4a269246..99edc161 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.MockAuthenticator; @@ -46,7 +61,8 @@ public class MQTTConnectionPublishTest { } private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) { - IAuthenticator mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); + IAuthenticator mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), + singletonMap(TEST_USER, TEST_PWD)); ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java index 920d2058..eca80c22 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.MockAuthenticator; @@ -68,7 +83,8 @@ public class PostOfficeInternalPublishTest { subscriptions.init(subscriptionsRepository); retainedRepository = new MemoryRetainedRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return sessionRegistry; @@ -285,7 +301,6 @@ public class PostOfficeInternalPublishTest { assertEquals(expectedSubscription, onlyMatchedSubscription); } - private void verifyNoPublishIsReceived(EmbeddedChannel channel) { final Object messageReceived = channel.readOutbound(); assertNull("Received an out message from processor while not expected", messageReceived); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index 80a9948b..bd1790dc 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.MockAuthenticator; @@ -75,7 +90,8 @@ public class PostOfficePublishTest { subscriptions.init(subscriptionsRepository); retainedRepository = new MemoryRetainedRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return sessionRegistry; @@ -235,7 +251,7 @@ public class PostOfficePublishTest { // Exercise final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()); - sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1,true, + sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1, true, MqttMessageBuilders.publish() .payload(Unpooled.copiedBuffer("Any payload", Charset.defaultCharset())) .qos(MqttQoS.AT_LEAST_ONCE) @@ -279,7 +295,7 @@ public class PostOfficePublishTest { ConnectionTestUtils.assertConnectAccepted(pubChannel); final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()); - sut.receivedPublishQos1(pubConn, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1,true, + sut.receivedPublishQos1(pubConn, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1, true, MqttMessageBuilders.publish() .payload(anyPayload.retainedDuplicate()) .qos(MqttQoS.AT_LEAST_ONCE) @@ -317,7 +333,7 @@ public class PostOfficePublishTest { ConnectionTestUtils.assertConnectAccepted(pubChannel); final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()); - sut.receivedPublishQos1(pubConn, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1,true, + sut.receivedPublishQos1(pubConn, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1, true, MqttMessageBuilders.publish() .payload(anyPayload.retainedDuplicate()) .qos(MqttQoS.AT_LEAST_ONCE) @@ -342,7 +358,7 @@ public class PostOfficePublishTest { ConnectionTestUtils.assertConnectAccepted(pubChannel); final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()); - sut.receivedPublishQos1(pubConn, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1,true, + sut.receivedPublishQos1(pubConn, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1, true, MqttMessageBuilders.publish() .payload(anyPayload) .qos(MqttQoS.AT_LEAST_ONCE) @@ -370,8 +386,8 @@ public class PostOfficePublishTest { .retained(true) .topicName(NEWS_TOPIC) .build(); - sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1,true, - publishMsg); + sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1, true, + publishMsg); assertMessageIsRetained(NEWS_TOPIC); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java index 739d6c92..9e8ffc30 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.server.netty.NettyUtils; @@ -166,7 +181,6 @@ public class PostOfficeSubscribeTest { verifyFailureQos(subAckMsg); } - private void verifyFailureQos(MqttSubAckMessage subAckMsg) { List grantedQoSes = subAckMsg.payload().grantedQoSLevels(); assertEquals(1, grantedQoSes.size()); @@ -179,7 +193,7 @@ public class PostOfficeSubscribeTest { ConnectionTestUtils.assertConnectAccepted(channel); assertEquals("After CONNECT subscription MUST be empty", 0, subscriptions.size()); subscribe(channel, NEWS_TOPIC, AT_MOST_ONCE); - assertEquals("After /news subscribe, subscription MUST contain it",1, subscriptions.size()); + assertEquals("After /news subscribe, subscription MUST contain it", 1, subscriptions.size()); //Exercise & verify subscribe(channel, NEWS_TOPIC, AT_MOST_ONCE); @@ -199,7 +213,7 @@ public class PostOfficeSubscribeTest { this.sut.subscribeClientToTopics(subscribe, FAKE_CLIENT_ID, FAKE_USER_NAME, connection); MqttSubAckMessage subAckMsg = channel.readOutbound(); - assertEquals("Bad topic CAN'T add any subscription",0, subscriptions.size()); + assertEquals("Bad topic CAN'T add any subscription", 0, subscriptions.size()); verifyFailureQos(subAckMsg); } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java index b46005dd..c34a3c0e 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; import io.moquette.spi.impl.MockAuthenticator; @@ -229,7 +244,7 @@ public class PostOfficeUnsubscribeTest { // publish from another channel final ByteBuf anyPayload = Unpooled.copiedBuffer(firstPayload, Charset.defaultCharset()); - sut.receivedPublishQos1(anotherConnection, new Topic(topic), TEST_USER, anyPayload, 1,false, + sut.receivedPublishQos1(anotherConnection, new Topic(topic), TEST_USER, anyPayload, 1, false, MqttMessageBuilders.publish() .payload(Unpooled.copiedBuffer(firstPayload, Charset.defaultCharset())) .qos(MqttQoS.AT_LEAST_ONCE) @@ -319,7 +334,7 @@ public class PostOfficeUnsubscribeTest { private void publishQos1(MQTTConnection publisher, String topic, String payload, int messageID) { final ByteBuf bytePayload = Unpooled.copiedBuffer(payload, Charset.defaultCharset()); - sut.receivedPublishQos1(publisher, new Topic(topic), TEST_USER, bytePayload, messageID,false, + sut.receivedPublishQos1(publisher, new Topic(topic), TEST_USER, bytePayload, messageID, false, MqttMessageBuilders.publish() .payload(Unpooled.copiedBuffer(payload, Charset.defaultCharset())) .qos(MqttQoS.AT_LEAST_ONCE) diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java index 9e62768f..31ddbc9d 100644 --- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java +++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java @@ -1,8 +1,21 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ package io.moquette.broker; -import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.MockAuthenticator; import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy; import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory; @@ -22,23 +35,12 @@ import static java.util.Collections.singleton; import static java.util.Collections.singletonMap; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import static org.mockito.Matchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; public class SessionRegistryTest { static final String FAKE_CLIENT_ID = "FAKE_123"; - static final String FAKE_CLIENT_ID2 = "FAKE_456"; - static final String PUBLISHER_ID = "Publisher"; - private static final String SUBSCRIBER_ID = "Subscriber"; - static final String NEWS_TOPIC = "/news"; - static final String BAD_FORMATTED_TOPIC = "#MQTTClient"; - static final String TEST_USER = "fakeuser"; static final String TEST_PWD = "fakepwd"; - static final String EVIL_TEST_USER = "eviluser"; - static final String EVIL_TEST_PWD = "unsecret"; private MQTTConnection connection; private EmbeddedChannel channel; @@ -60,7 +62,8 @@ public class SessionRegistryTest { } private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) { - IAuthenticator mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); + IAuthenticator mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), + singletonMap(TEST_USER, TEST_PWD)); ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java index 1e39769e..3a360809 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java @@ -47,7 +47,8 @@ import static org.junit.Assert.*; public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { - private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.class); + private static final Logger LOG = + LoggerFactory.getLogger(ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.class); static MqttClientPersistence s_dataStore; static MqttClientPersistence s_pubDataStore; diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java index 4c9d443e..7e714226 100644 --- a/broker/src/test/java/io/moquette/testclient/Client.java +++ b/broker/src/test/java/io/moquette/testclient/Client.java @@ -26,6 +26,7 @@ import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.charset.Charset; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -109,7 +110,7 @@ public class Client { MqttConnectPayload mqttConnectPayload = new MqttConnectPayload( this.clientId, willTestamentTopic, - willTestamentMsg.getBytes(), + willTestamentMsg.getBytes(Charset.forName("UTF-8")), null, null); MqttConnectMessage connectMessage = new MqttConnectMessage(