HiveMQ Enterprise Extension for Google Cloud Pub/Sub Customization

public class BusinessLogicTransformer implements MqttToPubSubTransformer {

    @Override
    public void transformMqttToPubSub(
            final @NotNull MqttToPubSubInput mqttToPubSubInput,
            final @NotNull MqttToPubSubOutput mqttToPubSubOutput) {

        // Get MQTT payload
        final ByteBuffer mqttPayload = mqttToPubSubInput.getPublishPacket().getPayload().orElse(ByteBuffer.allocate(0));

        // Manipulate the MQTT payload according to your business logic
        // For example, deserialize or add fields
        final byte[] data = ownBusinessLogic(mqttPayload);

        final OutboundPubSubMessage message = output.newOutboundPubSubMessageBuilder()
                .topicName(input.getPublishPacket().getTopic())
                .data(input.getPublishPacket().getPayload().orElse(ByteBuffer.allocate(0)))
                .build();

        // Set message as output
        mqttToPubSubOutput.setOutboundPubSubMessages(List.of(message));
    }
}

public class MulticastTransformer implements MqttToPubSubTransformer {

    private static final String PUBSUB_TOPIC_LOCATION_SERVICE = "location-service";
    private static final String PUBSUB_TOPIC_SPEED_SERVICE = "speed-control-service";
    private static final String PUBSUB_TOPIC_FUEL_SERVICE = "fuel-control-service";

    @Override
    public void transformMqttToPubSub(
            final @NotNull MqttToPubSubInput input,
            final @NotNull MqttToPubSubOutput output) {

        // Get MQTT payload
        final ByteBuffer mqttPayload = input.getPublishPacket().getPayload().orElse(ByteBuffer.allocate(0));

        // Extract information from MQTT payload
        byte[] location = extractLocation(mqttPayload);
        // Create record for the location information
        final OutboundPubSubMessage locationMessage = output.newOutboundPubSubMessageBuilder()
                .topicName(PUBSUB_TOPIC_LOCATION_SERVICE)
                .data(location)
                .build();
        // Create record for the speed information
        byte[] speed = extractSpeed(mqttPayload);
        final OutboundPubSubMessage speedMessage = output.newOutboundPubSubMessageBuilder()
                .topicName(PUBSUB_TOPIC_SPEED_SERVICE)
                .data(speed)
                .build();
        // Create record for the fuel information
        byte[] fuel = extractFuel(mqttPayload);
        final OutboundPubSubMessage fuelMessage = output.newOutboundPubSubMessageBuilder()
                .topicName(PUBSUB_TOPIC_FUEL_SERVICE)
                .data(fuel)
                .build();

        // Set the 3 PubSub messages as output
        output.setOutboundPubSubMessages(List.of(locationMessage, speedMessage, fuelMessage));
    }

    private byte[] extractFuel(final @NotNull ByteBuffer mqttPayload) {
        // Pseudo-implementation, add custom code
        return null;
    }

    private byte[] extractSpeed(final @NotNull ByteBuffer mqttPayload) {
        // Pseudo-implementation, add custom code
        return null;
    }

    private byte[] extractLocation(final @NotNull ByteBuffer mqttPayload) {
        // Pseudo-implementation, add custom code
        return null;
    }
}

public class DroppingTransformer implements MqttToPubSubTransformer {

    @Override
    public void transformMqttToPubSub(
            final @NotNull MqttToPubSubInput input,
            final @NotNull MqttToPubSubOutput output) {
        if (input.getPublishPacket().getPayload().isEmpty()) {
            output.setOutboundPubSubMessages(List.of());
            return;
        }

        final OutboundPubSubMessage message = output.newOutboundPubSubMessageBuilder()
                .topicName(input.getPublishPacket().getTopic())
                .data(input.getPublishPacket().getPayload())
                .build();

        output.setOutboundPubSubMessages(List.of(message));
    }
}

 <mqtt-to-pubsub-transformers>
        <mqtt-to-pubsub-transformer>
            <enabled>true</enabled>
            <id>my-mqtt-to-pubsub-transformer-01</id>
            <pubsub-connection>your-custom-connection-id</pubsub-connection>
            <transformer>fully.qualified.classname.to.CustomSettingsTransformer</transformer>
            <mqtt-topic-filters>
                <mqtt-topic-filter>mqtt/topic/a</mqtt-topic-filter>
            </mqtt-topic-filters>
            <custom-settings>
                <custom-setting>
                    <name>destination-topic</name>
                    <value>destination/pubsub/topic/a</value>
                </custom-setting>
            </custom-settings>
        </mqtt-to-pubsub-transformer>
 </mqtt-to-pubsub-transformers>

