### Question:
PahoRxMqttCallback implements RxMqttCallback, MqttCallbackExtended { public static PahoRxMqttCallback create( Consumer<Throwable> onConnectionLost, BiConsumer<Boolean, String> onConnectComplete) { return create( onConnectionLost, onConnectComplete, token -> { }); }  @Override void messageArrived(String topic, MqttMessage message); @Override void deliveryComplete(IMqttDeliveryToken token); static PahoRxMqttCallback create(
      Consumer<Throwable> onConnectionLost,
      BiConsumer<Boolean, String> onConnectComplete); static PahoRxMqttCallback create(
      Consumer<Throwable> onConnectionLost,
      BiConsumer<Boolean, String> onConnectComplete,
      Consumer<RxMqttToken> onDeliveryComplete);  }### Answer:
@Test public void whenConnectionLostOccurs() { PahoRxMqttCallback rxMqttCallback = spy(PahoRxMqttCallback.create(cause -> {}, (recon, uri) -> {}, t -> {})); PahoRxMqttException exception = new PahoRxMqttException( new MqttException(MqttException.REASON_CODE_CONNECTION_LOST)); ArgumentCaptor<Throwable> onConnectionLostCauseArgumentCaptor = ArgumentCaptor.forClass(Throwable.class); rxMqttCallback.connectionLost(exception); verify(rxMqttCallback).connectionLost(onConnectionLostCauseArgumentCaptor.capture()); assertThat(onConnectionLostCauseArgumentCaptor.getValue()).isNotNull(); assertThat(onConnectionLostCauseArgumentCaptor.getValue()).isInstanceOf(PahoRxMqttException.class); assertThat(onConnectionLostCauseArgumentCaptor.getValue()).hasCauseInstanceOf(MqttException.class); assertThat(onConnectionLostCauseArgumentCaptor.getValue()).isEqualTo(exception); }
@Test public void whenConnectCompleteOccurs() { PahoRxMqttCallback rxMqttCallback = spy(PahoRxMqttCallback.create(cause -> {}, (r, u) -> {}, t -> {})); boolean reconnect = true; String brokerUri = "tcp: ArgumentCaptor<Boolean> onConnectCompleteReconnectArgumentCaptor = ArgumentCaptor.forClass(Boolean.class); ArgumentCaptor<String> onConnectCompleteServerUriArgumentCaptor = ArgumentCaptor.forClass(String.class); rxMqttCallback.connectComplete(reconnect, brokerUri); verify(rxMqttCallback).connectComplete( onConnectCompleteReconnectArgumentCaptor.capture(), onConnectCompleteServerUriArgumentCaptor.capture()); assertThat(onConnectCompleteReconnectArgumentCaptor.getValue()).isNotNull(); assertThat(onConnectCompleteReconnectArgumentCaptor.getValue()).isEqualTo(reconnect); assertThat(onConnectCompleteServerUriArgumentCaptor.getValue()).isNotNull(); assertThat(onConnectCompleteServerUriArgumentCaptor.getValue()).isEqualTo(brokerUri); }