@SpringBootTest
@SpringJUnitConfig
internal open class ProvisioningTest @Autowired constructor(
    private val mqtt: IMqttAsyncClient,
    private val mapper: ObjectMapper
) {

    @Test
    fun provision() {
        val entity = Foley(
            rfid = UUID.randomUUID().toString(),
        )

        val called = AtomicBoolean(false)
        mqtt.subscribe("foley/created", 1) { _, _ -> called.set(true) }

        mqtt.publish("foley/new", MqttMessage(mapper.writeValueAsBytes(entity)))

        Awaitility.await().atMost(10, TimeUnit.SECONDS).untilTrue(called)
    }
}

@Service
open class Provisioning(private val repo: FoleyRepo) {
    private val log: Logger = LogManager.getLogger(this::class.java)

    @SendTo("foley.created")
    @RabbitListener(queuesToDeclare = [Queue("foley.new")] )
    open fun listen(entity: Foley): Foley {
        log.trace("saving: {}", entity)
        val save = repo.save(entity)
        log.debug("saved: {}", save)
        return save
    }

}

@Bean
open fun client(
    @Value("tcp://${mqtt.client.host:localhost}:${mqtt.client.port:1883}") uri: String,
    @Value("${mqtt.client.user:#{null}}") user: String?,
    @Value("${mqtt.client.pass:#{null}}") pass: CharArray?
): IMqttAsyncClient {

    val connOpt = MqttConnectOptions()
    user?.let { connOpt.userName = it }
    pass?.let { connOpt.password = it }
    connOpt.isCleanSession = false
    connOpt.isAutomaticReconnect = true
    val client = MqttAsyncClient(uri, MqttAsyncClient.generateClientId(), MemoryPersistence())
    client.connect(connOpt)
    return client
}

@Bean
open fun messageConverter(om: ObjectMapper): MessageConverter {
    return Jackson2JsonMessageConverter(om)
}

@Bean
open fun builder(): Jackson2ObjectMapperBuilderCustomizer {
    return Jackson2ObjectMapperBuilderCustomizer {
        it.modules(JavaTimeModule(), KotlinModule())
    }
}

@Bean
open fun foleyNew(): Queue {
    return Queue("foley.new")
}

@Bean
open fun foleyCreated(): Queue {
    return Queue("foley.created")
}

@Bean
open fun exchange(): TopicExchange {
    return TopicExchange("amq.topic")
}

@Bean
open fun binding(): Binding {
    return BindingBuilder.bind(foleyNew()).to(exchange()).with("foley.new")
}

Caused by: org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public phg.entity.monitor.Foley phg.event.monitor.Provisioning.save(phg.entity.monitor.Foley)]
Bean [phg.event.monitor.Provisioning@320770d7]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:193) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:127) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1547) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1473) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1461) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1456) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1405) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    ... 6 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [phg.entity.monitor.Foley] for GenericMessage [payload=byte[262], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=foley.new, amqp_receivedExchange=amq.topic, x-mqtt-publish-qos=1, x-mqtt-dup=false, amqp_deliveryTag=1, amqp_consumerQueue=foley.new, amqp_redelivered=false, id=9980af26-8b7d-87b9-dde3-64f0fbfe1fa2, amqp_consumerTag=amq.ctag-vOvIcybV6HMf8fD3kwiC3w, timestamp=1551852622698}]
    at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:144) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:117) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:49) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:190) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]