diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 1bd63904..a9c8b9dd 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -81,6 +81,7 @@ public class Server { /** * Starts Moquette bringing the configuration from the file * located at m_config/moquette.conf + * @throws IOException in case of any IO error. */ public void startServer() throws IOException { IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigFile()); @@ -95,6 +96,8 @@ public class Server { /** * Starts Moquette bringing the configuration from the given file + * @param configFile text file that contains the configuration. + * @throws IOException in case of any IO Error. */ public void startServer(File configFile) throws IOException { LOG.info("Using m_config file: " + configFile.getAbsolutePath()); @@ -111,6 +114,9 @@ public class Server { *
  • port
  • *
  • password_file
  • * + * + * @param configProps the properties map to use as configuration. + * @throws IOException in case of any IO Error. */ public void startServer(Properties configProps) throws IOException { final IConfig config = new MemoryConfig(configProps); @@ -119,6 +125,8 @@ public class Server { /** * Starts Moquette bringing the configuration files from the given Config implementation. + * @param config the configuration to use to start the broker. + * @throws IOException in case of any IO Error. */ public void startServer(IConfig config) throws IOException { startServer(config, null); @@ -127,6 +135,9 @@ public class Server { /** * Starts Moquette with config provided by an implementation of IConfig class and with the * set of InterceptHandler. + * @param config the configuration to use to start the broker. + * @param handlers the handlers to install in the broker. + * @throws IOException in case of any IO Error. * */ public void startServer(IConfig config, List handlers) throws IOException { startServer(config, handlers, null, null, null); @@ -220,6 +231,8 @@ public class Server { /** * SPI method used by Broker embedded applications to get list of subscribers. * Returns null if the broker is not started. + * + * @return list of subscriptions. */ public List getSubscriptions() { if (m_processorBootstrapper == null) { @@ -230,6 +243,8 @@ public class Server { /** * SPI method used by Broker embedded applications to add intercept handlers. + * @param interceptHandler the handler to add. + * @return true id operation was successful. * */ public boolean addInterceptHandler(InterceptHandler interceptHandler) { if (!m_initialized) { @@ -240,6 +255,8 @@ public class Server { /** * SPI method used by Broker embedded applications to remove intercept handlers. + * @param interceptHandler the handler to remove. + * @return true id operation was successful. * */ public boolean removeInterceptHandler(InterceptHandler interceptHandler) { if (!m_initialized) { diff --git a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java index d87f766b..534e21e2 100644 --- a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java @@ -136,6 +136,9 @@ public class AutoFlushHandler extends ChannelDuplexHandler { /** * Is called when the write timeout expire. + * + * @param ctx the channel context. + * @throws Exception in case of any IO error. */ protected void channelIdle(ChannelHandlerContext ctx/*, IdleStateEvent evt*/) throws Exception { // ctx.fireUserEventTriggered(evt); diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 90613f58..382c25ee 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -82,6 +82,8 @@ public class ClientSession { /** * Remove the messages stored in a cleanSession false. + * + * @param guid the guid of the message to remove from the queue. */ public void removeEnqueued(MessageGUID guid) { this.m_sessionsStore.removeEnqueued(this.clientID, guid); @@ -182,6 +184,7 @@ public class ClientSession { /** * Enqueue a message to be sent to the client. + * @param pubMessage the message to enqueue. * @return false if the queue is full. * */ public boolean enqueue(PublishMessage pubMessage) { diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 0499579b..4cf2d7f3 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -125,24 +125,31 @@ public interface IMessagesStore { /** * Persist the message. * If the message is empty then the topic is cleaned, else it's stored. + * @param topic for the retained. + * @param guid of the message to mark as retained. */ void storeRetained(String topic, MessageGUID guid); /** * Return a list of retained messages that satisfy the condition. + * @param condition the condition to match during the search. + * @return the collection of matching messages. */ Collection searchMatching(IMatchingCondition condition); /** * Persist the message. + * @param storedMessage the message to store for future usage. * @return the unique id in the storage (guid). * */ - MessageGUID storePublishForFuture(StoredMessage evt); + MessageGUID storePublishForFuture(StoredMessage storedMessage); /** * Return the list of persisted publishes for the given clientID. * For QoS1 and QoS2 with clean session flag, this method return the list of * missed publish events while the client was disconnected. + * @param guids the list of of guid to use as search keys. + * @return the list of stored messages matching the passed keys. */ List listMessagesInSession(Collection guids); diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index d5231033..b831460b 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -71,16 +71,20 @@ public interface ISessionsStore { /** * Add a new subscription to the session + * @param newSubscription the subscription to add. */ void addNewSubscription(Subscription newSubscription); /** * Removed a specific subscription + * @param topic the topic of the subscription. + * @param clientID the session client. * */ void removeSubscription(String topic, String clientID); /** * Remove all the subscriptions of the session + * @param sessionID the client ID */ void wipeSubscriptions(String sessionID); @@ -90,6 +94,7 @@ public interface ISessionsStore { List listAllSubscriptions(); /** + * @param couple the subscription descriptor. * @return the subscription stored by clientID and topicFilter, if any else null; * */ Subscription getSubscription(ClientTopicCouple couple); @@ -100,6 +105,7 @@ public interface ISessionsStore { List getSubscriptions(); /** + * @param clientID the session client ID. * @return true iff there are subscriptions persisted with clientID */ boolean contains(String clientID); @@ -115,33 +121,46 @@ public interface ISessionsStore { void inFlightAck(String clientID, int messageID); /** - * Save the binding messageID, clientID <-> guid + * Save the binding messageID, clientID - guid + * @param clientID the client ID + * @param messageID the message ID + * @param guid the uuid of the message to mark as inflight. * */ void inFlight(String clientID, int messageID, MessageGUID guid); /** * Return the next valid packetIdentifier for the given client session. + * @param clientID the clientID requesting next packet id. + * @return the next valid id. * */ int nextPacketID(String clientID); /** * Store the guid to be later published. + * @param guid the message guid to bind. + * @param clientID the clientID * */ void bindToDeliver(MessageGUID guid, String clientID); /** * List the guids for retained messages for the session + * @param clientID the client ID owning the queue. + * @return the list of enqueued message UUIDs. * */ Collection enqueued(String clientID); /** * Remove form the queue of stored messages for session. + * @param clientID the client ID owning the queue. + * @param guid the message guid to dequeue. * */ void removeEnqueued(String clientID, MessageGUID guid); void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); /** + * @param clientID the client ID accessing the second phase. + * @param messageID the message ID that reached the second phase. * @return the guid of message just acked. * */ MessageGUID secondPhaseAcknowledged(String clientID, int messageID); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index ad62a657..230e887a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -474,6 +474,8 @@ public class ProtocolProcessor { * for Qos1 and Qos2. * It also doesn't notifyTopicPublished because using internally the owner should already know where * it's publishing. + * + * @param msg the message to publish. * */ public void internalPublish(PublishMessage msg) { final AbstractMessage.QOSType qos = msg.getQos(); @@ -539,6 +541,8 @@ public class ProtocolProcessor { /** * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. + * @param channel the channel of the incoming message. + * @param msg the decoded pubrel message. * */ public void processPubRel(Channel channel, PubRelMessage msg) { this.qos2PublishHandler.processPubRel(channel, msg); @@ -684,7 +688,9 @@ public class ProtocolProcessor { /** * Remove the clientID from topic subscription, if not previously subscribed, - * doesn't reply any error + * doesn't reply any error. + * @param channel the channel of the incoming message. + * @param msg the decoded unsubscribe message. */ public void processUnsubscribe(Channel channel, UnsubscribeMessage msg) { List topics = msg.topicFilters(); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 7e481a72..6ede7332 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -74,6 +74,8 @@ public class ProtocolProcessorBootstrapper { * and fallback on the default one (permit all). * @param authorizator an implementation of the authorizator to be used, if null load that specified in config * and fallback on the default one (permit all). + * @param server the serber to init. + * @return the processor created for the broker. * */ public ProtocolProcessor init(IConfig props, List embeddedObservers, IAuthenticator authenticator, IAuthorizator authorizator, Server server) { diff --git a/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java b/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java index e9d18b2e..0719a971 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java @@ -24,7 +24,7 @@ import java.text.ParseException; /** * Parses the acl configuration file. * If a line starts with # it's comment. - * Blank lines are skipped. The format is "topic [read|write|readwrite] " + * Blank lines are skipped. The format is "topic [read|write|readwrite] {topic name}" * * @author andrea * */ @@ -33,6 +33,9 @@ public class ACLFileParser { /** * Parse the configuration from file. + * @param file to parse + * @return the collector of authorizations form reader passed into. + * @throws ParseException if the format is not compliant. */ public static AuthorizationsCollector parse(File file) throws ParseException { if (file == null) { @@ -54,8 +57,9 @@ public class ACLFileParser { /** * Parse the ACL configuration file - * - * @throws java.text.ParseException if the format is not compliant. + * @param reader to parse + * @return the collector of authorizations form reader passed into. + * @throws ParseException if the format is not compliant. */ public static AuthorizationsCollector parse(Reader reader) throws ParseException { if (reader == null) { diff --git a/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java index 6c5574e3..3fc90a4e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java @@ -19,7 +19,7 @@ import io.moquette.server.config.FileResourceLoader; /** * Load user credentials from a text file. - * Each line of the file is formatted as ":". The username mustn't contains : char. + * Each line of the file is formatted as "[username]:[sha256(password)]". The username mustn't contains : char. * * To encode your password from command line on Linux systems, you could use: *
    diff --git a/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java
    index 406b8353..9761ae3f 100644
    --- a/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java
    +++ b/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java
    @@ -33,7 +33,7 @@ import java.util.Map;
     
     /**
      * Load user credentials from a text resource.
    - * Each line of the file is formatted as ":". The username mustn't contains : char.
    + * Each line of the file is formatted as "[username]:[sha256(password)]". The username mustn't contains : char.
      *
      * To encode your password from command line on Linux systems, you could use:
      * 
    diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
    index 6330f9a2..b1ebe262 100644
    --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
    +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
    @@ -44,6 +44,8 @@ public class SubscriptionsStore {
     
         /**
          * Check if the topic filter of the subscription is well formed
    +     * @param topicFilter the filter to validate
    +     * @return true if it's correct.
          * */
         public static boolean validate(String topicFilter) {
             try {
    @@ -98,6 +100,7 @@ public class SubscriptionsStore {
         /**
          * Initialize the subscription tree with the list of subscriptions.
          * Maintained for compatibility reasons.
    +     * @param sessionsStore to be used as backing store from the subscription store.
          */
         public void init(ISessionsStore sessionsStore) {
             LOG.debug("init invoked");
    @@ -178,6 +181,7 @@ public class SubscriptionsStore {
         /**
          * Visit the topics tree to remove matching subscriptions with clientID.
          * It's a mutating structure operation so create a new subscription tree (partial or total).
    +     * @param clientID the client ID to remove.
          */
         public void removeForClient(String clientID) {
             TreeNode oldRoot;
    @@ -194,6 +198,9 @@ public class SubscriptionsStore {
          * Given a topic string return the clients subscriptions that matches it.
          * Topic string can't contain character # and + because they are reserved to
          * listeners subscriptions, and not topic publishing.
    +     *
    +     * @param topic to use fo searching matching subscriptions.
    +     * @return the list of matching subscriptions, or empty if not matching.
          */
         public List matches(String topic) {
             List tokens;
    @@ -252,6 +259,9 @@ public class SubscriptionsStore {
         
         /**
          * Verify if the 2 topics matching respecting the rules of MQTT Appendix A
    +     * @param msgTopic the topic to match from the message
    +     * @param subscriptionTopic the topic filter of the subscription
    +     * @return true if the two topics match.
          */
         //TODO reimplement with iterators or with queues
         public static boolean matchTopics(String msgTopic, String subscriptionTopic) {
    diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
    index f8c3d583..ebf7593a 100644
    --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
    +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
    @@ -74,18 +74,18 @@ class MapDBMessagesStore implements IMessagesStore {
         }
     
         @Override
    -    public MessageGUID storePublishForFuture(StoredMessage evt) {
    -        LOG.debug("storePublishForFuture store evt {}", evt);
    -        if (evt.getClientID() == null) {
    -            LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt);
    +    public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
    +        LOG.debug("storePublishForFuture store evt {}", storedMessage);
    +        if (storedMessage.getClientID() == null) {
    +            LOG.error("persisting a message without a clientID, bad programming error msg: {}", storedMessage);
                 throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
             }
             MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
    -        evt.setGuid(guid);
    +        storedMessage.setGuid(guid);
             LOG.debug("storePublishForFuture guid <{}>", guid);
    -        m_persistentMessageStore.put(guid, evt);
    -        ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID()));
    -        messageIdToGuid.put(evt.getMessageID(), guid);
    +        m_persistentMessageStore.put(guid, storedMessage);
    +        ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(storedMessage.getClientID()));
    +        messageIdToGuid.put(storedMessage.getMessageID(), guid);
             return guid;
         }
     
    diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
    index 164941e2..3d1cbb81 100644
    --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
    +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
    @@ -69,6 +69,7 @@ public class MapDBPersistentStore {
     
         /**
          * Factory method to create message store backed by MapDB
    +     * @return the message store instance.
          * */
         public IMessagesStore messagesStore() {
             return m_messageStore;
    diff --git a/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java b/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java
    index 5d53c0cd..1e3c6c21 100644
    --- a/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java
    +++ b/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java
    @@ -28,6 +28,10 @@ public interface IAuthorizator {
     
         /**
          * Ask the implementation of the authorizator if the topic can be used in a publish.
    +     * @param topic the topic to write to.
    +     * @param user the user
    +     * @param client the client
    +     * @return true if the user from client can publish data on topic.
          * */
         boolean canWrite(String topic, String user, String client);
     
    diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
    index 3cf8d5cb..74bb83f0 100644
    --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
    +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
    @@ -67,14 +67,14 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
     
         @Override
    -    public MessageGUID storePublishForFuture(StoredMessage evt) {
    -        LOG.debug("storePublishForFuture store evt {}", evt);
    +    public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
    +        LOG.debug("storePublishForFuture store evt {}", storedMessage);
             MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
    -        evt.setGuid(guid);
    -        m_persistentMessageStore.put(guid, evt);
    +        storedMessage.setGuid(guid);
    +        m_persistentMessageStore.put(guid, storedMessage);
             HashMap guids = (HashMap) defaultGet(m_messageToGuids,
    -                evt.getClientID(), new HashMap());
    -        guids.put(evt.getMessageID(), guid);
    +                storedMessage.getClientID(), new HashMap());
    +        guids.put(storedMessage.getMessageID(), guid);
             return guid;
         }