Hi All,
Following describes the current state/implementation of MQTT transport
integrated into the MB,
We use the moquette MQTT library as the protocol engine which could be
found in [1], which will accept and establish connections between MQTT
subscribers and publishers. MQTT has mainly three levels of QOS which are
0,1,2.
QOS 0 level would be similar to fire and forget scenario where the
acknowledgments will not be expected, also messages will not be ensured to
delivered 100%, this level of QOS is performance friendly in comparison to
other QOS levels but less reliable. QOS 1 would be a more guaranteed
delivery approach, where it will ensure to retry if an acknowledgment
doesn't arrive. but messages could be duplicated. QOS2 will ensure that the
message is delivered only once.
Current revision of MB 3.0.0 will accept and deliver MQTT messages in all
three QOS levels when running in the stand-alone mode. Also will support
MQTT messages in QOS level 0 in the clustered mode. The listed below are
some of the main classes which can be used as a reference,
1) Initialization of the server opening of ports done through the Main
class startMQTTBroker() method. This will open up a connection through a
Netty server where the connection will be NIO
2) When a message arrives the generalized flow of events would be performed
through the following classes (in summery),
ConnectDecoder -> NettyMQTTHandler -> ProtocolProcessor.
3) NettyMQTTHandler class would basically evaluate each message and
delegate it to ProtocolProcessor as events ( ex :-
subscribe/unsubscribe/publish) each events would be placed in a distruptor
ring.
4) ProtocolProcessor also maintains the subscriber channels. When message
is published the relevant subscribers will be notified through placing the
outcome also to distruptor ring, this way both the inbound and outbound
events will be handled though distruptor.
5) All the methods which are interfaced between the Andes Kernal and MQTT
library are in the class AndesBridge.
6) MQTTChannel class would basically handle between the data
insertion/removal to/from Cassandra queue.
Also note that per topic only one Andes level LocalSubscription will be
created, when the LocalSubscription class is notified of events it will
pass it to the library to notify the relevant subscribers. MB was tested
against the test suite in [2] which almost covers all the protocol level
validations. The test reports indicated failure in cases which involved
topics with WildCard characters. The issue was due to the Andes Kernal not
interpreting the characters properly, however there's a fix which was done
in the MB 2.2.0 version which need to be ported to 3.0.0.
To-Dos in terms of MQTT/MB integration
# Need to address QOS levels 1 and 2 for the clustered environment.
Currently its handled through using an in-memory DB for persistence.
Ideally a hybrid approach of having both the in-memory and Cassandra would
allow a performance gain
# Support topics with WildCard characters (this will be fixed once the
patch of 2.2.0 is ported to 3.0.0)
# Handling topics with clean session (durable subscriptions in terms of
AMQP). This needs to be discussed and handled in a common way for both MQTT
and AMQP.
# Integrating of users/role management policies applicable for MQTT.
Currently username and password is retrieved from a configuration file.
# Need to test MQTT cluster wide.
[1] https://code.google.com/p/moquette-mqtt/
[2] http://wiki.eclipse.org/File:Mqtt-test.zip
Thanks,
Pamod
--
*Pamod Sylvester *
*Senior Software Engineer *
Integration Technologies Team, WSO2 Inc.; http://wso2.com
email: [email protected] cell: +94 77 7779495
_______________________________________________
Dev mailing list
[email protected]
http://wso2.org/cgi-bin/mailman/listinfo/dev
