This repository comprises the rudimentary API extracted from the OSGi RFC 246 specification and the MQTT 5.0 implementation of it. This spec is not at all the official version as it is based on the preliminary API from the current version of the spec. This is just an experimental repository for OSGi Messaging specification since the official version is not yet released. The official version was expected to be released with the release of OSGi Enterprise R8 specification in the second quarter of 2021 (as mentioned during the EclipseCON talk). However, the official version is yet to be released.

Further to the above, there is a utility bundle comprising an easy-to-use functionality to manage remote resources or edge devices using MQTT 5.0. For more details, have a look at the Remote Resource (Edge Device) Management section.

🚀 Hot off the press! Grab the latest artifacts for your OSGi runtime:

1. Java 8
2. OSGi R8

This project comprises three projects -

To use it in the OSGi environment, you need to install in.bytehue.messaging.mqtt5.provider. If you want to make use of the remote device management, you also need to install in.bytehue.messaging.mqtt5.remote.adapter.

Import as Eclipse Projects

1. Install Bndtools
2. Import all the projects (File -> Import -> General -> Existing Projects into Workspace)

Run ./gradlew clean build in the project root directory

The in.bytehue.messaging.client PID can be used to configure the client. The configurable properties are as follows:

The in.bytehue.messaging.subscriber PID can be used to configure the subscriber.

The in.bytehue.messaging.publisher PID can be used to configure the publisher.

The in.bytehue.messaging.publisher.replyto PID can be used to configure the internal thread pool of the reply-to publisher.

The in.bytehue.messaging.mqtt5.provider.MessageSubscriptionRegistry PID can be used to configure the subscription registry.

The in.bytehue.messaging.whiteboard PID can be used to configure the reply-to whiteboard.

The in.bytehue.mqtt.debug PID can be used to configure the console log mirror.

The in.bytehue.mqtt.connection.ready PID identifies the ConnectionReadyService which registers a condition service when the MQTT connection is established. This service exposes osgi.condition.id=mqtt-ready condition.

The ConnectionReadyService is exposed as an OSGi condition under the mentioned config PID as it enables consumers to configure osgi.ds.satisfying.condition.target. For example, one might want to delay the registration of the connection ready service if some specific services are up and running or the OSGi runtime is ready and ACTIVE.

The included APIs also provide several helpful annotations for implementors as well as consumers. Refer to org.osgi.service.messaging.propertytypes and org.osgi.service.messaging.annotations packages.

The OSGi messaging specification is catered to provide a unified solution to accomodate the trending messaging solutions such as AMQP, MQTT etc. That's why no generic API exists for MQTT. To fill in this gap, an implementation specific API exists in in.bytehue.messaging.mqtt5.api.

• Simple Pub/Sub Example:

@Component
public final class Mqtt5PubSubExample {

    private PushStream<Message> stream;

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private MessagePublisher publisher;

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private MessageSubscription subscriber;

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private ComponentServiceObjects<MessageContextBuilder> mcbFactory;

    @Deactivate
    void deactivate() {
        stream.close();
    }

    public String sub(final String channel) {
        stream = subscriber.subscribe(channel);
        stream.forEach(m -> {
            System.out.println("Message Received");
            System.out.println(StandardCharsets.UTF_8.decode(m.payload()).toString());
        });
        return "Subscribed to " + channel;
    }

    public String pub(final String channel, final String data) {
        final MessageContextBuilder mcb = mcbFactory.getService();
        try {
            // @formatter:off
            publisher.publish(
                    mcb.content(ByteBuffer.wrap(data.getBytes()))
                       .channel(channel)
                       .buildMessage());
            // @formatter:on
        } finally {
            mcbFactory.ungetService(mcb);
        }
        return "Published to " + channel;
    }
}

• Simple Reply-To Publish Example

@Component
public final class Mqtt5ReplyToExample {

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private ReplyToPublisher mqttPublisher;

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private ComponentServiceObjects<MessageContextBuilder> mcbFactory;

    public void publishReplyToMessage() {
        final MessageContextBuilder mcb = mcbFactory.getService();
        try {
            // @formatter:off
            final Message request =
                    mcb.channel("/demo")
                       .correlationId("test123")
                       .replyTo("demo_response")
                       .content(ByteBuffer.wrap("Hello Word!".getBytes()))
                       .buildMessage();
            // @formatter:on
            mqttPublisher.publishWithReply(request).onSuccess(System.out::println);
        } finally {
            mcbFactory.ungetService(mcb);
        }
    }
}

