SubscriptionService implements MqttClientEventHandler { public void subscribe(final MessageMetadata messageMetadata) { final MqttDevice device = this.getOrCreateDevice(messageMetadata); final MqttClientAdapter mqttClientAdapter = this.mqttClientAdapterFactory.create(device, messageMetadata, this); mqttClientAdapter.connect(); } 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 subscribeNewDevice() { when(this.messageMetadata.getDeviceIdentification()).thenReturn("test-metadata-device-id"); when(this.messageMetadata.getIpAddress()).thenReturn("test-metadata-host"); when(this.mqttDeviceRepository.findByDeviceIdentification( this.messageMetadata.getDeviceIdentification())).thenReturn(null); when(this.mqttClientAdapterFactory.create(any(MqttDevice.class), eq(this.messageMetadata), eq(this.instance))).thenReturn(this.mqttClientAdapter); this.instance.subscribe(this.messageMetadata); verify(this.mqttDeviceRepository).save(this.deviceCaptor.capture()); final MqttDevice savedDevice = this.deviceCaptor.getValue(); assertEquals(this.messageMetadata.getDeviceIdentification(), savedDevice.getDeviceIdentification()); assertEquals(DEFAULT_QOS.name(), savedDevice.getQos()); assertEquals(this.messageMetadata.getIpAddress(), savedDevice.getHost()); assertEquals(DEFAULT_TOPICS, savedDevice.getTopics()); assertEquals(DEFAULT_PORT, savedDevice.getPort()); verify(this.mqttClientAdapter).connect(); }
@Test void subscribeExistingDevice() { when(this.messageMetadata.getDeviceIdentification()).thenReturn("test-metadata-device-id"); final MqttDevice device = mock(MqttDevice.class); when(this.mqttDeviceRepository.findByDeviceIdentification( this.messageMetadata.getDeviceIdentification())).thenReturn(device); when(this.mqttClientAdapterFactory.create(eq(device), eq(this.messageMetadata), eq(this.instance))).thenReturn( this.mqttClientAdapter); this.instance.subscribe(this.messageMetadata); verify(this.mqttClientAdapter).connect(); }