editorconfig: max line length set

max line length warnings removed
missing Copyright added
useless publics removed
unmaintainable Javadoc tags: @author removed
helper class warnings removed
field default value initialization removed
Multiple warnings removed
pref: checkstyle warnings removed
Checkstyle will fail the build process if it detects violations
switch default cases error logger added
This commit is contained in:
Andreas Mager
2017-04-13 22:16:57 +02:00
parent 6dc762f47f
commit 9221a87812
122 changed files with 999 additions and 454 deletions
+1
View File
@@ -10,3 +10,4 @@ insert_final_newline = true
charset = utf-8 charset = utf-8
indent_style = space indent_style = space
indent_size = 4 indent_size = 4
max_line_length=120
@@ -18,10 +18,7 @@ package io.moquette;
import java.io.File; import java.io.File;
/** public final class BrokerConstants {
* Created by andrea on 08/12/15.
*/
public class BrokerConstants {
public static final String INTERCEPT_HANDLER_PROPERTY_NAME = "intercept.handler"; public static final String INTERCEPT_HANDLER_PROPERTY_NAME = "intercept.handler";
public static final String BROKER_INTERCEPTOR_THREAD_POOL_SIZE = "intercept.thread_pool.size"; public static final String BROKER_INTERCEPTOR_THREAD_POOL_SIZE = "intercept.thread_pool.size";
@@ -60,4 +57,7 @@ public class BrokerConstants {
public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive"; public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive";
public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds"; public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds";
public static final String STORAGE_CLASS_NAME = "storage_class"; public static final String STORAGE_CLASS_NAME = "storage_class";
private BrokerConstants() {
}
} }
@@ -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.connections; package io.moquette.connections;
@@ -5,9 +20,6 @@ import java.util.Collection;
/** /**
* This interface will be used by an external codebase to retrieve and close physical connections. * This interface will be used by an external codebase to retrieve and close physical connections.
*
* @author lbarrios
*
*/ */
public interface IConnectionsManager { public interface IConnectionsManager {
@@ -16,7 +28,7 @@ public interface IConnectionsManager {
* *
* @return * @return
*/ */
public int getActiveConnectionsNo(); int getActiveConnectionsNo();
/** /**
* Determines wether a MQTT client is connected to the broker. * Determines wether a MQTT client is connected to the broker.
@@ -24,14 +36,14 @@ public interface IConnectionsManager {
* @param clientID * @param clientID
* @return * @return
*/ */
public boolean isConnected(String clientID); boolean isConnected(String clientID);
/** /**
* Returns the identifiers of the MQTT clients that are connected to the broker. * Returns the identifiers of the MQTT clients that are connected to the broker.
* *
* @return * @return
*/ */
public Collection<String> getConnectedClientIds(); Collection<String> getConnectedClientIds();
/** /**
* Closes a physical connection. * Closes a physical connection.
@@ -41,7 +53,7 @@ public interface IConnectionsManager {
* If false, the connection will be flushed before it is closed. * If false, the connection will be flushed before it is closed.
* @return * @return
*/ */
public boolean closeConnection(String clientID, boolean closeImmediately); boolean closeConnection(String clientID, boolean closeImmediately);
/** /**
* Returns the state of the session of a given client. * Returns the state of the session of a given client.
@@ -49,12 +61,12 @@ public interface IConnectionsManager {
* @param clientID * @param clientID
* @return * @return
*/ */
public MqttSession getSessionStatus(String clientID); MqttSession getSessionStatus(String clientID);
/** /**
* Returns the state of all the sessions * Returns the state of all the sessions
* *
* @return * @return
*/ */
public Collection<MqttSession> getSessions(); Collection<MqttSession> getSessions();
} }
@@ -1,11 +1,23 @@
/*
* 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.connections; package io.moquette.connections;
/** /**
* A class that represents the metrics of a given MQTT connection. * A class that represents the metrics of a given MQTT connection.
*
* @author lbarrios
*
*/ */
public class MqttConnectionMetrics { public class MqttConnectionMetrics {
@@ -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.connections; package io.moquette.connections;
@@ -6,9 +21,6 @@ import java.util.Collection;
/** /**
* A class that represents the overall connection status of a MQTT session. Its instances will be * A class that represents the overall connection status of a MQTT session. Its instances will be
* used by an external codebase when the broker is configured in embedded mode. * used by an external codebase when the broker is configured in embedded mode.
*
* @author lbarrios
*
*/ */
public class MqttSession { public class MqttSession {
@@ -1,11 +1,23 @@
/*
* 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.connections; package io.moquette.connections;
/** /**
* A class that represents a MQTT subscription. * A class that represents a MQTT subscription.
*
* @author lbarrios
*
*/ */
public class MqttSubscription { public class MqttSubscription {
@@ -26,8 +26,6 @@ import io.moquette.interception.messages.InterceptUnsubscribeMessage;
/** /**
* Basic abstract class usefull to avoid empty methods creation in subclasses. * Basic abstract class usefull to avoid empty methods creation in subclasses.
*
* Created by andrea on 08/12/15.
*/ */
public abstract class AbstractInterceptHandler implements InterceptHandler { public abstract class AbstractInterceptHandler implements InterceptHandler {
@@ -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.interception; package io.moquette.interception;
@@ -10,9 +25,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import static io.moquette.spi.impl.Utils.readBytesAndRewind; import static io.moquette.spi.impl.Utils.readBytesAndRewind;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastInterceptHandler extends AbstractInterceptHandler { public class HazelcastInterceptHandler extends AbstractInterceptHandler {
private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class); private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class);
@@ -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.interception; package io.moquette.interception;
@@ -5,9 +20,6 @@ import io.moquette.interception.messages.InterceptPublishMessage;
import java.io.Serializable; import java.io.Serializable;
import static io.moquette.spi.impl.Utils.readBytesAndRewind; import static io.moquette.spi.impl.Utils.readBytesAndRewind;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastMsg implements Serializable { public class HazelcastMsg implements Serializable {
private static final long serialVersionUID = -1431584750134928273L; private static final long serialVersionUID = -1431584750134928273L;
@@ -28,8 +28,6 @@ import io.netty.handler.codec.mqtt.MqttMessage;
* Almost every method receives a subclass of {@link MqttMessage}, except <code>onDisconnect</code> * Almost every method receives a subclass of {@link MqttMessage}, except <code>onDisconnect</code>
* that receives the client id string and <code>onSubscribe</code> and <code>onUnsubscribe</code> * that receives the client id string and <code>onSubscribe</code> and <code>onUnsubscribe</code>
* that receive a {@link Subscription} object. * that receive a {@link Subscription} object.
*
* @author Wagner Macedo
*/ */
public interface InterceptHandler { public interface InterceptHandler {
@@ -30,7 +30,6 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage;
* normally a field. So, the implementations should act as a proxy to a custom intercept handler. * normally a field. So, the implementations should act as a proxy to a custom intercept handler.
* *
* @see InterceptHandler * @see InterceptHandler
* @author Wagner Macedo
*/ */
public interface Interceptor { public interface Interceptor {
@@ -40,11 +39,11 @@ public interface Interceptor {
void notifyClientConnectionLost(String clientID, String username); void notifyClientConnectionLost(String clientID, String username);
void notifyTopicPublished(MqttPublishMessage msg, String clientID, final String username); void notifyTopicPublished(MqttPublishMessage msg, String clientID, String username);
void notifyTopicSubscribed(Subscription sub, final String username); void notifyTopicSubscribed(Subscription sub, String username);
void notifyTopicUnsubscribed(String topic, String clientID, final String username); void notifyTopicUnsubscribed(String topic, String clientID, String username);
void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg); void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg);
@@ -19,9 +19,6 @@ package io.moquette.interception.messages;
import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.handler.codec.mqtt.MqttQoS;
/**
* @author Wagner Macedo
*/
public abstract class InterceptAbstractMessage implements InterceptMessage { public abstract class InterceptAbstractMessage implements InterceptMessage {
private final MqttMessage msg; private final MqttMessage msg;
@@ -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.interception.messages; package io.moquette.interception.messages;
@@ -5,7 +20,7 @@ import static io.moquette.spi.IMessagesStore.StoredMessage;
public class InterceptAcknowledgedMessage implements InterceptMessage { public class InterceptAcknowledgedMessage implements InterceptMessage {
final private StoredMessage msg; private final StoredMessage msg;
private final String username; private final String username;
private final String topic; private final String topic;
private final int packetID; private final int packetID;
@@ -1,11 +1,23 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
import io.netty.handler.codec.mqtt.MqttConnectMessage; import io.netty.handler.codec.mqtt.MqttConnectMessage;
/**
* @author Wagner Macedo
*/
public class InterceptConnectMessage extends InterceptAbstractMessage { public class InterceptConnectMessage extends InterceptAbstractMessage {
private final MqttConnectMessage msg; private final MqttConnectMessage msg;
@@ -1,9 +1,21 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptConnectionLostMessage implements InterceptMessage { public class InterceptConnectionLostMessage implements InterceptMessage {
private final String clientID; private final String clientID;
@@ -1,9 +1,21 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptDisconnectMessage implements InterceptMessage { public class InterceptDisconnectMessage implements InterceptMessage {
private final String clientID; private final String clientID;
@@ -1,11 +1,23 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
/** /**
* An interface that sets the root of the interceptor messages type hierarchy. * An interface that sets the root of the interceptor messages type hierarchy.
*
* @author lbarrios
*
*/ */
public interface InterceptMessage { public interface InterceptMessage {
@@ -1,12 +1,24 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage;
/**
* @author Wagner Macedo
*/
public class InterceptPublishMessage extends InterceptAbstractMessage { public class InterceptPublishMessage extends InterceptAbstractMessage {
private final MqttPublishMessage msg; private final MqttPublishMessage msg;
@@ -1,12 +1,24 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.handler.codec.mqtt.MqttQoS;
/**
* @author Wagner Macedo
*/
public class InterceptSubscribeMessage implements InterceptMessage { public class InterceptSubscribeMessage implements InterceptMessage {
private final Subscription subscription; private final Subscription subscription;
@@ -1,9 +1,21 @@
/*
* 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.interception.messages; package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptUnsubscribeMessage implements InterceptMessage { public class InterceptUnsubscribeMessage implements InterceptMessage {
private final String topicFilter; private final String topicFilter;
@@ -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.logging; package io.moquette.logging;
@@ -5,7 +20,7 @@ import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import io.moquette.interception.InterceptHandler; import io.moquette.interception.InterceptHandler;
public class LoggingUtils { public final class LoggingUtils {
public static <T extends InterceptHandler> Collection<String> getInterceptorIds(Collection<T> handlers) { public static <T extends InterceptHandler> Collection<String> getInterceptorIds(Collection<T> handlers) {
Collection<String> result = new ArrayList<>(handlers.size()); Collection<String> result = new ArrayList<>(handlers.size());
@@ -14,4 +29,7 @@ public class LoggingUtils {
} }
return result; return result;
} }
private LoggingUtils() {
}
} }
@@ -19,16 +19,11 @@ package io.moquette.persistence;
import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.MessageGUID; import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.Utils;
import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.impl.subscriptions.Topic;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.*; import java.util.*;
import static io.moquette.spi.impl.Utils.defaultGet;
/**
* @author andrea
*/
public class MemoryMessagesStore implements IMessagesStore { public class MemoryMessagesStore implements IMessagesStore {
private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class); private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class);
@@ -33,10 +33,6 @@ import java.util.concurrent.BlockingQueue;
import static io.moquette.spi.impl.Utils.defaultGet; import static io.moquette.spi.impl.Utils.defaultGet;
/**
*
* @author andrea
*/
public class MemorySessionStore implements ISessionsStore { public class MemorySessionStore implements ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
@@ -74,7 +70,6 @@ public class MemorySessionStore implements ISessionsStore {
@Override @Override
public void initStore() { public void initStore() {
} }
@Override @Override
@@ -32,9 +32,6 @@ import io.netty.channel.Channel;
/** /**
* Value object to maintain the information of single connection, like ClientID, Channel, and clean * Value object to maintain the information of single connection, like ClientID, Channel, and clean
* session flag. * session flag.
*
*
* @author andrea
*/ */
public class ConnectionDescriptor { public class ConnectionDescriptor {
@@ -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.server; package io.moquette.server;
@@ -19,10 +19,13 @@ package io.moquette.server;
/** /**
* Server constants keeper * Server constants keeper
*/ */
public class Constants { public final class Constants {
public static final String ATTR_CLIENTID = "ClientID"; public static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession"; public static final String CLEAN_SESSION = "cleanSession";
public static final String KEEP_ALIVE = "keepAlive"; public static final String KEEP_ALIVE = "keepAlive";
public static final int MAX_MESSAGE_QUEUE = 1024; // number of messages public static final int MAX_MESSAGE_QUEUE = 1024; // number of messages
private Constants() {
}
} }
@@ -33,8 +33,6 @@ import java.security.cert.CertificateException;
/** /**
* Moquette server implementation to load SSL certificate from local filesystem path configured in * Moquette server implementation to load SSL certificate from local filesystem path configured in
* config file. * config file.
*
* Created by andrea on 13/12/15.
*/ */
class DefaultMoquetteSslContextCreator implements ISslContextCreator { class DefaultMoquetteSslContextCreator implements ISslContextCreator {
@@ -42,7 +40,7 @@ class DefaultMoquetteSslContextCreator implements ISslContextCreator {
private IConfig props; private IConfig props;
public DefaultMoquetteSslContextCreator(IConfig props) { DefaultMoquetteSslContextCreator(IConfig props) {
this.props = props; this.props = props;
} }
@@ -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.server; package io.moquette.server;
@@ -10,9 +25,6 @@ import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastListener implements MessageListener<HazelcastMsg> { public class HazelcastListener implements MessageListener<HazelcastMsg> {
private static final Logger LOG = LoggerFactory.getLogger(HazelcastListener.class); private static final Logger LOG = LoggerFactory.getLogger(HazelcastListener.class);
@@ -52,8 +52,6 @@ import static io.moquette.logging.LoggingUtils.getInterceptorIds;
/** /**
* Launch a configured version of the server. * Launch a configured version of the server.
*
* @author andrea
*/ */
public class Server { public class Server {
@@ -336,7 +334,7 @@ public class Server {
public ProtocolProcessor getProcessor() { public ProtocolProcessor getProcessor() {
return m_processor; return m_processor;
} }
public ScheduledExecutorService getScheduler() { public ScheduledExecutorService getScheduler() {
return scheduler; return scheduler;
} }
@@ -21,10 +21,6 @@ import io.moquette.server.config.IConfig;
import io.moquette.spi.impl.ProtocolProcessor; import io.moquette.spi.impl.ProtocolProcessor;
import io.moquette.spi.security.ISslContextCreator; import io.moquette.spi.security.ISslContextCreator;
/**
*
* @author andrea
*/
public interface ServerAcceptor { public interface ServerAcceptor {
void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator) throws IOException; void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator) throws IOException;
@@ -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.server.config; package io.moquette.server.config;
@@ -32,8 +32,6 @@ import java.util.Properties;
* *
* A line that at the very first has # is a comment Each line has key value format, where the * A line that at the very first has # is a comment Each line has key value format, where the
* separator used it the space. * separator used it the space.
*
* @author andrea
*/ */
class ConfigurationParser { class ConfigurationParser {
@@ -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.server.config; package io.moquette.server.config;
@@ -20,8 +20,6 @@ import io.moquette.BrokerConstants;
/** /**
* Base interface for all configuration implementations (filesystem, memory or classpath) * Base interface for all configuration implementations (filesystem, memory or classpath)
*
* @author andrea
*/ */
public abstract class IConfig { public abstract class IConfig {
@@ -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.server.config; package io.moquette.server.config;
@@ -21,8 +21,6 @@ import java.util.Properties;
/** /**
* Configuration backed by memory. * Configuration backed by memory.
*
* @author andrea
*/ */
public class MemoryConfig extends IConfig { public class MemoryConfig extends IConfig {
@@ -21,15 +21,15 @@ import io.netty.handler.codec.mqtt.*;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
public class MessageBuilder { public final class MessageBuilder {
public static class PublishBuilder { public static class PublishBuilder {
private String topic = null; private String topic;
private boolean retained = false; private boolean retained;
private MqttQoS qos; private MqttQoS qos;
private byte[] payload; private byte[] payload;
private int messageId = 0; private int messageId;
public PublishBuilder topicName(String topic) { public PublishBuilder topicName(String topic) {
this.topic = topic; this.topic = topic;
@@ -67,16 +67,16 @@ public class MessageBuilder {
private MqttVersion version = MqttVersion.MQTT_3_1_1; private MqttVersion version = MqttVersion.MQTT_3_1_1;
private String clientId; private String clientId;
private boolean cleanSession = false; private boolean cleanSession;
private boolean hasUser = false; private boolean hasUser;
private boolean hasPassword = false; private boolean hasPassword;
private int keepAliveSecs; private int keepAliveSecs;
private boolean willFlag = false; private boolean willFlag;
private MqttQoS willQos = MqttQoS.AT_MOST_ONCE; private MqttQoS willQos = MqttQoS.AT_MOST_ONCE;
private String willTopic = null; private String willTopic;
private String willMessage = null; private String willMessage;
private String username = null; private String username;
private String password = null; private String password;
public ConnectBuilder protocolVersion(MqttVersion version) { public ConnectBuilder protocolVersion(MqttVersion version) {
this.version = version; this.version = version;
@@ -196,7 +196,7 @@ public class MessageBuilder {
public static class UnsubscribeBuilder { public static class UnsubscribeBuilder {
private List<String> topicFilters = new ArrayList<>(); private List<String> topicFilters = new ArrayList<>();
private int messageId = 0; private int messageId;
public UnsubscribeBuilder addTopicFilter(String topic) { public UnsubscribeBuilder addTopicFilter(String topic) {
topicFilters.add(topic); topicFilters.add(topic);
@@ -236,4 +236,7 @@ public class MessageBuilder {
public static UnsubscribeBuilder unsubscribe() { public static UnsubscribeBuilder unsubscribe() {
return new UnsubscribeBuilder(); return new UnsubscribeBuilder();
} }
private MessageBuilder() {
}
} }
@@ -50,10 +50,6 @@ import java.util.List;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static io.moquette.BrokerConstants.*; import static io.moquette.BrokerConstants.*;
/**
*
* @author andrea
*/
public class NettyAcceptor implements ServerAcceptor { public class NettyAcceptor implements ServerAcceptor {
private static final String MQTT_SUBPROTOCOL_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1"; private static final String MQTT_SUBPROTOCOL_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1";
@@ -25,10 +25,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
/**
*
* @author andrea
*/
@Sharable @Sharable
public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
@@ -83,6 +79,9 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
MqttMessage pingResp = new MqttMessage(pingHeader); MqttMessage pingResp = new MqttMessage(pingHeader);
ctx.writeAndFlush(pingResp); ctx.writeAndFlush(pingResp);
break; break;
default:
LOG.error("Unkonwn MessageType:{}", messageType);
break;
} }
} catch (Throwable ex) { } 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);
@@ -24,10 +24,8 @@ import io.netty.util.AttributeKey;
/** /**
* Some Netty's channels utilities. * Some Netty's channels utilities.
*
* @author andrea
*/ */
public class NettyUtils { public final class NettyUtils {
public static final String ATTR_USERNAME = "username"; public static final String ATTR_USERNAME = "username";
public static final String ATTR_SESSION_STOLEN = "sessionStolen"; public static final String ATTR_SESSION_STOLEN = "sessionStolen";
@@ -70,4 +68,7 @@ public class NettyUtils {
public static String userName(Channel channel) { public static String userName(Channel channel) {
return (String) channel.attr(NettyUtils.ATTR_KEY_USERNAME).get(); return (String) channel.attr(NettyUtils.ATTR_KEY_USERNAME).get();
} }
private NettyUtils() {
}
} }
@@ -18,8 +18,8 @@ package io.moquette.server.netty.metrics;
public class BytesMetrics { public class BytesMetrics {
private long m_readBytes = 0; private long m_readBytes;
private long m_wroteBytes = 0; private long m_wroteBytes;
void incrementRead(long numBytes) { void incrementRead(long numBytes) {
m_readBytes += numBytes; m_readBytes += numBytes;
@@ -26,10 +26,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import static io.moquette.spi.impl.Utils.messageId; import static io.moquette.spi.impl.Utils.messageId;
/**
*
* @author andrea
*/
@Sharable @Sharable
public class MQTTMessageLogger extends ChannelDuplexHandler { public class MQTTMessageLogger extends ChannelDuplexHandler {
@@ -80,6 +76,9 @@ public class MQTTMessageLogger extends ChannelDuplexHandler {
case PUBACK: case PUBACK:
LOG.info("{} PUBACK <{}> packetID <{}>", direction, clientID, messageId(msg)); LOG.info("{} PUBACK <{}> packetID <{}>", direction, clientID, messageId(msg));
break; break;
default:
LOG.error("Unkonwn MessageType:{}", messageType);
break;
} }
} }
@@ -18,8 +18,8 @@ package io.moquette.server.netty.metrics;
public class MessageMetrics { public class MessageMetrics {
private long m_messagesRead = 0; private long m_messagesRead;
private long m_messageWrote = 0; private long m_messageWrote;
void incrementRead(long numMessages) { void incrementRead(long numMessages) {
m_messagesRead += numMessages; m_messagesRead += numMessages;
@@ -38,8 +38,6 @@ import java.util.concurrent.BlockingQueue;
* completely acknowledged.</li> * completely acknowledged.</li>
* <li>Optionally, QoS 0 messages pending transmission to the Client.</li> * <li>Optionally, QoS 0 messages pending transmission to the Client.</li>
* </ul> * </ul>
*
* @author andrea
*/ */
public class ClientSession { public class ClientSession {
@@ -76,7 +74,7 @@ public class ClientSession {
} }
} }
private final static Logger LOG = LoggerFactory.getLogger(ClientSession.class); private static final Logger LOG = LoggerFactory.getLogger(ClientSession.class);
public final String clientID; public final String clientID;
@@ -18,8 +18,6 @@ package io.moquette.spi;
import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.impl.subscriptions.Topic;
/**
*/
public interface IMatchingCondition { public interface IMatchingCondition {
boolean match(Topic key); boolean match(Topic key);
@@ -20,8 +20,6 @@ package io.moquette.spi;
* Callback interface used to be notified of some events from the input event queue. * Callback interface used to be notified of some events from the input event queue.
* *
* It's the abstraction of the messaging stuff attached in after the front protocol parsing stuff. * It's the abstraction of the messaging stuff attached in after the front protocol parsing stuff.
*
* @author andrea
*/ */
public interface IMessaging { public interface IMessaging {
} }
@@ -26,8 +26,6 @@ import java.util.concurrent.BlockingQueue;
/** /**
* Store used to handle the persistence of the subscriptions tree. * Store used to handle the persistence of the subscriptions tree.
*
* @author andrea
*/ */
public interface ISessionsStore { public interface ISessionsStore {
@@ -66,7 +64,6 @@ public interface ISessionsStore {
public String toString() { public String toString() {
return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}'; return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}';
} }
} }
void initStore(); void initStore();
@@ -1,5 +1,20 @@
package io.moquette.spi; /*
* 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;
public interface IStore { public interface IStore {
@@ -8,5 +23,4 @@ public interface IStore {
ISessionsStore sessionsStore(); ISessionsStore sessionsStore();
void close(); void close();
} }
@@ -37,8 +37,6 @@ import static io.moquette.logging.LoggingUtils.getInterceptorIds;
/** /**
* An interceptor that execute the interception tasks asynchronously. * An interceptor that execute the interception tasks asynchronously.
*
* @author Wagner Macedo
*/ */
final class BrokerInterceptor implements Interceptor { final class BrokerInterceptor implements Interceptor {
@@ -63,14 +61,14 @@ final class BrokerInterceptor implements Interceptor {
* *
* @param handlers * @param handlers
*/ */
public BrokerInterceptor(List<InterceptHandler> handlers) { BrokerInterceptor(List<InterceptHandler> handlers) {
this(1, handlers); this(1, handlers);
} }
/** /**
* Configures a broker interceptor using the pool size specified in the IConfig argument. * Configures a broker interceptor using the pool size specified in the IConfig argument.
*/ */
public BrokerInterceptor(IConfig props, List<InterceptHandler> handlers) { BrokerInterceptor(IConfig props, List<InterceptHandler> handlers) {
this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers);
} }
@@ -201,7 +199,7 @@ final class BrokerInterceptor implements Interceptor {
@Override @Override
public void addInterceptHandler(InterceptHandler interceptHandler) { public void addInterceptHandler(InterceptHandler interceptHandler) {
Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
LOG.info( "Adding MQTT message interceptor. InterceptorId={}, handledMessageTypes={}", LOG.info("Adding MQTT message interceptor. InterceptorId={}, handledMessageTypes={}",
interceptHandler.getID(), interceptedMessageTypes); interceptHandler.getID(), interceptedMessageTypes);
for (Class<?> interceptMessageType : interceptedMessageTypes) { for (Class<?> interceptMessageType : interceptedMessageTypes) {
this.handlers.get(interceptMessageType).add(interceptHandler); this.handlers.get(interceptMessageType).add(interceptHandler);
@@ -17,15 +17,13 @@
package io.moquette.spi.impl; package io.moquette.spi.impl;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import java.nio.ByteBuffer;
/** final class DebugUtils {
*
* @author andrea
*/
class DebugUtils {
static String payload2Str(ByteBuf content) { static String payload2Str(ByteBuf content) {
return new String(content.copy ().array()); return new String(content.copy().array());
}
private DebugUtils() {
} }
} }
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -22,7 +37,7 @@ class MessagesPublisher {
private final IMessagesStore m_messagesStore; private final IMessagesStore m_messagesStore;
private final PersistentQueueMessageSender messageSender; private final PersistentQueueMessageSender messageSender;
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) { IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) {
this.connectionDescriptors = connectionDescriptors; this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore; this.m_sessionsStore = sessionsStore;
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -15,7 +30,7 @@ class PersistentQueueMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class); private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
private final ConnectionDescriptorStore connectionDescriptorStore; private final ConnectionDescriptorStore connectionDescriptorStore;
public PersistentQueueMessageSender(ConnectionDescriptorStore connectionDescriptorStore) { PersistentQueueMessageSender(ConnectionDescriptorStore connectionDescriptorStore) {
this.connectionDescriptorStore = connectionDescriptorStore; this.connectionDescriptorStore = connectionDescriptorStore;
} }
@@ -24,7 +39,7 @@ class PersistentQueueMessageSender {
final int messageId = pubMessage.variableHeader().messageId(); final int messageId = pubMessage.variableHeader().messageId();
final String topicName = pubMessage.variableHeader().topicName(); final String topicName = pubMessage.variableHeader().topicName();
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId, LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId,
clientId, topicName, DebugUtils.payload2Str(pubMessage.payload())); clientId, topicName, DebugUtils.payload2Str(pubMessage.payload()));
} else { } else {
LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName); LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName);
@@ -34,8 +49,8 @@ class PersistentQueueMessageSender {
MqttQoS qos = pubMessage.fixedHeader().qosLevel(); MqttQoS qos = pubMessage.fixedHeader().qosLevel();
if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) { if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {
LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, qos={}, " + LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, "
"cleanSession={}", messageId, clientId, topicName, qos, false); + "qos={}, cleanSession={}", messageId, clientId, topicName, qos, false);
clientsession.enqueue(asStoredMessage(pubMessage)); clientsession.enqueue(asStoredMessage(pubMessage));
} else { } else {
LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " + LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " +
@@ -56,8 +56,6 @@ import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
* execution. * execution.
* *
* Used by the front facing class ProtocolProcessorBootstrapper. * Used by the front facing class ProtocolProcessorBootstrapper.
*
* @author andrea
*/ */
public class ProtocolProcessor { public class ProtocolProcessor {
@@ -68,7 +66,7 @@ public class ProtocolProcessor {
private final boolean retained; private final boolean retained;
private final MqttQoS qos; private final MqttQoS qos;
public WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) { WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) {
this.topic = topic; this.topic = topic;
this.payload = payload; this.payload = payload;
this.retained = retained; this.retained = retained;
@@ -474,7 +472,8 @@ public class ProtocolProcessor {
targetSession.inFlightAcknowledged(messageID); targetSession.inFlightAcknowledged(messageID);
String topic = inflightMsg.getTopic(); String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID)); m_interceptor
.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID));
} }
public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) { public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) {
@@ -511,6 +510,7 @@ public class ProtocolProcessor {
this.qos2PublishHandler.receivedPublishQos2(channel, msg); this.qos2PublishHandler.receivedPublishQos2(channel, msg);
break; break;
default: default:
LOG.error("Unkonwn QoS-Type:{}", qos);
break; break;
} }
} }
@@ -618,7 +618,8 @@ public class ProtocolProcessor {
StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID);
String username = NettyUtils.userName(channel); String username = NettyUtils.userName(channel);
String topic = inflightMsg.getTopic(); String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID)); m_interceptor
.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID));
} }
public void processDisconnect(Channel channel) throws InterruptedException { public void processDisconnect(Channel channel) throws InterruptedException {
@@ -921,7 +922,7 @@ public class ProtocolProcessor {
} else { } else {
// recreate a publish from stored publish in queue // recreate a publish from stored publish in queue
boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null; boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null;
MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQos(), msg.getMessage(), MqttPublishMessage pubMsg = createPublishForQos(msg.getTopic(), msg.getQos(), msg.getMessage(),
retained, 0); retained, 0);
channel.write(pubMsg); channel.write(pubMsg);
} }
@@ -936,11 +937,11 @@ public class ProtocolProcessor {
public void removeInterceptHandler(InterceptHandler interceptHandler) { public void removeInterceptHandler(InterceptHandler interceptHandler) {
this.m_interceptor.removeInterceptHandler(interceptHandler); this.m_interceptor.removeInterceptHandler(interceptHandler);
} }
public IMessagesStore getMessagesStore() { public IMessagesStore getMessagesStore() {
return m_messagesStore; return m_messagesStore;
} }
public ISessionsStore getSessionsStore() { public ISessionsStore getSessionsStore() {
return m_sessionsStore; return m_sessionsStore;
} }
@@ -44,8 +44,6 @@ import java.util.List;
/** /**
* It's main responsibility is bootstrap the ProtocolProcessor. * It's main responsibility is bootstrap the ProtocolProcessor.
*
* @author andrea
*/ */
public class ProtocolProcessorBootstrapper { public class ProtocolProcessorBootstrapper {
@@ -56,7 +54,7 @@ public class ProtocolProcessorBootstrapper {
private ISessionsStore m_sessionsStore; private ISessionsStore m_sessionsStore;
private Runnable storeShutdown = null; private Runnable storeShutdown;
private BrokerInterceptor m_interceptor; private BrokerInterceptor m_interceptor;
@@ -155,7 +153,7 @@ public class ProtocolProcessorBootstrapper {
LOG.info("Parsing ACL file. Path = {}", aclFilePath); LOG.info("Parsing ACL file. Path = {}", aclFilePath);
authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath)); authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
} catch (ParseException pex) { } catch (ParseException pex) {
LOG.error("Unable to parse ACL file. path="+ aclFilePath, pex); LOG.error("Unable to parse ACL file. path=" + aclFilePath, pex);
} }
} else { } else {
authorizator = new PermitAllAuthorizator(); authorizator = new PermitAllAuthorizator();
@@ -167,8 +165,10 @@ public class ProtocolProcessorBootstrapper {
connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore); connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore);
LOG.info("Initializing MQTT protocol processor..."); LOG.info("Initializing MQTT protocol processor...");
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); boolean allowAnonymous = Boolean
boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); .parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
boolean allowZeroByteClientId = Boolean
.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator, m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor,
props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
@@ -185,11 +185,19 @@ public class ProtocolProcessorBootstrapper {
// check if method getInstance exists // check if method getInstance exists
Method method = clazz.getMethod("getInstance", new Class[]{}); Method method = clazz.getMethod("getInstance", new Class[]{});
try { try {
LOG.info("Invoking getInstance() method. ClassName = {}, interfaceName = {}.", className, intrface.getName()); LOG.info(
"Invoking getInstance() method. ClassName = {}, interfaceName = {}.",
className,
intrface.getName());
instance = (T) method.invoke(null, new Object[]{}); instance = (T) method.invoke(null, new Object[]{});
} catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) { } catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) {
LOG.error("Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", LOG.error(
className, intrface.getName(), ex.getCause(), ex.getMessage()); "Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, "
+ "errorMessage = {}.",
className,
intrface.getName(),
ex.getCause(),
ex.getMessage());
return null; return null;
} }
} catch (NoSuchMethodException nsmex) { } catch (NoSuchMethodException nsmex) {
@@ -204,17 +212,31 @@ public class ProtocolProcessorBootstrapper {
.getConstructor(constructorArgClass) .getConstructor(constructorArgClass)
.newInstance(props); .newInstance(props);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.warn("Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", LOG.warn(
constructorArgClass.getName(), className, intrface.getName(), ex.getCause(), ex.getMessage()); "Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}"
+ ", errorMessage = {}.",
constructorArgClass.getName(),
className,
intrface.getName(),
ex.getCause(),
ex.getMessage());
return null; return null;
} catch (NoSuchMethodException | InvocationTargetException e) { } catch (NoSuchMethodException | InvocationTargetException e) {
try { try {
LOG.info("Invoking default constructor. ClassName = {}, interfaceName = {}.", className, intrface.getName()); LOG.info(
"Invoking default constructor. ClassName = {}, interfaceName = {}.",
className,
intrface.getName());
// fallback to default constructor // fallback to default constructor
instance = this.getClass().getClassLoader().loadClass(className).asSubclass(intrface).newInstance(); instance = this.getClass().getClassLoader().loadClass(className).asSubclass(intrface).newInstance();
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.error("Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", LOG.error(
className, intrface.getName(), ex.getCause(), ex.getMessage()); "Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, "
+ "errorMessage = {}.",
className,
intrface.getName(),
ex.getCause(),
ex.getMessage());
return null; return null;
} }
} }
@@ -222,8 +244,13 @@ public class ProtocolProcessorBootstrapper {
LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, intrface.getName()); LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, intrface.getName());
return null; return null;
} catch (SecurityException ex) { } catch (SecurityException ex) {
LOG.error("Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", LOG.error(
className, intrface.getName(), ex.getCause(), ex.getMessage()); "Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, "
+ "errorMessage = {}.",
className,
intrface.getName(),
ex.getCause(),
ex.getMessage());
return null; return null;
} }
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -23,8 +38,8 @@ class Qos0PublishHandler extends QosPublishHandler {
private final BrokerInterceptor m_interceptor; private final BrokerInterceptor m_interceptor;
private final MessagesPublisher publisher; private final MessagesPublisher publisher;
public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore,
IMessagesStore messagesStore, BrokerInterceptor interceptor, MessagesPublisher messagesPublisher) { BrokerInterceptor interceptor, MessagesPublisher messagesPublisher) {
super(authorizator); super(authorizator);
this.subscriptions = subscriptions; this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore; this.m_messagesStore = messagesStore;
@@ -1,10 +1,24 @@
/*
* 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; package io.moquette.spi.impl;
import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils; import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.impl.subscriptions.Topic;
@@ -31,9 +45,9 @@ class Qos1PublishHandler extends QosPublishHandler {
private final ConnectionDescriptorStore connectionDescriptors; private final ConnectionDescriptorStore connectionDescriptors;
private final MessagesPublisher publisher; private final MessagesPublisher publisher;
public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore,
IMessagesStore messagesStore, BrokerInterceptor interceptor, BrokerInterceptor interceptor, ConnectionDescriptorStore connectionDescriptors, String brokerPort,
ConnectionDescriptorStore connectionDescriptors, String brokerPort, MessagesPublisher messagesPublisher) { MessagesPublisher messagesPublisher) {
super(authorizator); super(authorizator);
this.subscriptions = subscriptions; this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore; this.m_messagesStore = messagesStore;
@@ -97,7 +111,7 @@ class Qos1PublishHandler extends QosPublishHandler {
LOG.debug("clientIDs are {}", connectionDescriptors); LOG.debug("clientIDs are {}", connectionDescriptors);
if (!connectionDescriptors.isConnected(clientId)) { if (!connectionDescriptors.isConnected(clientId)) {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s",
clientId,connectionDescriptors)); clientId, connectionDescriptors));
} }
connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId); connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId);
} catch (Throwable t) { } catch (Throwable t) {
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -34,7 +49,7 @@ class Qos2PublishHandler extends QosPublishHandler {
private final ISessionsStore m_sessionsStore; private final ISessionsStore m_sessionsStore;
private final MessagesPublisher publisher; private final MessagesPublisher publisher;
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor, IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort,
MessagesPublisher messagesPublisher) { MessagesPublisher messagesPublisher) {
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -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.spi.impl; package io.moquette.spi.impl;
@@ -9,7 +24,7 @@ import java.util.Map;
/** /**
* Utility static methods, like Map get with default value, or elvis operator. * Utility static methods, like Map get with default value, or elvis operator.
*/ */
public class Utils { public final class Utils {
public static <T, K> T defaultGet(Map<K, T> map, K key, T defaultValue) { public static <T, K> T defaultGet(Map<K, T> map, K key, T defaultValue) {
T value = map.get(key); T value = map.get(key);
@@ -30,4 +45,7 @@ public class Utils {
payload.readerIndex(mark); payload.readerIndex(mark);
return payloadContent; return payloadContent;
} }
private Utils() {
}
} }
@@ -24,10 +24,8 @@ import java.text.ParseException;
/** /**
* Parses the acl configuration file. If a line starts with # it's comment. Blank lines are skipped. * 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}" * The format is "topic [read|write|readwrite] {topic name}"
*
* @author andrea
*/ */
public class ACLFileParser { public final class ACLFileParser {
private static final Logger LOG = LoggerFactory.getLogger(ACLFileParser.class); private static final Logger LOG = LoggerFactory.getLogger(ACLFileParser.class);
@@ -110,4 +108,7 @@ public class ACLFileParser {
} }
return collector; return collector;
} }
private ACLFileParser() {
}
} }
@@ -1,11 +1,23 @@
/*
* 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.security; package io.moquette.spi.impl.security;
import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthenticator;
/**
* Created by andrea on 8/23/14.
*/
public class AcceptAllAuthenticator implements IAuthenticator { public class AcceptAllAuthenticator implements IAuthenticator {
public boolean checkValid(String clientId, String username, byte[] password) { public boolean checkValid(String clientId, String username, byte[] password) {
@@ -21,8 +21,6 @@ import io.moquette.spi.impl.subscriptions.Topic;
/** /**
* Carries the read/write authorization to topics for the users. * Carries the read/write authorization to topics for the users.
*
* @author andrea
*/ */
public class Authorization { public class Authorization {
@@ -18,8 +18,6 @@ package io.moquette.spi.impl.security;
import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator; import io.moquette.spi.security.IAuthorizator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException; import java.text.ParseException;
import java.util.*; import java.util.*;
@@ -28,18 +26,14 @@ import java.util.*;
* mode to check it topics matches the ACLs. * mode to check it topics matches the ACLs.
* *
* Not thread safe. * Not thread safe.
*
* @author andrea
*/ */
class AuthorizationsCollector implements IAuthorizator { class AuthorizationsCollector implements IAuthorizator {
private static final Logger LOG = LoggerFactory.getLogger(AuthorizationsCollector.class);
private List<Authorization> m_globalAuthorizations = new ArrayList<>(); private List<Authorization> m_globalAuthorizations = new ArrayList<>();
private List<Authorization> m_patternAuthorizations = new ArrayList<>(); private List<Authorization> m_patternAuthorizations = new ArrayList<>();
private Map<String, List<Authorization>> m_userAuthorizations = new HashMap<>(); private Map<String, List<Authorization>> m_userAuthorizations = new HashMap<>();
private boolean m_parsingUsersSpecificSection = false; private boolean m_parsingUsersSpecificSection;
private boolean m_parsingPatternSpecificSection = false; private boolean m_parsingPatternSpecificSection;
private String m_currentUser = ""; private String m_currentUser = "";
static final AuthorizationsCollector emptyImmutableCollector() { static final AuthorizationsCollector emptyImmutableCollector() {
@@ -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.spi.impl.security; package io.moquette.spi.impl.security;
@@ -13,10 +28,7 @@ import java.sql.*;
/** /**
* Load user credentials from a SQL database. sql driver must be provided at runtime * Load user credentials from a SQL database. sql driver must be provided at runtime
*
* @author mackristof
*/ */
public class DBAuthenticator implements IAuthenticator { public class DBAuthenticator implements IAuthenticator {
private static final Logger LOG = LoggerFactory.getLogger(DBAuthenticator.class); private static final Logger LOG = LoggerFactory.getLogger(DBAuthenticator.class);
@@ -19,9 +19,6 @@ package io.moquette.spi.impl.security;
import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator; import io.moquette.spi.security.IAuthorizator;
/**
* @author andrea
*/
public class DenyAllAuthorizator implements IAuthorizator { public class DenyAllAuthorizator implements IAuthorizator {
@Override @Override
@@ -31,7 +31,6 @@ import io.moquette.server.config.FileResourceLoader;
* NB -n is important because echo append a newline by default at the of string. -n avoid this * NB -n is important because echo append a newline by default at the of string. -n avoid this
* behaviour. * behaviour.
* *
* @author andrea
* @deprecated user {@link ResourceAuthenticator} instead * @deprecated user {@link ResourceAuthenticator} instead
*/ */
public class FileAuthenticator extends ResourceAuthenticator { public class FileAuthenticator extends ResourceAuthenticator {
@@ -19,9 +19,6 @@ package io.moquette.spi.impl.security;
import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator; import io.moquette.spi.security.IAuthorizator;
/**
* @author andrea
*/
public class PermitAllAuthorizator implements IAuthorizator { public class PermitAllAuthorizator implements IAuthorizator {
@Override @Override
@@ -42,19 +42,16 @@ import java.util.Map;
* *
* NB -n is important because echo append a newline by default at the of string. -n avoid this * NB -n is important because echo append a newline by default at the of string. -n avoid this
* behaviour. * behaviour.
*
* @author andrea
*/ */
public class ResourceAuthenticator implements IAuthenticator { public class ResourceAuthenticator implements IAuthenticator {
protected static final Logger LOG = LoggerFactory.getLogger(ResourceAuthenticator.class); protected static final Logger LOG = LoggerFactory.getLogger(ResourceAuthenticator.class);
private Map<String, String> m_identities = new HashMap<>(); private Map<String, String> m_identities = new HashMap<>();
private MessageDigest m_digest;
public ResourceAuthenticator(IResourceLoader resourceLoader, String resourceName) { public ResourceAuthenticator(IResourceLoader resourceLoader, String resourceName) {
try { try {
this.m_digest = MessageDigest.getInstance("SHA-256"); MessageDigest.getInstance("SHA-256");
} catch (NoSuchAlgorithmException nsaex) { } catch (NoSuchAlgorithmException nsaex) {
LOG.error("Can't find SHA-256 for password encoding", nsaex); LOG.error("Can't find SHA-256 for password encoding", nsaex);
throw new RuntimeException(nsaex); throw new RuntimeException(nsaex);
@@ -22,9 +22,6 @@ import java.io.Serializable;
/** /**
* Maintain the information about which Topic a certain ClientID is subscribed and at which QoS * Maintain the information about which Topic a certain ClientID is subscribed and at which QoS
*
*
* @author andrea
*/ */
public final class Subscription implements Serializable { public final class Subscription implements Serializable {
@@ -26,8 +26,6 @@ import java.util.concurrent.atomic.AtomicReference;
/** /**
* Represents a tree of topics subscriptions. * Represents a tree of topics subscriptions.
*
* @author andrea
*/ */
public class SubscriptionsStore { public class SubscriptionsStore {
@@ -125,7 +123,7 @@ public class SubscriptionsStore {
couple.createdNode.addSubscription(newSubscription); //createdNode could be null? couple.createdNode.addSubscription(newSubscription); //createdNode could be null?
couple.root.recalculateSubscriptionsSize(); couple.root.recalculateSubscriptionsSize();
//spin lock repeating till we can, swap root, if can't swap just re-do the operation //spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root)); } while (!subscriptions.compareAndSet(oldRoot, couple.root));
LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot); LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot);
} }
@@ -168,7 +166,7 @@ public class SubscriptionsStore {
couple.createdNode.remove(new ClientTopicCouple(clientID, topic)); couple.createdNode.remove(new ClientTopicCouple(clientID, topic));
couple.root.recalculateSubscriptionsSize(); couple.root.recalculateSubscriptionsSize();
//spin lock repeating till we can, swap root, if can't swap just re-do the operation //spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root)); } while (!subscriptions.compareAndSet(oldRoot, couple.root));
} }
/** /**
@@ -26,7 +26,7 @@ class TreeNode {
List<TreeNode> m_children = new ArrayList<>(); List<TreeNode> m_children = new ArrayList<>();
Set<ClientTopicCouple> m_subscriptions = new HashSet<>(); Set<ClientTopicCouple> m_subscriptions = new HashSet<>();
private int subtreeSubscriptions = 0; private int subtreeSubscriptions;
TreeNode() { TreeNode() {
} }
@@ -18,8 +18,6 @@ package io.moquette.spi.security;
/** /**
* username and password checker * username and password checker
*
* @author andrea
*/ */
public interface IAuthenticator { public interface IAuthenticator {
@@ -23,8 +23,6 @@ import io.moquette.spi.impl.subscriptions.Topic;
* *
* Create an authorizator that matches topic names with same grammar of subscriptions. The # is * Create an authorizator that matches topic names with same grammar of subscriptions. The # is
* always a terminator and its the multilevel matcher. The + sign is the single level matcher. * always a terminator and its the multilevel matcher. The + sign is the single level matcher.
*
* @author andrea
*/ */
public interface IAuthorizator { public interface IAuthorizator {
@@ -20,8 +20,6 @@ import javax.net.ssl.SSLContext;
/** /**
* SSL certificate loader used to open SSL connections (websocket and MQTT-S). * SSL certificate loader used to open SSL connections (websocket and MQTT-S).
*
* Created by andrea on 13/12/15.
*/ */
public interface ISslContextCreator { public interface ISslContextCreator {
+5 -9
View File
@@ -20,17 +20,10 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled; import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.CorruptedFrameException;
import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.AttributeMap;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException; import java.io.UnsupportedEncodingException;
/** public final class Utils {
*
* @author andrea
*/
public class Utils {
public static final int MAX_LENGTH_LIMIT = 268435455; public static final int MAX_LENGTH_LIMIT = 268435455;
@@ -188,7 +181,7 @@ public class Utils {
flags |= 0x01; flags |= 0x01;
} }
flags |= ((message.fixedHeader().qosLevel().value() & 0x03) << 1); flags |= (message.fixedHeader().qosLevel().value() & 0x03) << 1;
return flags; return flags;
} }
@@ -203,4 +196,7 @@ public class Utils {
// } // }
// return protocolVersion == VERSION_3_1_1; // return protocolVersion == VERSION_3_1_1;
// } // }
private Utils() {
}
} }
@@ -31,10 +31,6 @@ import java.io.IOException;
import java.util.Properties; import java.util.Properties;
import static org.junit.Assert.*; import static org.junit.Assert.*;
/**
*
* @author luca <[email protected]>
*/
public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizator { public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizator {
Server m_server; Server m_server;
@@ -16,20 +16,16 @@
package io.moquette.server; package io.moquette.server;
import io.moquette.server.config.IConfig;
import java.io.File; import java.io.File;
import java.util.Properties; import java.util.Properties;
import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
import static io.moquette.BrokerConstants.PORT_PROPERTY_NAME; import static io.moquette.BrokerConstants.PORT_PROPERTY_NAME;
import static org.junit.Assert.assertFalse;
/** /**
* Used to carry integration configurations. * Used to carry integration configurations.
*
* Created by andrea on 4/7/15.
*/ */
public class IntegrationUtils { public final class IntegrationUtils {
static String localMapDBPath() { static String localMapDBPath() {
String currentDir = System.getProperty("user.dir"); String currentDir = System.getProperty("user.dir");
@@ -69,4 +65,7 @@ public class IntegrationUtils {
} }
assertFalse(dbFile.exists()); assertFalse(dbFile.exists());
}*/ }*/
private IntegrationUtils() {
}
} }
@@ -30,17 +30,11 @@ import org.eclipse.jetty.websocket.api.annotations.WebSocket;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
@WebSocket @WebSocket
public class MQTTWebSocket { public class MQTTWebSocket {
private static final Logger LOG = LoggerFactory.getLogger(MQTTWebSocket.class); private static final Logger LOG = LoggerFactory.getLogger(MQTTWebSocket.class);
private Session session;
private final CountDownLatch connectSentLatch; private final CountDownLatch connectSentLatch;
public MQTTWebSocket() { public MQTTWebSocket() {
@@ -50,14 +44,12 @@ public class MQTTWebSocket {
@OnWebSocketClose @OnWebSocketClose
public void onClose(int statusCode, String reason) { public void onClose(int statusCode, String reason) {
System.out.printf("Connection closed: %d - %s%n", statusCode, reason); System.out.printf("Connection closed: %d - %s%n", statusCode, reason);
this.session = null;
} }
@OnWebSocketConnect @OnWebSocketConnect
public void onConnect(Session session) throws IOException { public void onConnect(Session session) throws IOException {
System.out.printf("Got connect: %s%n", session); System.out.printf("Got connect: %s%n", session);
LOG.info("Got connect: %s%n", session); LOG.info("Got connect: %s%n", session);
this.session = session;
// Trivial MQTT Connect message // Trivial MQTT Connect message
// assertEquals(12, out.readByte()); //remaining length // assertEquals(12, out.readByte()); //remaining length
@@ -26,7 +26,7 @@ import java.util.concurrent.*;
*/ */
public class MessageCollector implements MqttCallback { public class MessageCollector implements MqttCallback {
private class ReceivedMessage { private final class ReceivedMessage {
private final MqttMessage message; private final MqttMessage message;
private final String topic; private final String topic;
@@ -38,7 +38,7 @@ public class MessageCollector implements MqttCallback {
} }
private BlockingQueue<ReceivedMessage> m_messages = new LinkedBlockingQueue<>(); private BlockingQueue<ReceivedMessage> m_messages = new LinkedBlockingQueue<>();
private boolean m_connectionLost = false; private boolean m_connectionLost;
/** /**
* Return the message from the queue if not empty, else return null with wait period. * Return the message from the queue if not empty, else return null with wait period.
@@ -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.server; package io.moquette.server;
@@ -18,9 +33,6 @@ import java.sql.SQLException;
import java.util.Properties; import java.util.Properties;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
/**
* Created by mackristof on 13/05/2016.
*/
public class ServerIntegrationDBAuthenticatorTest { public class ServerIntegrationDBAuthenticatorTest {
private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationDBAuthenticatorTest.class); private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationDBAuthenticatorTest.class);
@@ -29,10 +29,6 @@ import java.util.Properties;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static org.junit.Assert.*; import static org.junit.Assert.*;
/**
*
* @author andrea
*/
public class ServerIntegrationFuseTest { public class ServerIntegrationFuseTest {
private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class);
@@ -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.server; package io.moquette.server;
@@ -34,10 +34,6 @@ import java.io.IOException;
import java.util.Properties; import java.util.Properties;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
/**
*
* @author andrea
*/
public class ServerIntegrationQoSValidationTest { public class ServerIntegrationQoSValidationTest {
private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class);
@@ -54,40 +54,55 @@ import org.slf4j.LoggerFactory;
* *
* <pre> * <pre>
* # generate server certificate chain (valid for 10000 days) * # generate server certificate chain (valid for 10000 days)
* keytool -genkeypair -alias signedtestserver -dname cn=moquette.eclipse.org -keyalg RSA -keysize 2048 -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv -validity 10000 * keytool -genkeypair -alias signedtestserver -dname cn=moquette.eclipse.org -keyalg RSA -keysize 2048 \
* keytool -genkeypair -alias signedtestserver_sub -dname cn=moquette.eclipse.org -keyalg RSA -keysize 2048 -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv * -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv -validity 10000
* keytool -genkeypair -alias signedtestserver_sub -dname cn=moquette.eclipse.org -keyalg RSA -keysize 2048 \
* -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv
* *
* # sign server subcertificate with server certificate (valid for 10000 days) * # sign server subcertificate with server certificate (valid for 10000 days)
* keytool -certreq -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv | \ * keytool -certreq -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* keytool -gencert -alias signedtestserver -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv -validity 10000 | \ * -storepass passw0rdsrv | \
* keytool -importcert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv * keytool -gencert -alias signedtestserver -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* -storepass passw0rdsrv -validity 10000 | \
* keytool -importcert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* -storepass passw0rdsrv
* *
* # generate client keypair * # generate client keypair
* keytool -genkeypair -alias signedtestclient -dname cn=moquette.eclipse.org -keyalg RSA -keysize 2048 -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd * keytool -genkeypair -alias signedtestclient -dname cn=moquette.eclipse.org -keyalg RSA -keysize 2048 \
* -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd
* *
* # create signed client certificate with server subcertificate and import to client keystore (valid for 10000 days) * # create signed client certificate with server subcertificate and import to client keystore (valid for 10000 days)
* keytool -certreq -alias signedtestclient -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd | \ * keytool -certreq -alias signedtestclient -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd | \
* keytool -gencert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv -validity 10000 | \ * keytool -gencert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* keytool -importcert -alias signedtestclient -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd -noprompt * -storepass passw0rdsrv -validity 10000 | \
* keytool -importcert -alias signedtestclient -keystore signedclientkeystore.jks -keypass passw0rd \
* -storepass passw0rd -noprompt
* *
* # import server certificates into signed truststore * # import server certificates into signed truststore
* keytool -exportcert -alias signedtestserver -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv | \ * keytool -exportcert -alias signedtestserver -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* keytool -importcert -trustcacerts -noprompt -alias signedtestserver -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd * -storepass passw0rdsrv | \
* keytool -exportcert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv | \ * keytool -importcert -trustcacerts -noprompt -alias signedtestserver -keystore signedclientkeystore.jks \
* keytool -importcert -trustcacerts -noprompt -alias signedtestserver_sub -keystore signedclientkeystore.jks -keypass passw0rd -storepass passw0rd * -keypass passw0rd -storepass passw0rd
* keytool -exportcert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* -storepass passw0rdsrv | \
* keytool -importcert -trustcacerts -noprompt -alias signedtestserver_sub -keystore signedclientkeystore.jks \
* -keypass passw0rd -storepass passw0rd
* *
* # create unsigned client certificate (valid for 10000 days) * # create unsigned client certificate (valid for 10000 days)
* keytool -genkeypair -alias unsignedtestclient -dname cn=moquette.eclipse.org -validity 10000 -keyalg RSA -keysize 2048 -keystore unsignedclientkeystore.jks -keypass passw0rd -storepass passw0rd * keytool -genkeypair -alias unsignedtestclient -dname cn=moquette.eclipse.org -validity 10000 -keyalg RSA \
* -keysize 2048 -keystore unsignedclientkeystore.jks -keypass passw0rd -storepass passw0rd
* *
* # import server certificates into unsigned truststore * # import server certificates into unsigned truststore
* keytool -exportcert -alias signedtestserver -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv | \ * keytool -exportcert -alias signedtestserver -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* keytool -importcert -trustcacerts -noprompt -alias signedtestserver -keystore unsignedclientkeystore.jks -keypass passw0rd -storepass passw0rd * -storepass passw0rdsrv | \
* keytool -exportcert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv -storepass passw0rdsrv | \ * keytool -importcert -trustcacerts -noprompt -alias signedtestserver -keystore unsignedclientkeystore.jks \
* keytool -importcert -trustcacerts -noprompt -alias signedtestserver_sub -keystore unsignedclientkeystore.jks -keypass passw0rd -storepass passw0rd * -keypass passw0rd -storepass passw0rd
* keytool -exportcert -alias signedtestserver_sub -keystore signedserverkeystore.jks -keypass passw0rdsrv \
* -storepass passw0rdsrv | \
* keytool -importcert -trustcacerts -noprompt -alias signedtestserver_sub -keystore unsignedclientkeystore.jks \
* -keypass passw0rd -storepass passw0rd
* </pre> * </pre>
* </p> * </p>
*
* @author andrea / konradm
*/ */
public class ServerIntegrationSSLClientAuthTest { public class ServerIntegrationSSLClientAuthTest {
@@ -49,8 +49,6 @@ import org.slf4j.LoggerFactory;
/** /**
* Check that Moquette could also handle SSL. * Check that Moquette could also handle SSL.
*
* @author andrea
*/ */
public class ServerIntegrationSSLTest { public class ServerIntegrationSSLTest {
@@ -34,8 +34,6 @@ import static org.junit.Assert.assertTrue;
/** /**
* Integration test to check the function of Moquette with a WebSocket channel. * Integration test to check the function of Moquette with a WebSocket channel.
*
* @author andrea
*/ */
public class ServerIntegrationWebSocketTest { public class ServerIntegrationWebSocketTest {
@@ -32,10 +32,6 @@ import java.util.Properties;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static org.junit.Assert.*; import static org.junit.Assert.*;
/**
*
* @author andrea
*/
public class ServerLowlevelMessagesIntegrationTests { public class ServerLowlevelMessagesIntegrationTests {
private static final Logger LOG = LoggerFactory.getLogger(ServerLowlevelMessagesIntegrationTests.class); private static final Logger LOG = LoggerFactory.getLogger(ServerLowlevelMessagesIntegrationTests.class);
@@ -30,10 +30,6 @@ import java.util.Properties;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull; import static org.junit.Assert.assertNull;
/**
*
* @author andrea
*/
public class ServerRestartIntegrationTest { public class ServerRestartIntegrationTest {
static MqttClientPersistence s_dataStore; static MqttClientPersistence s_dataStore;
@@ -63,7 +59,6 @@ public class ServerRestartIntegrationTest {
@Before @Before
public void setUp() throws Exception { public void setUp() throws Exception {
String dbPath = IntegrationUtils.localMapDBPath();
startServer(); startServer();
m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore);
@@ -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.server.config; package io.moquette.server.config;
@@ -16,7 +16,6 @@
package io.moquette.server.config; package io.moquette.server.config;
import io.moquette.BrokerConstants;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import java.io.Reader; import java.io.Reader;
@@ -24,13 +23,8 @@ import java.io.StringReader;
import java.text.ParseException; import java.text.ParseException;
import java.util.Properties; import java.util.Properties;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
/**
*
* @author andrea
*/
public class ConfigurationParserTest { public class ConfigurationParserTest {
ConfigurationParser m_parser; ConfigurationParser m_parser;
@@ -96,13 +90,4 @@ public class ConfigurationParserTest {
assertEquals("localhost", props.getProperty("host")); assertEquals("localhost", props.getProperty("host"));
assertEquals("multi word string property", props.getProperty("fake")); assertEquals("multi word string property", props.getProperty("fake"));
} }
/**
* Helper method to verify default options.
*/
private void verifyDefaults(Properties props) {
assertEquals(BrokerConstants.PORT, Integer.parseInt(props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)));
assertEquals(BrokerConstants.HOST, props.getProperty(BrokerConstants.HOST_PROPERTY_NAME));
assertNull(props.getProperty(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME));
}
} }
@@ -28,11 +28,6 @@ import org.junit.Before;
import org.junit.Ignore; import org.junit.Ignore;
import org.junit.Test; import org.junit.Test;
/**
*
* @author pkhanal
*
*/
public class ConnectionIT { public class ConnectionIT {
Server m_server; Server m_server;
@@ -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.spi; package io.moquette.spi;
@@ -32,9 +32,6 @@ import static org.mockito.Matchers.refEq;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
/**
* @author Wagner Macedo
*/
public class BrokerInterceptorTest { public class BrokerInterceptorTest {
// value to check for changes after every notification // value to check for changes after every notification
@@ -22,8 +22,6 @@ import io.moquette.spi.security.IAuthenticator;
/** /**
* Test utility to implements authenticator instance. * Test utility to implements authenticator instance.
*
* @author andrea
*/ */
class MockAuthenticator implements IAuthenticator { class MockAuthenticator implements IAuthenticator {
@@ -27,7 +27,7 @@ import static org.junit.Assert.assertTrue;
/** /**
* Some useful assertions used by Netty's EmbeddedChannel in tests. * Some useful assertions used by Netty's EmbeddedChannel in tests.
*/ */
class NettyChannelAssertions { final class NettyChannelAssertions {
static void assertEqualsConnAck(MqttConnectReturnCode expectedCode, Object connAck) { static void assertEqualsConnAck(MqttConnectReturnCode expectedCode, Object connAck) {
assertEqualsConnAck(null, expectedCode, connAck); assertEqualsConnAck(null, expectedCode, connAck);
@@ -54,4 +54,6 @@ class NettyChannelAssertions {
// assertEquals(expectedCode, connAckMsg.getReturnCode()); // assertEquals(expectedCode, connAckMsg.getReturnCode());
} }
private NettyChannelAssertions() {
}
} }
@@ -40,10 +40,6 @@ import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDe
import static org.junit.Assert.*; import static org.junit.Assert.*;
import static org.mockito.Mockito.*; import static org.mockito.Mockito.*;
/**
*
* @author andrea
*/
public class ProtocolProcessorTest { public class ProtocolProcessorTest {
static final String FAKE_CLIENT_ID = "FAKE_123"; static final String FAKE_CLIENT_ID = "FAKE_123";
@@ -263,7 +259,7 @@ public class ProtocolProcessorTest {
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
assertEquals(1, subscriptions.size()); assertEquals(1, subscriptions.size());
//Exercise //Exercise
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
@@ -311,7 +307,8 @@ public class ProtocolProcessorTest {
@Test @Test
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
// simulate a connect that register a clientID to an IoSession // simulate a connect that register a clientID to an IoSession
final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); final Subscription subscription =
new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true // subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() { SubscriptionsStore subs = new SubscriptionsStore() {
@@ -43,10 +43,6 @@ import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
/**
*
* @author andrea
*/
public class ProtocolProcessor_CONNECT_Test { public class ProtocolProcessor_CONNECT_Test {
EmbeddedChannel m_session; EmbeddedChannel m_session;
@@ -23,9 +23,6 @@ import java.io.StringReader;
import java.text.ParseException; import java.text.ParseException;
import static org.junit.Assert.*; import static org.junit.Assert.*;
/**
* @author andrea
*/
public class ACLFileParserTest { public class ACLFileParserTest {
@Test @Test
@@ -22,9 +22,6 @@ import io.moquette.spi.impl.subscriptions.Topic;
import java.text.ParseException; import java.text.ParseException;
import static org.junit.Assert.*; import static org.junit.Assert.*;
/**
* @author andrea
*/
public class AuthorizationsCollectorTest { public class AuthorizationsCollectorTest {
private static final Authorization RW_ANEMOMETER = new Authorization(new Topic("/weather/italy/anemometer")); private static final Authorization RW_ANEMOMETER = new Authorization(new Topic("/weather/italy/anemometer"));

Some files were not shown because too many files have changed in this diff Show More