Fixed javadoc (issue #164)

This commit is contained in:
andsel
2017-01-04 12:52:42 +01:00
parent 480e36579c
commit 12a3c65c9b
15 changed files with 98 additions and 22 deletions
@@ -81,6 +81,7 @@ public class Server {
/** /**
* Starts Moquette bringing the configuration from the file * Starts Moquette bringing the configuration from the file
* located at m_config/moquette.conf * located at m_config/moquette.conf
* @throws IOException in case of any IO error.
*/ */
public void startServer() throws IOException { public void startServer() throws IOException {
IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigFile()); IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigFile());
@@ -95,6 +96,8 @@ public class Server {
/** /**
* Starts Moquette bringing the configuration from the given file * 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 { public void startServer(File configFile) throws IOException {
LOG.info("Using m_config file: " + configFile.getAbsolutePath()); LOG.info("Using m_config file: " + configFile.getAbsolutePath());
@@ -111,6 +114,9 @@ public class Server {
* <li>port</li> * <li>port</li>
* <li>password_file</li> * <li>password_file</li>
* </ul> * </ul>
*
* @param configProps the properties map to use as configuration.
* @throws IOException in case of any IO Error.
*/ */
public void startServer(Properties configProps) throws IOException { public void startServer(Properties configProps) throws IOException {
final IConfig config = new MemoryConfig(configProps); final IConfig config = new MemoryConfig(configProps);
@@ -119,6 +125,8 @@ public class Server {
/** /**
* Starts Moquette bringing the configuration files from the given Config implementation. * 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 { public void startServer(IConfig config) throws IOException {
startServer(config, null); startServer(config, null);
@@ -127,6 +135,9 @@ public class Server {
/** /**
* Starts Moquette with config provided by an implementation of IConfig class and with the * Starts Moquette with config provided by an implementation of IConfig class and with the
* set of InterceptHandler. * 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<? extends InterceptHandler> handlers) throws IOException { public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
startServer(config, handlers, null, null, null); 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. * SPI method used by Broker embedded applications to get list of subscribers.
* Returns null if the broker is not started. * Returns null if the broker is not started.
*
* @return list of subscriptions.
*/ */
public List<Subscription> getSubscriptions() { public List<Subscription> getSubscriptions() {
if (m_processorBootstrapper == null) { if (m_processorBootstrapper == null) {
@@ -230,6 +243,8 @@ public class Server {
/** /**
* SPI method used by Broker embedded applications to add intercept handlers. * 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) { public boolean addInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) { if (!m_initialized) {
@@ -240,6 +255,8 @@ public class Server {
/** /**
* SPI method used by Broker embedded applications to remove intercept handlers. * 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) { public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) { if (!m_initialized) {
@@ -136,6 +136,9 @@ public class AutoFlushHandler extends ChannelDuplexHandler {
/** /**
* Is called when the write timeout expire. * 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 { protected void channelIdle(ChannelHandlerContext ctx/*, IdleStateEvent evt*/) throws Exception {
// ctx.fireUserEventTriggered(evt); // ctx.fireUserEventTriggered(evt);
@@ -82,6 +82,8 @@ public class ClientSession {
/** /**
* Remove the messages stored in a cleanSession false. * 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) { public void removeEnqueued(MessageGUID guid) {
this.m_sessionsStore.removeEnqueued(this.clientID, guid); this.m_sessionsStore.removeEnqueued(this.clientID, guid);
@@ -182,6 +184,7 @@ public class ClientSession {
/** /**
* Enqueue a message to be sent to the client. * Enqueue a message to be sent to the client.
* @param pubMessage the message to enqueue.
* @return false if the queue is full. * @return false if the queue is full.
* */ * */
public boolean enqueue(PublishMessage pubMessage) { public boolean enqueue(PublishMessage pubMessage) {
@@ -125,24 +125,31 @@ public interface IMessagesStore {
/** /**
* Persist the message. * Persist the message.
* If the message is empty then the topic is cleaned, else it's stored. * 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); void storeRetained(String topic, MessageGUID guid);
/** /**
* Return a list of retained messages that satisfy the condition. * 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<StoredMessage> searchMatching(IMatchingCondition condition); Collection<StoredMessage> searchMatching(IMatchingCondition condition);
/** /**
* Persist the message. * Persist the message.
* @param storedMessage the message to store for future usage.
* @return the unique id in the storage (guid). * @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. * Return the list of persisted publishes for the given clientID.
* For QoS1 and QoS2 with clean session flag, this method return the list of * For QoS1 and QoS2 with clean session flag, this method return the list of
* missed publish events while the client was disconnected. * 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<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids); List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids);
@@ -71,16 +71,20 @@ public interface ISessionsStore {
/** /**
* Add a new subscription to the session * Add a new subscription to the session
* @param newSubscription the subscription to add.
*/ */
void addNewSubscription(Subscription newSubscription); void addNewSubscription(Subscription newSubscription);
/** /**
* Removed a specific subscription * Removed a specific subscription
* @param topic the topic of the subscription.
* @param clientID the session client.
* */ * */
void removeSubscription(String topic, String clientID); void removeSubscription(String topic, String clientID);
/** /**
* Remove all the subscriptions of the session * Remove all the subscriptions of the session
* @param sessionID the client ID
*/ */
void wipeSubscriptions(String sessionID); void wipeSubscriptions(String sessionID);
@@ -90,6 +94,7 @@ public interface ISessionsStore {
List<ClientTopicCouple> listAllSubscriptions(); List<ClientTopicCouple> listAllSubscriptions();
/** /**
* @param couple the subscription descriptor.
* @return the subscription stored by clientID and topicFilter, if any else null; * @return the subscription stored by clientID and topicFilter, if any else null;
* */ * */
Subscription getSubscription(ClientTopicCouple couple); Subscription getSubscription(ClientTopicCouple couple);
@@ -100,6 +105,7 @@ public interface ISessionsStore {
List<Subscription> getSubscriptions(); List<Subscription> getSubscriptions();
/** /**
* @param clientID the session client ID.
* @return true iff there are subscriptions persisted with clientID * @return true iff there are subscriptions persisted with clientID
*/ */
boolean contains(String clientID); boolean contains(String clientID);
@@ -115,33 +121,46 @@ public interface ISessionsStore {
void inFlightAck(String clientID, int messageID); 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); void inFlight(String clientID, int messageID, MessageGUID guid);
/** /**
* Return the next valid packetIdentifier for the given client session. * 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); int nextPacketID(String clientID);
/** /**
* Store the guid to be later published. * Store the guid to be later published.
* @param guid the message guid to bind.
* @param clientID the clientID
* */ * */
void bindToDeliver(MessageGUID guid, String clientID); void bindToDeliver(MessageGUID guid, String clientID);
/** /**
* List the guids for retained messages for the session * 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<MessageGUID> enqueued(String clientID); Collection<MessageGUID> enqueued(String clientID);
/** /**
* Remove form the queue of stored messages for session. * 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 removeEnqueued(String clientID, MessageGUID guid);
void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); 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. * @return the guid of message just acked.
* */ * */
MessageGUID secondPhaseAcknowledged(String clientID, int messageID); MessageGUID secondPhaseAcknowledged(String clientID, int messageID);
@@ -474,6 +474,8 @@ public class ProtocolProcessor {
* for Qos1 and Qos2. * for Qos1 and Qos2.
* It also doesn't notifyTopicPublished because using internally the owner should already know where * It also doesn't notifyTopicPublished because using internally the owner should already know where
* it's publishing. * it's publishing.
*
* @param msg the message to publish.
* */ * */
public void internalPublish(PublishMessage msg) { public void internalPublish(PublishMessage msg) {
final AbstractMessage.QOSType qos = msg.getQos(); 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 * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
* to all interested subscribers. * 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) { public void processPubRel(Channel channel, PubRelMessage msg) {
this.qos2PublishHandler.processPubRel(channel, msg); this.qos2PublishHandler.processPubRel(channel, msg);
@@ -684,7 +688,9 @@ public class ProtocolProcessor {
/** /**
* Remove the clientID from topic subscription, if not previously subscribed, * 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) { public void processUnsubscribe(Channel channel, UnsubscribeMessage msg) {
List<String> topics = msg.topicFilters(); List<String> topics = msg.topicFilters();
@@ -74,6 +74,8 @@ public class ProtocolProcessorBootstrapper {
* and fallback on the default one (permit all). * 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 * @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). * 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<? extends InterceptHandler> embeddedObservers, public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> embeddedObservers,
IAuthenticator authenticator, IAuthorizator authorizator, Server server) { IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
@@ -24,7 +24,7 @@ import java.text.ParseException;
/** /**
* Parses the acl configuration file. * Parses the acl configuration file.
* If a line starts with # it's comment. * If a line starts with # it's comment.
* Blank lines are skipped. The format is "topic [read|write|readwrite] <topic name>" * Blank lines are skipped. The format is "topic [read|write|readwrite] {topic name}"
* *
* @author andrea * @author andrea
* */ * */
@@ -33,6 +33,9 @@ public class ACLFileParser {
/** /**
* Parse the configuration from file. * 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 { public static AuthorizationsCollector parse(File file) throws ParseException {
if (file == null) { if (file == null) {
@@ -54,8 +57,9 @@ public class ACLFileParser {
/** /**
* Parse the ACL configuration file * Parse the ACL configuration file
* * @param reader to parse
* @throws java.text.ParseException if the format is not compliant. * @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 { public static AuthorizationsCollector parse(Reader reader) throws ParseException {
if (reader == null) { if (reader == null) {
@@ -19,7 +19,7 @@ import io.moquette.server.config.FileResourceLoader;
/** /**
* Load user credentials from a text file. * Load user credentials from a text file.
* Each line of the file is formatted as "<username>:<sha256(password)>". 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: * To encode your password from command line on Linux systems, you could use:
* <pre> * <pre>
@@ -33,7 +33,7 @@ import java.util.Map;
/** /**
* Load user credentials from a text resource. * Load user credentials from a text resource.
* Each line of the file is formatted as "<username>:<sha256(password)>". 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: * To encode your password from command line on Linux systems, you could use:
* <pre> * <pre>
@@ -44,6 +44,8 @@ public class SubscriptionsStore {
/** /**
* Check if the topic filter of the subscription is well formed * 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) { public static boolean validate(String topicFilter) {
try { try {
@@ -98,6 +100,7 @@ public class SubscriptionsStore {
/** /**
* Initialize the subscription tree with the list of subscriptions. * Initialize the subscription tree with the list of subscriptions.
* Maintained for compatibility reasons. * Maintained for compatibility reasons.
* @param sessionsStore to be used as backing store from the subscription store.
*/ */
public void init(ISessionsStore sessionsStore) { public void init(ISessionsStore sessionsStore) {
LOG.debug("init invoked"); LOG.debug("init invoked");
@@ -178,6 +181,7 @@ public class SubscriptionsStore {
/** /**
* Visit the topics tree to remove matching subscriptions with clientID. * 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). * 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) { public void removeForClient(String clientID) {
TreeNode oldRoot; TreeNode oldRoot;
@@ -194,6 +198,9 @@ public class SubscriptionsStore {
* Given a topic string return the clients subscriptions that matches it. * Given a topic string return the clients subscriptions that matches it.
* Topic string can't contain character # and + because they are reserved to * Topic string can't contain character # and + because they are reserved to
* listeners subscriptions, and not topic publishing. * 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<Subscription> matches(String topic) { public List<Subscription> matches(String topic) {
List<Token> tokens; List<Token> tokens;
@@ -252,6 +259,9 @@ public class SubscriptionsStore {
/** /**
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A * 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 //TODO reimplement with iterators or with queues
public static boolean matchTopics(String msgTopic, String subscriptionTopic) { public static boolean matchTopics(String msgTopic, String subscriptionTopic) {
@@ -74,18 +74,18 @@ class MapDBMessagesStore implements IMessagesStore {
} }
@Override @Override
public MessageGUID storePublishForFuture(StoredMessage evt) { public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
LOG.debug("storePublishForFuture store evt {}", evt); LOG.debug("storePublishForFuture store evt {}", storedMessage);
if (evt.getClientID() == null) { if (storedMessage.getClientID() == null) {
LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt); 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"); throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
} }
MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
evt.setGuid(guid); storedMessage.setGuid(guid);
LOG.debug("storePublishForFuture guid <{}>", guid); LOG.debug("storePublishForFuture guid <{}>", guid);
m_persistentMessageStore.put(guid, evt); m_persistentMessageStore.put(guid, storedMessage);
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID())); ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(storedMessage.getClientID()));
messageIdToGuid.put(evt.getMessageID(), guid); messageIdToGuid.put(storedMessage.getMessageID(), guid);
return guid; return guid;
} }
@@ -69,6 +69,7 @@ public class MapDBPersistentStore {
/** /**
* Factory method to create message store backed by MapDB * Factory method to create message store backed by MapDB
* @return the message store instance.
* */ * */
public IMessagesStore messagesStore() { public IMessagesStore messagesStore() {
return m_messageStore; return m_messageStore;
@@ -28,6 +28,10 @@ public interface IAuthorizator {
/** /**
* Ask the implementation of the authorizator if the topic can be used in a publish. * 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); boolean canWrite(String topic, String user, String client);
@@ -67,14 +67,14 @@ public class MemoryMessagesStore implements IMessagesStore {
} }
@Override @Override
public MessageGUID storePublishForFuture(StoredMessage evt) { public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
LOG.debug("storePublishForFuture store evt {}", evt); LOG.debug("storePublishForFuture store evt {}", storedMessage);
MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
evt.setGuid(guid); storedMessage.setGuid(guid);
m_persistentMessageStore.put(guid, evt); m_persistentMessageStore.put(guid, storedMessage);
HashMap<Integer, MessageGUID> guids = (HashMap<Integer, MessageGUID>) defaultGet(m_messageToGuids, HashMap<Integer, MessageGUID> guids = (HashMap<Integer, MessageGUID>) defaultGet(m_messageToGuids,
evt.getClientID(), new HashMap<Integer, MessageGUID>()); storedMessage.getClientID(), new HashMap<Integer, MessageGUID>());
guids.put(evt.getMessageID(), guid); guids.put(storedMessage.getMessageID(), guid);
return guid; return guid;
} }