• Simple Request-Response Multiplexer Example

@Component
public final class Mqtt5MultiplexerExample {

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private MqttRequestMultiplexer multiplexer;

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private ComponentServiceObjects<MessageContextBuilder> mcbFactory;

    public void request(final String payload) {
        final MessageContextBuilder mcb = mcbFactory.getService();
        try {
            // The Request must go to a SPECIFIC topic
            final Message request = mcb.channel("my/request/topic")
                                       .replyTo("my/response/topic/123") // Specific reply topic
                                       .correlationId(UUID.randomUUID().toString())
                                       .content(ByteBuffer.wrap(payload.getBytes()))
                                       .buildMessage();

            // The Context defines the SHARED subscription (Wildcard allowed)
            final MessageContext subCtx = mcb.channel("my/response/topic/#")
                                             .buildContext();

            multiplexer.request(request, subCtx)
                       .timeout(5000)
                       .onSuccess(response -> System.out.println("Response received"));
        } finally {
            mcbFactory.ungetService(mcb);
        }
    }
}

• Simple Reply-To Single Subscription Handler Example:

@Component
@ReplyToSubscription(target = "(&(osgi.messaging.protocol=mqtt5)(osgi.messaging.name=mqtt5-hivemq-adapter)(osgi.messaging.feature=replyTo)))", channel = "a/b", replyChannel = "c/d")
public final class Mqtt5ReplyToSingleSubscriptionHandler implements ReplyToSingleSubscriptionHandler {

    @Override
    public Message handleResponse(final Message requestMessage, final MessageContextBuilder responseBuilder) {
        final String content = StandardCharsets.UTF_8.decode(requestMessage.payload()).append("EXAMPLE").toString();
        return responseBuilder.content(ByteBuffer.wrap(content.getBytes())).buildMessage();
    }

}

• Simple Message Acknowledgement Example:

@Component
public final class Mqtt5AckExample {

    @Reference(target = "(&(osgi.messaging.protocol=mqtt5)(osgi.messaging.feature=acknowledge))")
    private MessageSubscription mqttSubscription;

    @Reference(target = "(&(osgi.messaging.protocol=mqtt5)(osgi.messaging.feature=acknowledge))")
    private ComponentServiceObjects<AcknowledgeMessageContextBuilder> amcbFactory;

    public void acknowledge() {
        final AcknowledgeMessageContextBuilder ackBuilder = amcbFactory.getService();
        try {
            final MessageContext context = ackBuilder
                    .handleAcknowledge("(foo=bar)")
                    .postAcknowledge(m -> {
                            final AcknowledgeMessageContext ctx = (AcknowledgeMessageContext) m.getContext();
                            System.out.println("Acknowledge state is: " + ctx.getAcknowledgeState());
                      })
                    .messageContextBuilder()
                    .channel("/demo")
                    .buildContext();
            mqttSubscription.subscribe(context);
        } finally {
            amcbFactory.ungetService(ackBuilder);
        }
    }
}

• Updating LWT dynamically by sending publish request:

@Component
public final class Mqtt5LwtPublish {

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private MessagePublisher publisher;

    @Reference(target = "(osgi.messaging.protocol=mqtt5)")
    private ComponentServiceObjects<MessageContextBuilder> mcbFactory;

    public void lastWillPublish() {
        final MessageContextBuilder mcb = mcbFactory.getService();
        try {
            publisher.publish(
                    mcb.content(ByteBuffer.wrap("CLOSED_CONNECTION".getBytes()))
                       .channel("last/will/topc/example")
                       .extensionEntry(EXTENSION_QOS, 2)
                       .extensionEntry(EXTENSION_LAST_WILL, true)
                       .extensionEntry(LAST_WILL_DELAY_INTERVAL, 30)
                       .extensionEntry(MESSAGE_EXPIRY_INTERVAL, 30)
                       .buildMessage());
        } finally {
            mcbFactory.ungetService(mcb);
        }
    }

}

