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

Merge the switch from byte[] to ByteBuffer

Browse Source
This commit is contained in:
selva.andre
2013-11-27 11:01:16 +01:00
parent 101a1060e7 bd3a1f6c4d
commit 527e045c69
21 changed files with 345 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
+1 -1
View File
@@ -41,7 +41,7 @@
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.6.4</version>
<version>1.7.5</version>
</dependency>
<dependency>
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+2 -1
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi;
import java.nio.ByteBuffer;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
@@ -18,7 +19,7 @@ public interface IStorageService {
* */
void initStore();
void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos);
void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos);
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+57 -30
View File
@@ -1,9 +1,15 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.*;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.storage.StoredPublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.server.Server;
@@ -12,11 +18,6 @@ import org.fusesource.hawtdb.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
/**
* Implementation of IStorageService backed by HawtDB
@@ -38,8 +39,8 @@ public class HawtDBStorageService implements IStorageService {
return m_qos;
}
byte[] getPayload() {
return m_payload;
ByteBuffer getPayload() {
return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip();
}
String getTopic() {
@@ -54,12 +55,12 @@ public class HawtDBStorageService implements IStorageService {
private PageFileFactory pageFactory;
//maps clientID to the list of pending messages stored
private SortedIndex<String, List<PublishEvent>> m_persistentMessageStore;
private SortedIndex<String, List<StoredPublishEvent>> m_persistentMessageStore;
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
private SortedIndex<String, StoredPublishEvent> m_inflightStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_qos2Store;
private SortedIndex<String, StoredPublishEvent> m_qos2Store;
//persistent Map of clientID, set of Subscriptions
private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
@@ -101,10 +102,10 @@ public class HawtDBStorageService implements IStorageService {
private void initPersistentMessageStore() {
BTreeIndexFactory<String, List<PublishEvent>> indexFactory = new BTreeIndexFactory<String, List<PublishEvent>>();
BTreeIndexFactory<String, List<StoredPublishEvent>> indexFactory = new BTreeIndexFactory<String, List<StoredPublishEvent>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
m_persistentMessageStore = (SortedIndex<String, List<StoredPublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
}
private void initPersistentSubscriptions() {
@@ -119,31 +120,33 @@ public class HawtDBStorageService implements IStorageService {
* messages in flight.
*/
private void initInflightMessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
BTreeIndexFactory<String, StoredPublishEvent> indexFactory = new BTreeIndexFactory<String, StoredPublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
m_inflightStore = (SortedIndex<String, StoredPublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
private void initPersistentQoS2MessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
BTreeIndexFactory<String, StoredPublishEvent> indexFactory = new BTreeIndexFactory<String, StoredPublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_qos2Store = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory);
m_qos2Store = (SortedIndex<String, StoredPublishEvent>) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory);
}
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
if (message.length == 0) {
public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
if (!message.hasRemaining()) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
m_retainedStore.put(topic, new StoredMessage(message, qos, topic));
byte[] raw = new byte[message.remaining()];
message.get(raw);
m_retainedStore.put(topic, new StoredMessage(raw, qos, topic));
}
}
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size());
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<StoredMessage>();
@@ -158,21 +161,29 @@ public class HawtDBStorageService implements IStorageService {
}
public void storePublishForFuture(PublishEvent evt) {
List<PublishEvent> storedEvents;
List<StoredPublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<PublishEvent>();
storedEvents = new ArrayList<StoredPublishEvent>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(evt);
storedEvents.add(convertToStored(evt));
m_persistentMessageStore.put(clientID, storedEvents);
LOG.debug(String.format("Stored published message for client <%s> on topic <%s>, content %s",
clientID, evt.getTopic(), new String(evt.getMessage())));
//NB rewind the evt message content
LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic());
}
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
return m_persistentMessageStore.get(clientID);
List<StoredPublishEvent> storedEvts = m_persistentMessageStore.get(clientID);
if (storedEvts == null) {
return null;
}
List<PublishEvent> liveEvts = new ArrayList<PublishEvent>();
for (StoredPublishEvent storedEvt : storedEvts) {
liveEvts.add(convertFromStored(storedEvt));
}
return liveEvts;
}
public void cleanPersistedPublishes(String clientID) {
@@ -184,7 +195,8 @@ public class HawtDBStorageService implements IStorageService {
}
public void addInFlight(PublishEvent evt, String publishKey) {
m_inflightStore.put(publishKey, evt);
StoredPublishEvent storedEvt = convertToStored(evt);
m_inflightStore.put(publishKey, storedEvt);
}
public void addNewSubscription(Subscription newSubscription, String clientID) {
@@ -221,8 +233,8 @@ public class HawtDBStorageService implements IStorageService {
/*-------- QoS 2 storage management --------------*/
public void persistQoS2Message(String publishKey, PublishEvent evt) {
LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt));
m_qos2Store.put(publishKey, evt);
LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt);
m_qos2Store.put(publishKey, convertToStored(evt));
}
public void removeQoS2Message(String publishKey) {
@@ -230,6 +242,21 @@ public class HawtDBStorageService implements IStorageService {
}
public PublishEvent retrieveQoS2Message(String publishKey) {
return m_qos2Store.get(publishKey);
StoredPublishEvent storedEvt = m_qos2Store.get(publishKey);
return convertFromStored(storedEvt);
}
private StoredPublishEvent convertToStored(PublishEvent evt) {
StoredPublishEvent storedEvt = new StoredPublishEvent(evt);
return storedEvt;
}
private PublishEvent convertFromStored(StoredPublishEvent evt) {
byte[] message = evt.getMessage();
ByteBuffer bbmessage = ByteBuffer.wrap(message);
//bbmessage.flip();
PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(),
bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID(), null);
return liveEvt;
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+9 -6
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
@@ -32,18 +33,20 @@ public class MemoryStorageService implements IStorageService {
//To change body of implemented methods use File | Settings | File Templates.
}
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
if (message.length == 0) {
public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
if (!message.hasRemaining()) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(message, qos, topic));
byte[] raw = new byte[message.remaining()];
message.get(raw);
m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(raw, qos, topic));
}
}
public Collection<HawtDBStorageService.StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size());
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<HawtDBStorageService.StoredMessage> results = new ArrayList<HawtDBStorageService.StoredMessage>();
@@ -58,7 +61,7 @@ public class MemoryStorageService implements IStorageService {
}
public void storePublishForFuture(PublishEvent evt) {
LOG.debug("storePublishForFuture store evt " + evt);
LOG.debug("storePublishForFuture store evt {}", evt);
List<PublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
@@ -115,7 +118,7 @@ public class MemoryStorageService implements IStorageService {
}
public void persistQoS2Message(String publishKey, PublishEvent evt) {
LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt));
LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt);
m_qos2Store.put(publishKey, evt);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+43 -42
View File
@@ -4,6 +4,7 @@ import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
@@ -78,7 +79,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
void processConnect(ServerChannel session, ConnectMessage msg) {
LOG.debug("processConnect for client " + msg.getClientID());
LOG.debug("processConnect for client {}", msg.getClientID());
if (msg.getProcotolVersion() != 0x03) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
@@ -112,7 +113,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
m_clientIDs.put(msg.getClientID(), connDescr);
int keepAlive = msg.getKeepAlive();
LOG.debug(String.format("Connect with keepAlive %d s", keepAlive));
LOG.debug("Connect with keepAlive {} s", keepAlive);
session.setAttribute(Constants.KEEP_ALIVE, keepAlive);
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
@@ -123,8 +124,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//Handle will flag
if (msg.isWillFlag()) {
AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()];
PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(),
msg.isWillRetain(), msg.getClientID(), session);
byte[] willPayload = msg.getWillMessage().getBytes();
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos,
bb, msg.isWillRetain(), msg.getClientID(), session);
processPublish(pubEvt);
}
@@ -155,7 +158,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
LOG.debug("processConnect sent OK ConnAck");
session.write(okResp);
LOG.info(String.format("Connected client ID <%s> with clean session %s", msg.getClientID(), msg.isCleanSession()));
LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
if (!msg.isCleanSession()) {
//force the republish of stored QoS1 and QoS2
@@ -167,11 +170,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
LOG.debug("republishStored invoked");
List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(clientID);
if (publishedEvents == null) {
LOG.info(String.format("No stored messages for client <%s>", clientID));
LOG.info("No stored messages for client <{}>", clientID);
return;
}
LOG.info(String.format("republishing stored messages to client <%s>", clientID));
LOG.info("republishing stored messages to client <{}>", clientID);
for (PublishEvent pubEvt : publishedEvents) {
sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
@@ -179,7 +182,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
private void processRemoveAllSubscriptions(String clientID) {
LOG.info(String.format("cleaning old saved subscriptions for client <%s>", clientID));
LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
subscriptions.removeForClient(clientID);
//remove also the messages stored of type QoS1/2
@@ -190,11 +193,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
LOG.debug("processPublish invoked with " + evt);
final String topic = evt.getTopic();
final AbstractMessage.QOSType qos = evt.getQos();
final byte[] message = evt.getMessage();
final ByteBuffer message = evt.getMessage();
boolean retain = evt.isRetain();
LOG.info(String.format("Publish recieved from clientID <%s> on topic <%s> with QoS %s",
evt.getClientID(), evt.getTopic(), evt.getQos()));
LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}",
evt.getClientID(), evt.getTopic(), evt.getQos());
String publishKey = null;
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
@@ -216,7 +219,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
m_storageService.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
LOG.debug("replying with PubAck to MSG ID " + evt.getMessageID());
LOG.debug("replying with PubAck to MSG ID {}", evt.getMessageID());
}
if (retain) {
@@ -227,11 +230,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic " + topic);
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retain, Integer messageID) {
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
for (final Subscription sub : subscriptions.matches(topic)) {
LOG.debug(String.format("Broker publishing to client <%s> topic <%s> qos <%s>",
sub.getClientId(), sub.getTopic(), qos));
LOG.debug("Broker publishing to client <{}> topic <{}> qos <{}>",
sub.getClientId(), sub.getTopic(), qos);
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS 0
sendPublish(sub.getClientId(), topic, qos, message, false);
@@ -256,19 +259,19 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained) {
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
sendPublish(clientId, topic, qos, message, retained, 0);
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained, int messageID) {
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) {
LOG.debug("notify invoked with event ");
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
LOG.info(String.format("send publish message to <%s> on topic <%s>, message is [%s]",
clientId, topic, new String(message)));
LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
pubMessage.setMessageID(messageID);
}
@@ -277,11 +280,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
if (m_clientIDs == null) {
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
}
LOG.debug("clientIDs are " + m_clientIDs);
LOG.debug("clientIDs are {}", m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs));
}
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession());
// m_clientIDs.get(clientId).getSession().write(pubMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage));
}catch(Throwable t) {
@@ -290,7 +293,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
private void sendPubRec(String clientID, int messageID) {
LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID));
LOG.debug("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
PubRecMessage pubRecMessage = new PubRecMessage();
pubRecMessage.setMessageID(messageID);
@@ -310,11 +313,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
if (m_clientIDs == null) {
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
}
LOG.debug("clientIDs are " + m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
LOG.debug("clientIDs are {}", m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
}
LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
// m_clientIDs.get(clientId).getSession().write(pubAckMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubAckMessage));
}catch(Throwable t) {
@@ -332,22 +335,20 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
final String topic = evt.getTopic();
final AbstractMessage.QOSType qos = evt.getQos();
final byte[] message = evt.getMessage();
boolean retain = evt.isRetain();
publish2Subscribers(topic, qos, message, retain, evt.getMessageID());
publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID());
m_storageService.removeQoS2Message(publishKey);
if (retain) {
m_storageService.storeRetained(topic, message, qos);
if (evt.isRetain()) {
m_storageService.storeRetained(topic, evt.getMessage(), qos);
}
sendPubComp(clientID, messageID);
}
private void sendPubComp(String clientID, int messageID) {
LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID));
LOG.debug("sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
PubCompMessage pubCompMessage = new PubCompMessage();
pubCompMessage.setMessageID(messageID);
@@ -357,7 +358,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
void processPubRec(String clientID, int messageID) {
//once received a PUBREC reply with a PUBREL(messageID)
LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID));
LOG.debug("processPubRec invoked for clientID {} ad messageID {}", clientID, messageID);
PubRelMessage pubRelMessage = new PubRelMessage();
pubRelMessage.setMessageID(messageID);
@@ -383,7 +384,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//de-activate the subscriptions for this ClientID
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
subscriptions.deactivate(clientID);
LOG.info(String.format("Disconnected client <%s> with clean session %s", clientID, cleanSession));
LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession);
}
void proccessConnectionLost(String clientID) {
@@ -392,7 +393,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//de-activate the subscriptions for this ClientID
subscriptions.deactivate(clientID);
LOG.info(String.format("Lost connection with client <%s>", clientID));
LOG.info("Lost connection with client <{}>", clientID);
}
}
@@ -410,13 +411,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
LOG.info(String.format("replying with UnsubAck to MSG ID %s", messageID));
LOG.info("replying with UnsubAck to MSG ID {}", messageID);
session.write(ackMessage);
}
void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) {
LOG.debug(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID()));
LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID());
for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
@@ -432,14 +433,14 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
for (int i = 0; i < msg.subscriptions().size(); i++) {
ackMessage.addType(AbstractMessage.QOSType.MOST_ONE);
}
LOG.debug("replying with SubAck to MSG ID " + msg.getMessageID());
LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID());
session.write(ackMessage);
}
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
LOG.info(String.format("<%s> subscribed to topic <%s> with QoS %s",
LOG.info("<{}> subscribed to topic <{}> with QoS {}",
newSubscription.getClientId(), topic,
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())));
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
@@ -451,13 +452,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
for (HawtDBStorageService.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic " + topic);
LOG.debug("send publish message for topic {}", topic);
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
}
}
private void disruptorPublish(OutputMessagingEvent msgEvent) {
LOG.debug("disruptorPublish publishing event on output " + msgEvent);
LOG.debug("disruptorPublish publishing event on output {}", msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+9 -11
View File
@@ -69,7 +69,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private void disruptorPublish(MessagingEvent msgEvent) {
LOG.debug("disruptorPublish publishing event " + msgEvent);
LOG.debug("disruptorPublish publishing event {}", msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
@@ -97,7 +97,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
LOG.info("onEvent processing messaging event from input ringbuffer" + evt);
LOG.info("onEvent processing messaging event from input ringbuffer {}", evt);
if (evt instanceof PublishEvent) {
m_processor.processPublish((PublishEvent) evt);
} else if (evt instanceof StopEvent) {
@@ -112,17 +112,15 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
if (message instanceof ConnectMessage) {
m_processor.processConnect(session, (ConnectMessage) message);
} else if (message instanceof PublishMessage) {
PublishMessage pubMsg = (PublishMessage) message;
PublishEvent pubEvt;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
if (message.getQos() == QOSType.MOST_ONE) {
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
} else {
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
}
pubEvt = new PublishEvent((PublishMessage) message, clientID, session);
// if (message.getQos() == QOSType.MOST_ONE) {
// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
//
// } else {
// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
// }
m_processor.processPublish(pubEvt);
} else if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+20 -6
View File
@@ -2,25 +2,39 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.server.ServerChannel;
/**
*
* @author andrea
*/
public class PublishEvent extends MessagingEvent implements Serializable {
public class PublishEvent extends MessagingEvent {
String m_topic;
QOSType m_qos;
byte[] m_message;
//byte[] m_message;
ByteBuffer m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
int m_msgID;
transient ServerChannel m_session;
public PublishEvent(PublishMessage pubMsg, String clientID, ServerChannel session) {
m_topic = pubMsg.getTopicName();
m_qos = pubMsg.getQos();
m_message = pubMsg.getPayload();
m_retain = pubMsg.isRetainFlag();
m_clientID = clientID;
m_session = session;
if (pubMsg.getQos() != QOSType.MOST_ONE) {
m_msgID = pubMsg.getMessageID();
}
}
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, ServerChannel session) {
m_topic = topic;
m_qos = qos;
@@ -30,7 +44,7 @@ public class PublishEvent extends MessagingEvent implements Serializable {
m_session = session;
}
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, int msgID, ServerChannel session) {
this(topic, qos, message, retain, clientID, session);
m_msgID = msgID;
@@ -44,7 +58,7 @@ public class PublishEvent extends MessagingEvent implements Serializable {
return m_qos;
}
public byte[] getMessage() {
public ByteBuffer getMessage() {
return m_message;
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/storage/StoredPublishEvent.java
+58
View File
@@ -0,0 +1,58 @@
package org.dna.mqtt.moquette.messaging.spi.impl.storage;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
/**
* Publish event serialized to the DB.
*
* @author andrea
*/
public class StoredPublishEvent implements Serializable {
String m_topic;
QOSType m_qos;
byte[] m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
int m_msgID;
public StoredPublishEvent(PublishEvent wrapped) {
m_topic = wrapped.getTopic();
m_qos = wrapped.getQos();
m_retain = wrapped.isRetain();
m_clientID = wrapped.getClientID();
m_msgID = wrapped.getMessageID();
ByteBuffer buffer = wrapped.getMessage();
m_message = new byte[buffer.remaining()];
buffer.get(m_message);
buffer.rewind();
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public byte[] getMessage() {
return m_message;
}
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public int getMessageID() {
return m_msgID;
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java
+3 -3
View File
@@ -72,7 +72,7 @@ public class SubscriptionsStore {
//reload any subscriptions persisted
LOG.debug("Reloading all stored subscriptions...");
for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic());
addDirect(subscription);
}
LOG.debug("Finished loading");
@@ -166,7 +166,7 @@ public class SubscriptionsStore {
}
public void activate(String clientID) {
LOG.debug(String.format("Activating subscriptions for clientID <%s>", clientID));
LOG.debug("Activating subscriptions for clientID <{}>", clientID);
subscriptions.activate(clientID);
}
@@ -176,7 +176,7 @@ public class SubscriptionsStore {
* listeners subscriptions, and not topic publishing.
*/
public List<Subscription> matches(String topic) {
List<Token> tokens = new ArrayList<Token>();
List<Token> tokens;
try {
tokens = splitTopic(topic);
} catch (ParseException ex) {
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
+1 -1
View File
@@ -85,7 +85,7 @@ public class NettyAcceptor implements ServerAcceptor {
MessageMetrics metrics = m_metricsCollector.computeMetrics();
//LOG.info(String.format("Bytes read: %d, bytes wrote: %d", metrics.readBytes(), metrics.wroteBytes()));
LOG.info(String.format("Msg read: %d, msg wrote: %d", metrics.messagesRead(), metrics.messagesWrote()));
LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote());
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java
+1 -1
View File
@@ -28,7 +28,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
try {
switch (msg.getMessageType()) {
case CONNECT:
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+5 -2
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -168,7 +169,8 @@ public class ProtocolProcessorTest {
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI", null);
m_processor.processPublish(pubEvt);
//Verify
@@ -254,7 +256,8 @@ public class ProtocolProcessorTest {
connectMessage.setProcotolVersion((byte)3);
connectMessage.setCleanSession(subscription.isCleanSession());
m_processor.processConnect(m_session, connectMessage);
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "FakeCLI", null);
m_processor.processPublish(pubEvt);
//Exercise
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
+2
View File
@@ -306,6 +306,7 @@ public class ServerIntegrationPahoTest {
anotherClient.disconnect();
}
//BUG
@Test
public void testPublishWithQoS2() throws Exception {
LOG.info("*** testPublishWithQoS2 ***");
@@ -324,6 +325,7 @@ public class ServerIntegrationPahoTest {
assertEquals(2, m_callback.getMessage().getQos());
}
//BUG
@Test
public void testPublishReceiveWithQoS2() throws Exception {
LOG.info("*** testPublishReceiveWithQoS2 ***");
netty_parser/pom.xml
+1 -1
View File
@@ -23,7 +23,7 @@
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-parser-commons</artifactId>
<version>0.2</version>
<version>${project.version}</version>
</dependency>
<dependency>
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java
+7 -5
View File
@@ -1,6 +1,7 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
@@ -18,14 +19,14 @@ class PublishDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
LOG.info("decode invoked with buffer " + in);
LOG.info("decode invoked with buffer {}", in);
in.resetReaderIndex();
int startPos = in.readerIndex();
//Common decoding part
PublishMessage message = new PublishMessage();
if (!decodeCommonHeader(message, in)) {
LOG.info("decode ask for more data after " + in);
LOG.info("decode ask for more data after {}", in);
in.resetReaderIndex();
return;
}
@@ -51,9 +52,10 @@ class PublishDecoder extends DemuxDecoder {
in.resetReaderIndex();
return;
}
byte[] b = new byte[payloadSize];
in.readBytes(b);
message.setPayload(b);
// byte[] b = new byte[payloadSize];
ByteBuf bb = Unpooled.buffer(payloadSize);
in.readBytes(bb);
message.setPayload(bb.nioBuffer());
out.add(message);
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java
+19 -3
View File
@@ -2,6 +2,8 @@ package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
@@ -65,7 +67,8 @@ public class PublishDecoderTest {
public void testHeaderWithMessageID_Payload() throws Exception {
m_buff = Unpooled.buffer(14);
int messageID = 123;
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
// byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
ByteBuffer payload = ByteBuffer.allocate(3).put(new byte[]{0x0A, 0x0B, 0x0C});
initHeaderWithMessageID_Payload(m_buff, messageID, payload);
//Exercise
@@ -76,7 +79,8 @@ public class PublishDecoderTest {
assertNotNull(message);
assertEquals("Fake Topic", message.getTopicName());
assertEquals(messageID, (int) message.getMessageID());
TestUtils.verifyEquals(payload, message.getPayload());
// TestUtils.verifyEquals(payload, message.getPayload());
assertEquals(payload, message.getPayload());
}
@Test
@@ -141,7 +145,9 @@ public class PublishDecoderTest {
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
assertEquals("/topic", message.getTopicName());
assertEquals("Test my payload", new String(message.getPayload()));
// assertEquals("Test my payload", new String(message.getPayload()));
Buffer expectedPayload = ByteBuffer.allocate(15).put("Test my payload".getBytes()).flip();
assertEquals(expectedPayload, message.getPayload());
}
private void initHeader(ByteBuf buff) throws IllegalAccessException {
@@ -170,6 +176,16 @@ public class PublishDecoderTest {
buff.writeBytes(tmp);
}
private void initHeaderWithMessageID_Payload(ByteBuf buff, int messageID, ByteBuffer payload) throws IllegalAccessException {
ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic"));
tmp.writeShort(messageID);
tmp.writeBytes(payload);
buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1
.writeBytes(Utils.encodeRemainingLength(tmp.readableBytes()));
//topic name
buff.writeBytes(tmp);
}
private ByteBuf generatePublishQoS0(ByteBuf payload) throws IllegalAccessException {
int size = payload.capacity();
ByteBuf messageBody = Unpooled.buffer(size);
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java
+18 -10
View File
@@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import java.nio.ByteBuffer;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
@@ -35,7 +36,8 @@ public class PublishEncoderTest {
msg.setTopicName(topic);
//variable part
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C};
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip();
msg.setPayload(payload);
//Exercise
@@ -48,7 +50,8 @@ public class PublishEncoderTest {
//Variable part
verifyString(topic, m_out);
verifyBuff(payload.length, payload, m_out);
payload.rewind();
verifyBuff(bpayload.length, payload, m_out);
}
@@ -61,7 +64,8 @@ public class PublishEncoderTest {
msg.setTopicName(topic);
//variable part
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C};
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip();
msg.setPayload(payload);
//Exercise
@@ -76,7 +80,8 @@ public class PublishEncoderTest {
verifyString(topic, m_out);
assertEquals(0, m_out.readByte()); //MessageID MSB
assertEquals(1, m_out.readByte()); //MessageID LSB
verifyBuff(payload.length, payload, m_out);
payload.rewind();
verifyBuff(bpayload.length, payload, m_out);
}
@Test(expected = IllegalArgumentException.class)
@@ -120,7 +125,8 @@ public class PublishEncoderTest {
msg.setTopicName(null);
//variable part
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C};
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip();
msg.setPayload(payload);
ByteBuf out = Unpooled.buffer();
@@ -136,7 +142,8 @@ public class PublishEncoderTest {
msg.setTopicName(topic);
//variable part
byte[] payload = "Test my payload".getBytes();
byte[] bpayload = "Test my payload".getBytes();
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(bpayload.length).put(bpayload).flip();
msg.setPayload(payload);
//Exercise
@@ -145,16 +152,17 @@ public class PublishEncoderTest {
//Verify
//2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload
int size = m_out.readableBytes();
assertEquals(10+payload.length, size);
assertEquals(10 + bpayload.length, size);
assertEquals(0x30, m_out.readByte()); //1 byte
assertEquals(23, m_out.readByte()); //1 byte the length
}
@Test
public void testEncodeBigMessage() throws Exception {
byte[] payload = new byte[256];
byte[] bpayload = new byte[256];
TestUtils.generateRandomPayload(256).readBytes(payload);
TestUtils.generateRandomPayload(256).readBytes(bpayload);
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(bpayload.length).put(bpayload).flip();
String topic = "/topic";
PublishMessage msg = new PublishMessage();
@@ -168,7 +176,7 @@ public class PublishEncoderTest {
m_encoder.encode(m_mockedContext, msg, m_out);
int size = m_out.readableBytes();
assertEquals(11+payload.length, size);
assertEquals(11 + bpayload.length, size);
assertEquals(0x30, m_out.readByte()); //1 byte
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java
+6
View File
@@ -6,6 +6,7 @@ import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.*;
@@ -31,6 +32,11 @@ public class TestUtils {
}
}
static void verifyBuff(int numBytes, ByteBuffer bytes, ByteBuf buff) {
assertTrue(numBytes <= buff.readableBytes());
assertEquals(bytes, buff.nioBuffer());
}
/**
* Verify the presence of the given string starting from the current position
* inside the buffer.
parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
+5 -3
View File
@@ -1,5 +1,7 @@
package org.dna.mqtt.moquette.proto.messages;
import java.nio.ByteBuffer;
/**
*
* @author andrea
@@ -8,7 +10,7 @@ public class PublishMessage extends MessageIDMessage {
private String m_topicName;
// private Integer m_messageID; //could be null if Qos is == 0
private byte[] m_payload;
private ByteBuffer m_payload;
/*public Integer getMessageID() {
return m_messageID;
@@ -30,11 +32,11 @@ public class PublishMessage extends MessageIDMessage {
this.m_topicName = topicName;
}
public byte[] getPayload() {
public ByteBuffer getPayload() {
return m_payload;
}
public void setPayload(byte[] payload) {
public void setPayload(ByteBuffer payload) {
this.m_payload = payload;
}
}
tools_scripts/fuse_publisher.groovy
+27
View File
@@ -0,0 +1,27 @@
@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5')
import java.net.URISyntaxException
import org.fusesource.mqtt.client.BlockingConnection
import org.fusesource.mqtt.client.MQTT
import org.fusesource.mqtt.client.QoS
String host = args[0]
MQTT mqtt = new MQTT()
//mqtt.setHost("test.mosquitto.org", 1883);
mqtt.setHost(host, 1883)
BlockingConnection connection = mqtt.blockingConnection()
connection.connect()
print "publishing.."
long startTime = System.currentTimeMillis()
(1..100000).each {
connection.publish("/topic", 'Hello world!!'.bytes, QoS.AT_MOST_ONCE, false)
}
connection.publish('/exit', 'Exit'.bytes, QoS.AT_MOST_ONCE, false)
long stopTime = System.currentTimeMillis()
long spentTime = stopTime -startTime
println "published in ${spentTime} ms"
connection.disconnect()
println "disconnected"
tools_scripts/fuse_subscriber.groovy
+51
View File
@@ -0,0 +1,51 @@
@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5')
import java.net.URISyntaxException
import org.fusesource.mqtt.client.BlockingConnection
import org.fusesource.mqtt.client.MQTT
import org.fusesource.mqtt.client.Message
import org.fusesource.mqtt.client.QoS
import org.fusesource.mqtt.client.Topic
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
String host = args[0]
MQTT mqtt = new MQTT()
//mqtt.setHost("test.mosquitto.org", 1883);
mqtt.setHost(host, 1883)
mqtt.setCleanSession(true)
mqtt.setHost(host, 1883)
mqtt.setClientId("SubscriberClient")
BlockingConnection connection = mqtt.blockingConnection()
connection.connect()
println "Connected to ${host}"
Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE), new Topic("/exit", QoS.AT_MOST_ONCE)]
byte[] qoses = connection.subscribe(topics)
println "subscribed to /topic qos: ${qoses[0]}"
println "subscribed to /exit qos: ${qoses[1]}"
int numReceived = 0
long startTime = System.currentTimeMillis()
boolean exit = false
while(!exit) {
Message message = connection.receive()
if (message.topic == "/exit") {
exit = true
} else {
numReceived++
// if ((numReceived % 10000) == 0) {
// print '.'
// }
}
message.ack()
}
connection.disconnect()
long stopTime = System.currentTimeMillis()
long spentTime = stopTime - startTime
println "/exit received"
println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms"