Unexpected behaviour of PublishInboundInterceptor and PublishService

[kastn]

Expected behavior

A PUBLISH that is sent via the Publish Service is processed identical to a PUBLISH sent by a client. See: https://www.hivemq.com/docs/4.1/extensions/services.html#publish-service

Actual behavior

A PUBLISH that is sent via the Publish Services isn't triggering the registered PublishInboundInterceptor.

To Reproduce

Steps

Publish a message via PublishService and the PublishInboundInterceptor isn't getting called.
Publish a message via MQTT Client and the PublishInboundInterceptor is getting called.

Reproducer code

        //Publish message "test" to topic "test" every 5 seconds
        final PublishService publishService = Services.publishService();
        final ManagedExtensionExecutorService executorService = Services.extensionExecutorService();

        final Publish publish = Builders.publish()
                .topic("test")
                .qos(Qos.AT_LEAST_ONCE)
                .payload(StandardCharsets.UTF_8.encode("test"))
                .build();

        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                publishService.publish(publish);
                LOG.info("PublishService published message");
            }
        }, 5000, 5000, TimeUnit.MILLISECONDS);


        //publish inbound interceptor logging received publishes
        final PublishInboundInterceptor publishInboundInterceptor = new PublishInboundInterceptor() {
            @Override
            public void onInboundPublish(@NotNull PublishInboundInput publishInboundInput, @NotNull PublishInboundOutput publishInboundOutput) {
                if (!publishInboundInput.getPublishPacket().getPayload().isPresent()) {
                    LOG.info("No payload present");
                    return;
                }
                ByteBuffer payloadBuffer = publishInboundInput.getPublishPacket().getPayload().get();
                LOG.info("Incoming publish with message: " + StandardCharsets.UTF_8.decode(payloadBuffer));
            }
        };

        final ClientInitializer clientInitializer = new ClientInitializer() {
            @Override
            public void initialize(@NotNull InitializerInput initializerInput, @NotNull ClientContext clientContext) {
                clientContext.addPublishInboundInterceptor(publishInboundInterceptor);
            }
        };

        Services.initializerRegistry().setClientInitializer(clientInitializer);

Details

• Affected HiveMQ CE version(s): all
• Used JVM version: 12 Open JDK

[fraschbi]

Hi @kastn,

first of all:
Thank you for participating and trying to contribute to the HiveMQ Community.

As to your request per se:
Any PUBLISH sent via the PublishService is per definition an outbound PUBLISH, as it is sent "from the broker". Therefore it is expected behaviour that the PublishInboundInterceptor is not triggered. There is no inbound PUBLISH to intercept.

On a side note:
I opened Issues for the HiveMQ Extension SDK repository, so potential issues like this can be reported their in the future.

Kind regards,
Florian