• Note that, the connection to the MQTT broker can be terminated anytime due to network issues. In such a case, you can track the availability of a connection to the broker using an OSGi service which gets registered if the connection to the broker is maintained. The service will disappear if the connection gets broken. This service contains mqtt.connection.ready property that is set to true. Also note that, the service is exported under TargetCondition marker interface (Refer to Target Condition Satisfiability for MQTT client below)
• Since more than one implementations can coexist in the OSGi runtime, we can search for the MQTT services by means of the provided service properties.
• Refer to the examples above.
• Also note that, the in.bytehue.messaging.mqtt5.provider bundle packages the APIs and implementation together. This bundle also packages and exports the HiveMQ Java client APIs to perform enhanced configuration to the client.
• For more details, have a look at the example project

In certain circumstances the MQTT client requires few services to be up and running before the client is connected to the broker. For example, you could provide the following services for the client to use before it connects to the server:

• javax.net.ssl.TrustManagerFactory
• in.bytehue.messaging.mqtt5.api.MqttMessageCorrelationIdGenerator
• com.hivemq.client.mqtt.lifecycle.MqttClientConnectedListener
• com.hivemq.client.mqtt.lifecycle.MqttClientDisconnectedListener
• com.hivemq.client.mqtt.mqtt5.auth.Mqtt5EnhancedAuthMechanism
• com.hivemq.client.mqtt.mqtt5.advanced.interceptor.qos1.Mqtt5IncomingQos1Interceptor
• com.hivemq.client.mqtt.mqtt5.advanced.interceptor.qos2.Mqtt5IncomingQos2Interceptor
• com.hivemq.client.mqtt.mqtt5.advanced.interceptor.qos1.Mqtt5OutgoingQos1Interceptor
• com.hivemq.client.mqtt.mqtt5.advanced.interceptor.qos2.Mqtt5OutgoingQos2Interceptor

