hypi-universe/mop

MQTT on Pulsar implemented using Pulsar Protocol Handler

MQTT on Pulsar (MoP)

MQTT-on-Pulsar (aka MoP) is developed to support MQTT protocol natively on Apache Pulsar.

Get started

Download or build MoP protocol handler

1. Clone the MoP project from GitHub to your local.

   git clone https://github.com/streamnative/mop.git
   cd mop

2. Build the project.

   mvn clean install -DskipTests

3. The NAR file can be found at this location.

   ./mqtt-impl/target/pulsar-protocol-handler-mqtt-${version}.nar

Install MoP protocol handler

Configure the Pulsar broker to run the MoP protocol handler as a plugin by adding configurations to the Pulsar configuration file, such as broker.conf or standalone.conf.

1. Set the configuration of the MoP protocol handler.

   Add the following properties and set their values in the Pulsar configuration file, such as conf/broker.conf or conf/standalone.conf.

   Property	Suggested value	Default value
   messagingProtocols	mqtt	null
   protocolHandlerDirectory	Location of MoP NAR file	./protocols

   Example

   messagingProtocols=mqtt
   protocolHandlerDirectory=./protocols
   

2. Set the MQTT server listeners.

   Note

   The hostname in listeners should be the same as Pulsar broker's advertisedAddress.

   Example

   mqttListeners=mqtt://127.0.0.1:1883
   advertisedAddress=127.0.0.1
   

Load MoP protocol handler

After you install the MoP protocol handler to Pulsar broker, you can restart the Pulsar brokers to load the MoP protocol handler.

How to use Proxy

To use the proxy, follow the following steps. For detailed steps, refer to Deploy a cluster on bare metal.

1. Prepare a ZooKeeper cluster.

2. Initialize the cluster metadata.

3. Prepare a BookKeeper cluster.

4. Copy the pulsar-protocol-handler-mqtt-${version}.nar to the $PULSAR_HOME/protocols directory.

5. Start the Pulsar broker.

   Here is an example of the Pulsar broker configuration.

   messagingProtocols=mqtt
   protocolHandlerDirectory=./protocols
   brokerServicePort=6651
   mqttListeners=mqtt://127.0.0.1:1883
   advertisedAddress=127.0.0.1
   
   mqttProxyEnable=true
   mqttProxyPort=5682

Verify MoP protocol handler

There are many MQTT client that can be used to verify the MoP protocol handler, such as MQTTBox, MQTT Toolbox. You can choose a CLI tool or interface tool to verify the MoP protocol handler.

The following example shows how to verify the MoP protocol handler with FuseSource MqttClient.

1. Add the dependency.

   <dependency>
       <groupId>org.fusesource.mqtt-client</groupId>
       <artifactId>mqtt-client</artifactId>
       <version>1.16</version>
   </dependency>

2. Publish messages and consume messages.

   MQTT mqtt = new MQTT();
   mqtt.setHost("127.0.0.1", 1883);
   BlockingConnection connection = mqtt.blockingConnection();
   connection.connect();
   Topic[] topics = { new Topic("persistent://public/default/my-topic", QoS.AT_LEAST_ONCE) };
   connection.subscribe(topics);
   
   // publish message
   connection.publish("persistent://public/default/my-topic", "Hello MOP!".getBytes(), QoS.AT_LEAST_ONCE, false);
   
   // receive message
   Message received = connection.receive();