Is this the correct way to process results through the Spring Integration Flow?

I am currently working on a personal project - in which I need my Spring application to take queries from an EMQX (MQTT Server) and query its data for corresponding results, and then push the results to a topic with the query UUID.

This is working - after many hours understanding how the Spring Integration framework works. But I think the way in which the handler is using "block" is incorrect - and not in keeping with the manner in which the Integration Flow should operate. Whilst this works I do want to make sure it is being done properly - out of respect for the work - and to avoid future issues.

The code snippet below should be enough to understand what it is that I'm trying to achieve - and where the potential issue lies.

    @Bean
fun mqttInFlow() : Publisher<Message<String>> {
    return IntegrationFlows.from(inbound())
        .handle<String> { payload, headers ->
            val emotionalOutput: EmotionalOutput = gson.fromJson(payload, EmotionalOutput::class.java)
            emotionalPrintService.populateEmotionalOutput(emotionalOutput).map {
                MessageBuilder.withPayload(gson.toJson(it))
                    .copyHeaders(headers)
                    .setHeader(MqttHeaders.TOPIC, "query/" + it.query_uuid).build()
            }.block()
        }
        .channel(outgoingChannel())
        .toReactivePublisher()
}

EDIT - Thanks for the advice - here is what I understood to be the potential edit for the Kotlin DSL solution - this is now producing an error - complaining that an output-channel or replyChannel was not available - nothing outside of the this function has been changed.

    @Bean
fun newMqttInFlow() =
    integrationFlow (inbound()) {
       wireTap {
            handle<String> { payload, headers ->
                gson.fromJson<EmotionalOutput>(payload, EmotionalOutput::class.java).let { emotionalOutput ->
                    emotionalPrintService.populateEmotionalOutput(emotionalOutput).map { populatedEmotionalOutput ->
                        MessageBuilder.withPayload(gson.toJson(populatedEmotionalOutput))
                            .copyHeaders(headers)
                            .setHeader(MqttHeaders.TOPIC, populatedEmotionalOutput.query_uuid)
                    }
                }
            }
        }
        channel("outgoingChannel")
    }

Exception is :

exception is org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available

Although I have many years experience with Java - this approach is new - so thank you very much for your assistance. It's appreciated. If the whole class would be useful - I can post that.

EDIT

Here is the Configuration file - which might give a better insight into what might be causing this secondary error -

021-03-28 21:59:48.008 ERROR 84492 --- [T Call: divnrin] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageHandlingException: error occurred in message handler [bean 'mqttOutbound'; defined in: 'class path resource [io/divnr/appserver/configuration/MQTTConfiguration.class]'; from source: 'org.springframework.core.type.classreading.SimpleMethodMetadata@4a9419d7']; nested exception is java.lang.IllegalArgumentException: This default converter can only handle 'byte[]' or 'String' payloads; consider adding a transformer to your flow definition, or provide a BytesMessageMapper, or subclass this converter for reactor.core.publisher.MonoMapFuseable payloads, failedMessage=GenericMessage [payload=MonoMapFuseable, headers={mqtt_receivedRetained=false, mqtt_id=0, mqtt_duplicate=false, id=c5a75283-c0fe-ebac-4168-dabddd989da9, mqtt_receivedTopic=source/d9e50e8f-67e0-4505-7ca2-4d05b1242207, mqtt_receivedQos=0, timestamp=1616961588004}] at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:192) at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:65) at

The full class is provided here.

@Configuration
@EnableIntegration
@IntegrationComponentScan
class MQTTConfiguration(val emotionalPrintService: EmotionalPrintService,
                    val gson: Gson,
                    val applicationConfiguration: ApplicationConfiguration) {

@Bean
fun mqttServiceFactory() : MqttPahoClientFactory {
    return DefaultMqttPahoClientFactory().apply {
        connectionOptions = MqttConnectOptions().apply {
            serverURIs = arrayOf<String>(applicationConfiguration.mqttServerAddress)
        }
    }
}

@Bean
fun newMqttInFlow() =
    integrationFlow (inbound()) {
        handle<String> { payload, headers ->
            gson.fromJson<EmotionalOutput>(payload, EmotionalOutput::class.java).let { emotionalOutput ->
                emotionalPrintService.populateEmotionalOutput(emotionalOutput).map { populatedEmotionalOutput ->
                    MessageBuilder.withPayload(gson.toJson(populatedEmotionalOutput))
                        .copyHeaders(headers)
                        .setHeader(MqttHeaders.TOPIC, populatedEmotionalOutput.query_uuid).build()
                }
            }
        }
        channel(outgoingChannel())
    }

@Bean
@ServiceActivator(requiresReply = "false", inputChannel = "outgoingChannel")
fun mqttOutbound(): MessageHandler {
    val messageHandler = MqttPahoMessageHandler("divnrout", mqttServiceFactory())
    messageHandler.setAsync(true)
    return messageHandler
}

@Bean
fun outgoingChannel() : FluxMessageChannel {
    return FluxMessageChannel()
}

@Bean
fun inbound(): MessageProducerSupport {
    return MqttPahoMessageDrivenChannelAdapter("divnrin", mqttServiceFactory(),
        "source/" + applicationConfiguration.sourceUuid).apply {
        setConverter(DefaultPahoMessageConverter())
        setQos(1)
    }
}
}