SubscriptionService implements MqttClientEventHandler { @Override public void onReceive(final MqttClientAdapter mqttClientAdapter, final byte[] payloadAsBytes) { final String payload = new String(payloadAsBytes); final MessageMetadata messageMetadata = mqttClientAdapter.getMessageMetadata(); LOG.info(String.format("Client for device:%s received payload:%s", messageMetadata.getDeviceIdentification(), payload)); final ResponseMessage responseMessage = new ProtocolResponseMessage.Builder() .deviceMessageMetadata(new DeviceMessageMetadata(messageMetadata)) .domain(messageMetadata.getDomain()) .domainVersion(messageMetadata.getDomainVersion()) .dataObject(payload) .result(ResponseMessageResultType.OK) .build(); this.outboundOsgpCoreResponseMessageSender.send(responseMessage); } SubscriptionService(final MqttDeviceRepository mqttDeviceRepository,
            final MqttClientAdapterFactory mqttClientAdapterFactory,
            final OutboundOsgpCoreResponseMessageSender outboundOsgpCoreResponseMessageSender,
            @Value("#{new Integer('${mqtt.broker.defaultPort}')}") final int defaultPort,
            @Value("${mqtt.broker.defaultTopics}") final String defaultTopics,
            @Value("${mqtt.broker.defaultQos}") final String defaultQos); void subscribe(final MessageMetadata messageMetadata); @Override void onConnect(final MqttClientAdapter mqttClientAdapter, final Mqtt3ConnAck ack,
            final Throwable throwable); @Override void onSubscribe(final MqttClientAdapter mqttClientAdapter, final Mqtt3SubAck subAck,
            final Throwable throwable); @Override void onReceive(final MqttClientAdapter mqttClientAdapter, final byte[] payloadAsBytes);  }
@Test void onReceive() { when(this.messageMetadata.getMessageType()).thenReturn("test-message-type"); when(this.messageMetadata.getCorrelationUid()).thenReturn("test-correlation-uuid"); when(this.messageMetadata.getOrganisationIdentification()).thenReturn("test-organisation-id"); when(this.messageMetadata.getDeviceIdentification()).thenReturn("test-device-id"); when(this.messageMetadata.getMessagePriority()).thenReturn(2345); when(this.messageMetadata.isBypassRetry()).thenReturn(true); when(this.messageMetadata.getScheduleTime()).thenReturn(null); when(this.messageMetadata.getDomain()).thenReturn("test-device-id"); when(this.messageMetadata.getDomainVersion()).thenReturn("test-device-id"); final String payload = "12345"; final byte[] bytes = payload.getBytes(); this.instance.onReceive(this.mqttClientAdapter, bytes); verify(this.outboundOsgpCoreResponseMessageSender).send(this.protocolResponseMessageCaptor.capture()); final ProtocolResponseMessage protocolResponseMessage = this.protocolResponseMessageCaptor.getValue(); assertEquals(this.messageMetadata.getMessageType(), protocolResponseMessage.getMessageType()); assertEquals(this.messageMetadata.getCorrelationUid(), protocolResponseMessage.getCorrelationUid()); assertEquals(this.messageMetadata.getOrganisationIdentification(), protocolResponseMessage.getOrganisationIdentification()); assertEquals(this.messageMetadata.getDeviceIdentification(), protocolResponseMessage.getDeviceIdentification()); assertEquals(this.messageMetadata.getMessagePriority(), protocolResponseMessage.getMessagePriority()); assertEquals(this.messageMetadata.isBypassRetry(), protocolResponseMessage.bypassRetry()); assertFalse(protocolResponseMessage.isScheduled()); assertEquals(this.messageMetadata.getDomain(), protocolResponseMessage.getDomain()); assertEquals(this.messageMetadata.getDomainVersion(), protocolResponseMessage.getDomainVersion()); assertEquals(payload, protocolResponseMessage.getDataObject()); assertEquals(ResponseMessageResultType.OK, protocolResponseMessage.getResult()); }