In such scenario, consumers can implement in.bytehue.messaging.mqtt5.api.TargetCondition marker API to provide services with properties that can be used as conditional target to the MQTT client (refer to condition.target property in in.bytehue.messaging.client configuration. That means, consumer can provide filters that should be satisfied before MQTT client is up and running.

Assume we have a service R that represents a link to a remote service. For performance and reliability reasons, we require at least 3 of those services to be present before we can start the MQTT client. Additionally, these services must come from at least 2 different regions. For this reason, we define a property region that can take the values south, east, north, and west.

You can, therefore, use this filter: (&(#>=3)([unq]region>=2))

You can make use of the following extended calculated values:

This will ensure that your services will be up and running before the client gets activated. This also guarantees that the start order of the bundles is not at all required in this scenario.

The bundle provides several Gogo commands to interact with the MQTT client and inspect its state. These commands are available under the mqtt scope.

You can extend the mqtt:runtime command output by implementing the in.bytehue.messaging.mqtt5.api.MqttCommandExtension interface and registering it as an OSGi service.

@Component
public class MyCustomMqttInfo implements MqttCommandExtension {

    @Override
    public String rowName() {
        return "My Custom Metric";
    }

    @Override
    public String rowValue() {
        return "Some Value";
    }
}

This will add a new row to the mqtt:runtime output table.

This comprises the guidelines to structure your MQTT topic namespace for managing the remote resources or edge devices using MQTT.

The remote resources can receive two different types of requests:

• Command to perform an action (or popularly known as Request/Response pattern)
• Unsolicited events when the remote resource or edge device reports status of something periodically

In MQTT 5.0, adding a response topic to the MQTT publish request enables the subscriber to reply to that specific topic. But there is no such standard or practice that has been advised by the MQTT specification. It is, therefore, left to the user to introduce own convention to follow.

In this section, I will propose an efficient yet flexible way of performing request-response communication using MQTT 5.0.

A publisher or popularly known as the requester can send a MQTT payload to the topic conforming to the following pattern:

control-topic-prefix/control-topic/client-id/application-id/method/resource-id

for example,

• CTRL/com/company/ABCD-1234/CONF-V1/GET/configurations
• CTRL/com/company/ABCD-1234/CONF-V1/PUT/configurations/a.b.c.d
• CTRL/com/company/ABCD-1234/APP-V1/GET/sensors/temperature
• CTRL/com/company/ABCD-1234/COMMAND-V1/EXEC/watchdog/monitor

Let's first discuss the pattern mentioned above to understand the workflow better:

The provider emits OSGi EventAdmin events to notify about the status of MQTT subscription requests. These events are useful for tracking whether a subscription was successfully acknowledged by the broker, rejected, or timed out.

Event Topics:

• mqtt/subscription/ACKED: The subscription was successfully acknowledged by the broker (SUBACK received with success codes).
• mqtt/subscription/FAILED: The subscription was rejected by the broker (SUBACK received with failure codes).
• mqtt/subscription/NO_ACK: No acknowledgement was received within the configured timeout (timeout or client-side error).

Event Properties:

A requester can read resources from the remote edge device by sending a GET MQTT request to the following topic pattern:

control-topic-prefix/control-topic/client-id/application-id/GET/resource-id

For example,

• CTRL/com/company/ABCD-1234/CONF-V1/GET/configurations
• CTRL/com/company/ABCD-1234/CONF-V1/GET/bundles
• CTRL/com/company/ABCD-1234/APP-V1/GET/sensors/dining_room/temperature

Creating resources can be done by sending a PUT request to the following topic pattern:

control-topic-prefix/control-topic/client-id/application-id/PUT/resource-id

For example,

• CTRL/com/company/ABCD-1234/CONF-V1/PUT/configurations/c.d.e.f

Updating resources can be achieved by sending a POST request to the following topic pattern:

control-topic-prefix/control-topic/client-id/application-id/POST/resource-id

For example,

• CTRL/com/company/ABCD-1234/CONF-V1/POST/configurations/c.d.e.f
• CTRL/com/company/ABCD-1234/APP-V1/POST/sensors/bedroom/temperature

Similarly, any requester can delete resources on the remote edge device by sending a DELETE request to the following topic pattern:

control-topic-prefix/control-topic/client-id/application-id/DELETE/resource-id

for example,

• CTRL/com/company/ABCD-1234/CONF-V2/DELETE/configurations/c.d.e.f
• CTRL/com/company/ABCD-1234/CONF-V2/DELETE/nodes/my_node

You can also execute remote resources by sending a EXEC request to the following topic pattern:

control-topic-prefix/control-topic/client-id/application-id/EXEC/resource-id

for example,

• CTRL/com/company/ABCD-1234/COMMAND-V1/EXEC/ifconfig
• CTRL/com/company/ABCD-1234/DEPLOY-V2/EXEC/com.company.bundle/start
• CTRL/com/company/ABCD-1234/DEPLOY-V2/EXEC/threads/my_thread/watch

1. The receiving or subscribing application would always reply to the reply to address with any kind of content (or popularly known as payload) that both the parties (publisher and subscriber) understand. It can be (de)/serialized using Protobuf or JSON or XML, or any other serializer.

The response contains the content if available. In addition, there also exist some other user properties that denote the status of the response:

The following properties will be available in the user properties of the MQTT response:

• response.code - The available response codes are:

• response.exception.message - Optional exception message or the string version of the exception itself if there is no message available

2. The requester should always provide a correlation ID in the request message
3. The remote resource or the responder (subscriber) doesn't care about the correlation ID at all since the correlation ID will be mapped automatically when replying back to the requester
4. The remote resource or the responder (subscriber) remains always data agnostic and just cares about the resource that needs to accessed by the requester and the request message payload (Refer to the example below)
5. The response codes and messages must never be added by the remote resource application explicitly. It will automatically be done internally while replying to the requester.
6. It is recommended that the requester sets a timeout to the requested message to control the amount of time that it waits for a response from the remote resource or edge device. If a response is not received within the timeout interval, the server can expect that either the edge device or the resource is offline.
7. It is recommended to never use the control topic for unsolicited events where the remote resource or edge device periodically sends updates

Any remote device can introduce MQTT application to leverage the remote resource management functionality.

You just need to implement in.bytehue.messaging.mqtt5.remote.api.MqttApplication.

@Component
@MqttApplicationId("APP-V1")
public final class MyMqttApplicationExample implements MqttApplication {

    @Override
    public Message doGET(
            final String resource,
            final Message requestMessage, 
            final MessageContextBuilder messageBuilder) throws Exception {
        return messageBuilder.content(ByteBuffer.wrap("RESPONSE".getBytes())).buildMessage();
    }

    @Override
    public Message doPUT(
            final String resource,
            final Message requestMessage, 
            final MessageContextBuilder messageBuilder) throws Exception {
        final String[] res = resource.split("/");
        if (res[0].equalsIgnoreCase("my-apps")) {
            String resourceId = res[1];
            String status = udpateResource();
            return messageBuilder.content(ByteBuffer.wrap(status.getBytes())).buildMessage();
        }
        throw new IllegalStateException("Specified resource cannot be updated");
    }

    @Override
    public Message doEXEC(
            final String resource,
            final Message requestMessage, 
            final MessageContextBuilder messageBuilder) throws Exception {
        final String[] res = resource.split("/");
        if (res[0].equalsIgnoreCase("my-commands")) {
            String command = res[1];
            String response = executeCommand(command);
            return messageBuilder.content(ByteBuffer.wrap(response.getBytes())).buildMessage();
        }
        throw new IllegalStateException("Specified command cannot be executed");
    }
    
    private static String updateResource(String resource) {
        .....
    }

    private static String executeCommand(String resource) {
        .....
    }

}

The following is the API of in.bytehue.messaging.mqtt5.remote.api.MqttApplication:

public interface MqttApplication {

    /**
     * The service property denoting the application identifier
     */
    String APPLICATION_ID_PROPERTY = "mqtt.application.id";

    /**
     * Used to implement a READ request for a resource
     *
     * @param resource the resource identifier
     * @param requestMessage the received message
     * @param messageBuilder the builder to build the response message
     *
     * @return the response to be provided back as {@link Message}
     * @throws Exception
     *             An exception is thrown in every condition where the request cannot be full fitted due to wrong
     *             request parameters or exceptions during processing
     */
    default Message doGET(
            final String resource,
            final Message requestMessage,
            final MqttMessageContextBuilder messageBuilder) throws Exception {
        throw new UnsupportedOperationException();
    }

    /**
     * Used to implement a CREATE request for a resource
     *
     * @param resource the resource identifier
     * @param requestMessage the received message
     * @param messageBuilder the builder to build the response message
     *
     * @return the response to be provided back as {@link Message}
     * @throws Exception
     *             An exception is thrown in every condition where the request cannot be full fitted due to wrong
     *             request parameters or exceptions during processing
     */
    default Message doPUT(
            final String resource,
            final Message requestMessage,
            final MqttMessageContextBuilder messageBuilder) throws Exception {
        throw new UnsupportedOperationException();
    }

    /**
     * Used to implement an UPDATE request for a resource
     *
     * @param resource the resource identifier
     * @param requestMessage the received message
     * @param messageBuilder the builder to build the response message
     *
     * @return the response to be provided back as {@link Message}
     * @throws Exception
     *             An exception is thrown in every condition where the request cannot be full fitted due to wrong
     *             request parameters or exceptions during processing
     */
    default Message doPOST(
            final String resource,
            final Message requestMessage,
            final MqttMessageContextBuilder messageBuilder) throws Exception {
        throw new UnsupportedOperationException();
    }

    /**
     * Used to implement a DELETE request for a resource
     *
     * @param resource the resource identifier
     * @param requestMessage the received message
     * @param messageBuilder the builder to build the response message
     *
     * @return the response to be provided back as {@link Message}
     * @throws Exception
     *             An exception is thrown in every condition where the request cannot be full fitted due to wrong
     *             request parameters or exceptions during processing
     */
    default Message doDELETE(
            final String resource,
            final Message requestMessage,
            final MqttMessageContextBuilder messageBuilder) throws Exception {
        throw new UnsupportedOperationException();
    }

    /**
     * Used to perform application operation not necessary tied to a given resource
     *
     * @param resource the resource identifier
     * @param requestMessage the received message
     * @param messageBuilder the builder to build the response message
     *
     * @return the response to be provided back as {@link Message}
     * @throws Exception
     *             An exception is thrown in every condition where the request cannot be full fitted due to wrong
     *             request parameters or exceptions during processing
     */
    default Message doEXEC(
            final String resource,
            final Message requestMessage,
            final MqttMessageContextBuilder messageBuilder) throws Exception {
        throw new UnsupportedOperationException();
    }

}

It is consumer's sole responsibility which type of functionality to provide. If an application is only required to retrieve resources, it should only implement doGET(..) and if an application decides to create resources and execute commands on the resources, it should implement doPUT(..) and doEXEC(..) and so on. I believe you got the idea 😉

The in.bytehue.messaging.mqtt5.remote PID is used to provide the necessary configurations for remote resource management.

Amit Kumar Mondal (admin@amitinside.com)

Want to contribute? Great! Check out Contribution Guide

This project is licensed under Apache License Version 2.0