public class CustomSettingsTransformer implements MqttToPubSubTransformer {

    private @NotNull Optional<String> destinationTopic = Optional.empty();

    @Override
    public void init(@NotNull MqttToPubSubInitInput transformerInitInput) {
        final CustomSettings customSettings = transformerInitInput.getCustomSettings();
        destinationTopic = customSettings.getFirst("destination-topic");
    }

    @Override
    public void transformMqttToPubSub(
            final @NotNull MqttToPubSubInput input,
            final @NotNull MqttToPubSubOutput output) {
        final OutboundPubSubMessage message = output.newOutboundPubSubMessageBuilder()
                .topicName(destinationTopic.orElse("default/destination/topic"))
                .data(input.getPublishPacket().getPayload().orElse(ByteBuffer.allocate(0)))
                .build();
        output.setOutboundPubSubMessages(List.of(message));
    }
}

public class CustomTransformer implements PubSubToMqttTransformer {

    @Override
    public void transformPubSubToMqtt(
            final @NotNull PubSubToMqttInput input,
            final @NotNull PubSubToMqttOutput output) {

        // Get the PubSub message
        final InboundPubSubMessage pubSubMessage = input.getInboundPubSubMessage();

        final byte[] mqttPayload = ownBusinessLogic(pubSubMessage.getDataAsByteArray().orElse(new byte[0]));

        // Build MQTT publish
        final Publish publish = output.newPublishBuilder()
                .topic(pubSubMessage.getSubscriptionName())
                .qos(Qos.AT_LEAST_ONCE)
                .payload(ByteBuffer.wrap(mqttPayload))
                .build();

        // Set MQTT publish as output
        output.setPublishes(List.of(publish));
    }

    private @NotNull byte[] ownBusinessLogic(final byte @NotNull [] mqttPayload) {
        // Insert your own business logic
        return mqttPayload;
    }
}

public class SkippingTransformer implements PubSubToMqttTransformer {

    @Override
    public void transformPubSubToMqtt(
            final @NotNull PubSubToMqttInput input,
            final @NotNull PubSubToMqttOutput output) {
        // Get the PubSub message
        final InboundPubSubMessage message = input.getInboundPubSubMessage();

        // Skip the PubSub message if the data is missing
        if (message.getData().isEmpty()) {
            output.setPublishes(List.of());
            return;
        }

        // Build MQTT publish
        final Publish publish = output.newPublishBuilder()
                .topic(message.getSubscriptionName())
                .payload(message.getData().get())
                .build();
        output.setPublishes(List.of(publish));
    }
}

<pubsub-to-mqtt-transformers>
    <pubsub-to-mqtt-transformer>
        <id>my-pubsub-to-mqtt-transformer-01</id>
        <enabled>true</enabled>
        <pubsub-connection>your-custom-connection-id</pubsub-connection>
        <pubsub-subscriptions>
            <pubsub-subscription>
                <name>your-subscription</name>
            </pubsub-subscription>
        </pubsub-subscriptions>
        <transformer>fully.qualified.classname.to.CustomSettingsTransformer</transformer>
        <custom-settings>
            <custom-setting>
                <name>destination-topic</name>
                <value>destination/mqtt/topic/a</value>
            </custom-setting>
        </custom-settings>
    </pubsub-to-mqtt-transformer>
</pubsub-to-mqtt-transformers>

public class CustomSettingsTransformer implements PubSubToMqttTransformer {

    private @NotNull Optional<String> destinationTopic = Optional.empty();

    @Override
    public void init(@NotNull PubSubToMqttInitInput transformerInitInput) {
        final CustomSettings customSettings = transformerInitInput.getCustomSettings();
        destinationTopic = customSettings.getFirst("destination-topic");
    }

    @Override
    public void transformPubSubToMqtt(
            final @NotNull PubSubToMqttInput input,
            final @NotNull PubSubToMqttOutput output) {
        // Get the PubSub message
        final InboundPubSubMessage message = input.getInboundPubSubMessage();

        // Build MQTT publish
        final Publish publish = output.newPublishBuilder()
                .topic(destinationTopic.orElse("default/destination/topic"))
                .payload(message.getData().orElse(ByteBuffer.allocate(0)))
                .build();
        output.setPublishes(List.of(publish));
    }
}