/**
 * Used for calling the subscribe handler when the remote MQTT client subscribes to topics
 *
 * @param msg message with subscribe information
 */
void handleSubscribe(MqttSubscribeMessage msg) {

  synchronized (this.so) {
    if (this.checkConnected()) {
      this.endpoint.handleSubscribe(msg);
    }
  }
}
 

@Override
public MqttSubscribeMessage getMessage() {
    return message;
}
 

@SuppressWarnings("unchecked")
private void testOnSubscribeRegistersAndClosesConnection(final MqttQoS qos) {

    // GIVEN a device connected to an adapter
    final Promise<ProtonDelivery> outcome = Promise.promise();
    outcome.complete(mock(ProtonDelivery.class));
    final DownstreamSender sender = givenAnEventSenderForOutcome(outcome);
    final MqttServer server = getMqttServer(false);
    final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server);
    final MqttEndpoint endpoint = mockEndpoint();
    when(endpoint.keepAliveTimeSeconds()).thenReturn(10); // 10 seconds

    // WHEN a device subscribes to commands
    final ProtocolAdapterCommandConsumer commandConsumer = mock(ProtocolAdapterCommandConsumer.class);
    when(commandConsumer.close(any())).thenReturn(Future.succeededFuture());
    when(commandConsumerFactory.createCommandConsumer(eq("tenant"), eq("deviceId"), any(Handler.class), any(), any()))
                    .thenReturn(Future.succeededFuture(commandConsumer));
    final List<MqttTopicSubscription> subscriptions = Collections.singletonList(
            newMockTopicSubscription(getCommandSubscriptionTopic("tenant", "deviceId"), qos));
    final MqttSubscribeMessage msg = mock(MqttSubscribeMessage.class);
    when(msg.messageId()).thenReturn(15);
    when(msg.topicSubscriptions()).thenReturn(subscriptions);

    final CommandSubscriptionsManager<MqttProtocolAdapterProperties> cmdSubscriptionsManager = new CommandSubscriptionsManager<>(vertx, config);
    endpoint.closeHandler(
            handler -> adapter.close(endpoint, new Device("tenant", "deviceId"), cmdSubscriptionsManager, OptionalInt.empty()));
    adapter.onSubscribe(endpoint, null, msg, cmdSubscriptionsManager, OptionalInt.empty());

    // THEN the adapter creates a command consumer that is checked periodically
    verify(commandConsumerFactory).createCommandConsumer(eq("tenant"), eq("deviceId"), any(Handler.class), any(), any());
    // and the adapter registers a hook on the connection to the device
    final ArgumentCaptor<Handler<Void>> closeHookCaptor = ArgumentCaptor.forClass(Handler.class);
    verify(endpoint).closeHandler(closeHookCaptor.capture());
    // which closes the command consumer when the device disconnects
    closeHookCaptor.getValue().handle(null);
    verify(commandConsumer).close(any());
    // and sends an empty notification downstream with TTD 0
    final ArgumentCaptor<Message> msgCaptor = ArgumentCaptor.forClass(Message.class);
    verify(sender, times(2)).sendAndWaitForOutcome(msgCaptor.capture(), any());
    assertThat(msgCaptor.getValue().getContentType()).isEqualTo(EventConstants.CONTENT_TYPE_EMPTY_NOTIFICATION);
    assertThat(MessageHelper.getTimeUntilDisconnect(msgCaptor.getValue())).isEqualTo(0);
}
 

/**
 * Verifies that the adapter includes a status code for each topic filter in its SUBACK packet.
 */
@SuppressWarnings("unchecked")
@Test
public void testOnSubscribeIncludesStatusCodeForEachFilter() {

    // GIVEN a device connected to an adapter
    final Promise<ProtonDelivery> outcome = Promise.promise();
    outcome.complete(mock(ProtonDelivery.class));
    final DownstreamSender sender = givenAnEventSenderForOutcome(outcome);
    final MqttServer server = getMqttServer(false);
    final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server);
    final MqttEndpoint endpoint = mockEndpoint();
    when(endpoint.isConnected()).thenReturn(true);

    // WHEN a device sends a SUBSCRIBE packet for several unsupported filters
    final List<MqttTopicSubscription> subscriptions = new ArrayList<>();
    subscriptions.add(newMockTopicSubscription("unsupported/#", MqttQoS.AT_LEAST_ONCE));
    subscriptions.add(newMockTopicSubscription("bumlux/+/+/#", MqttQoS.AT_MOST_ONCE));
    subscriptions.add(newMockTopicSubscription("bumlux/+/+/#", MqttQoS.AT_MOST_ONCE));
    // and for subscribing to commands
    final ProtocolAdapterCommandConsumer commandConsumer = mock(ProtocolAdapterCommandConsumer.class);
    when(commandConsumer.close(any())).thenReturn(Future.succeededFuture());
    when(commandConsumerFactory.createCommandConsumer(eq("tenant-1"), eq("device-A"), any(Handler.class), any(), any()))
                    .thenReturn(Future.succeededFuture(commandConsumer));
    subscriptions.add(
            newMockTopicSubscription(getCommandSubscriptionTopic("tenant-1", "device-A"), MqttQoS.AT_MOST_ONCE));
    subscriptions.add(
            newMockTopicSubscription(getCommandSubscriptionTopic("tenant-1", "device-B"), MqttQoS.EXACTLY_ONCE));
    final MqttSubscribeMessage msg = mock(MqttSubscribeMessage.class);
    when(msg.messageId()).thenReturn(15);
    when(msg.topicSubscriptions()).thenReturn(subscriptions);

    adapter.onSubscribe(endpoint, null, msg, new CommandSubscriptionsManager<>(vertx, config), OptionalInt.empty());

    // THEN the adapter sends a SUBACK packet to the device
    // which contains a failure status code for each unsupported filter
    final ArgumentCaptor<List<MqttQoS>> codeCaptor = ArgumentCaptor.forClass(List.class);
    verify(endpoint).subscribeAcknowledge(eq(15), codeCaptor.capture());
    assertThat(codeCaptor.getValue()).hasSize(5);
    assertThat(codeCaptor.getValue().get(0)).isEqualTo(MqttQoS.FAILURE);
    assertThat(codeCaptor.getValue().get(1)).isEqualTo(MqttQoS.FAILURE);
    assertThat(codeCaptor.getValue().get(2)).isEqualTo(MqttQoS.FAILURE);
    assertThat(codeCaptor.getValue().get(3)).isEqualTo(MqttQoS.AT_MOST_ONCE);
    // and sends an empty notification downstream with TTD -1
    final ArgumentCaptor<Message> msgCaptor = ArgumentCaptor.forClass(Message.class);
    verify(sender).sendAndWaitForOutcome(msgCaptor.capture(), any());
    assertThat(MessageHelper.getDeviceId(msgCaptor.getValue())).isEqualTo("device-A");
    assertThat(msgCaptor.getValue().getContentType()).isEqualTo(EventConstants.CONTENT_TYPE_EMPTY_NOTIFICATION);
    assertThat(MessageHelper.getTimeUntilDisconnect(msgCaptor.getValue())).isEqualTo(-1);

}
 

/**
 * Set a subscribe handler on the MQTT endpoint. This handler is called when a SUBSCRIBE
 * message is received by the remote MQTT client
 *
 * @param handler the handler
 * @return a reference to this, so the API can be used fluently
 */
@Fluent
MqttEndpoint subscribeHandler(Handler<MqttSubscribeMessage> handler);
 

MqttSubscribeMessage getMessage();