MQTT source and sink Connector, version 4.0.0

MQTT

If you already have a publicly accessible deployment of MQTT, you can skip ahead and Configure and install a connector, using the Axual Self-Service UI. Use the configuration instructions below.
If you don’t have one available, the following section instructs you to deploy a publicly available instance of MQTT.

Deploying a MQTT instance

  1. Sign up for a free trial of Hive MQTT Broker.

  2. After successful registration, a free cluster should be automatically created in your cluster overview. Note down the broker’s URL and port. we’ll use them later, as part of the connect.mqtt.hosts connector-config.

  3. Click "Manage Cluster" and then click the "Access Management" button in the top-right of the screen. Create a user and note down its credentials: we’ll use them later to connect to the cluster, for the connect.mqtt.username and connect.mqtt.password connector-config.

  4. We’ll use the HiveMQ WebSocket Client to create sample data for our MQTT cluster. Fill the following fields:

    • Host: The cluster URL you saw in the cluster overview page (which you noted down for the connect.mqtt.hosts config)

    • Port: 8884 (this is a special port opened for this browser-client)

    • ClientID: can use the default provided value

    • Username: value you just created

    • Password: value you just created

    • SSL: tick the box

  5. Click "Connect".

  6. Expand the "Subscriptions" section. Click "Add New Topic Subscription". In the pop-up window, edit only the "Topic" field: my_mqtt_topic, then click "Subscribe".

  7. Expand the "Publish" section. Use my_mqtt_topic as the topic name (without a trailing '/1') and write some text in the message box, then click "Publish". You should see the message appear in the "Messages" section below.

  8. Keep this page open. We’ll return later to produce more events, after configuring the Connector-Application in the self-service portal.

Configuring a new source Connector

  1. Follow the Creating streams documentation in order to create one stream and deploy it onto an environment.
    The name of the stream will be my_mqtt_kafka_topic.
    The key/value types will be String/String.

  2. Follow the Configure and install a connector documentation to set up a new Connector-Application.
    Let’s call it my_mqtt_app.
    The plugin name is com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector.
    If a plugin isn’t available, ask a platform operator to install plugins.

    For advanced configuration, see the official source connector documentation.

  3. Provide the following minimal configuration in order to connect to the previously configured MQTT instance.

    connect.mqtt.hosts
    Format:
    ssl://<MQTT-Broker-URL>:<MQTT-Broker-Port>

    Example value:
    ssl://mylonguuidfromhivemq.s1.eu.hivemq.cloud:8883

    connect.mqtt.kcql

    		 
    INSERT INTO my_mqtt_kafka_topic SELECT * FROM my_mqtt_topic

    connect.mqtt.service.quality

    1

    connect.mqtt.username

    Name of user you created

    connect.mqtt.password

    Password of user you created

  4. Authorize the my_mqtt_app source Connector-Application to produce to the my_mqtt_kafka_topic stream.

  5. You can now start the source Connector-Application.
    The Connector we just created will not read events produced in the past.

  6. Return to the HiveMQ WebSocket Client tab and produce some more events.

  7. You can now check the my_mqtt_kafka_topic stream to see the events published by the Connector.
    The values will be base64 encoded.

Cleanup

Once you are done, stop the Connector-Application and cleanup the unused axual resources.

Don’t forget to return to HiveMQ’s cluster overview → Manage Clusters → and delete your cluster if you’re not using it anymore.

License

MQTT source and sink-Connector is licensed under the Apache License, Version 2.0.

Source code

The source code for the Connect-Plugin can be found at github.com/lensesio/stream-reactor.