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

Minor, fixed checkstyle and find bugs warns

Browse Source
This commit is contained in:
andsel
2018-11-10 18:31:51 +01:00
parent 9ceb26307e
commit 284417ef50
34 changed files with 407 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/broker/Authorizator.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/IRetainedRepository.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/InflightResender.java
+2 -1
View File
@@ -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());
}
broker/src/main/java/io/moquette/broker/MQTTConnection.java
+26 -9
View File
@@ -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);
}
}
broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/broker/PostOffice.java
+19 -3
View File
@@ -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;
}
broker/src/main/java/io/moquette/broker/Server.java
+17 -1
View File
@@ -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<InterceptHandler> observers = new ArrayList<>(embeddedObservers);
String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
if (interceptorClassName != null && !interceptorClassName.isEmpty()) {
InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, io.moquette.broker.Server.class, this);
InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class,
io.moquette.broker.Server.class, this);
if (handler != null) {
observers.add(handler);
}
broker/src/main/java/io/moquette/broker/Session.java
+25 -6
View File
@@ -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);
broker/src/main/java/io/moquette/broker/SessionCorruptedException.java
+15
View File
@@ -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 {
broker/src/main/java/io/moquette/broker/SessionRegistry.java
+19 -3
View File
@@ -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<SessionRegistry.EnqueuedMessage> sessionQueue = queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>());
final Queue<SessionRegistry.EnqueuedMessage> sessionQueue =
queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>());
final boolean clean = msg.variableHeader().isCleanSession();
final Session newSession;
if (msg.variableHeader().isWillFlag()) {
broker/src/main/java/io/moquette/interception/InterceptHandler.java
+2 -6
View File
@@ -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();
broker/src/main/java/io/moquette/server/config/IConfig.java
+7
View File
@@ -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);
broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java
+2 -1
View File
@@ -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:
broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java
+2 -2
View File
@@ -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);
broker/src/main/java/io/moquette/spi/IMessagesStore.java
-1
View File
@@ -44,7 +44,6 @@ public interface IMessagesStore {
m_topic = topic;
}
public void setQos(MqttQoS qos) {
this.m_qos = qos;
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+1
View File
@@ -27,6 +27,7 @@ import java.util.Set;
/**
* Store used to handle the persistence of the subscriptions tree.
*/
@Deprecated
public interface ISessionsStore {
void initStore();
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+4 -1
View File
@@ -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<InterceptHandler> 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<InterceptHandler> handlers) {
this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers);
broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java
+5 -5
View File
@@ -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) {
broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java
+3
View File
@@ -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;
broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java
+3
View File
@@ -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);
broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java
+23 -3
View File
@@ -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<InterceptHandler> 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());
broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
+15
View File
@@ -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;
broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
+17 -1
View File
@@ -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();
broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
+17 -2
View File
@@ -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);
broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
+23 -7
View File
@@ -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);
broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
+17 -3
View File
@@ -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<Integer> 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);
}
broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
+17 -2
View File
@@ -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)
broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+17 -14
View File
@@ -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();
broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java
+2 -1
View File
@@ -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;
broker/src/test/java/io/moquette/testclient/Client.java
+2 -1
View File
@@ -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(