Java 类org.eclipse.paho.client.mqttv3.IMqttMessageListener 实例源码

@Test
public void whenAMessageArrivesThenTheObserverIsNotified() throws Exception {
    @SuppressWarnings("unchecked")
    final FlowableEmitter<MqttMessage> observer = Mockito.mock(FlowableEmitter.class);
    final IMqttMessageListener listener = new SubscriberMqttMessageListener(observer);
    final String expectedTopic = "expected";
    final byte[] expectedPayload = new byte[]{ 'a', 'b', 'c' };
    final org.eclipse.paho.client.mqttv3.MqttMessage expectedMessage = new org.eclipse.paho.client.mqttv3.MqttMessage(expectedPayload);
    expectedMessage.setQos(2);
    expectedMessage.setId(1);
    expectedMessage.setRetained(true);
    final ArgumentCaptor<MqttMessage> actualMessage = ArgumentCaptor.forClass(MqttMessage.class);
    listener.messageArrived(expectedTopic, expectedMessage);
    Mockito.verify(observer).onNext(actualMessage.capture());
    Assert.assertArrayEquals(expectedPayload, actualMessage.getValue().getPayload());
    Assert.assertEquals(2, actualMessage.getValue().getQos());
    Assert.assertEquals(1, actualMessage.getValue().getId());
    Assert.assertTrue(actualMessage.getValue().isRetained());
}

@Test
public void whenCreateIsCalledThenAnObservableIsReturned() throws Exception {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final SubscribeFactory factory = new SubscribeFactory(client);
    final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor.forClass(IMqttActionListener.class);
    final ArgumentCaptor<IMqttMessageListener[]> messageListener = ArgumentCaptor.forClass(IMqttMessageListener[].class);
    final String[] topics = new String[]{ "topic1", "topic2" };
    final int[] qos = new int[]{ 1, 2 };
    final Flowable<MqttMessage> obs = factory.create(topics, qos, BackpressureStrategy.ERROR);
    Assert.assertNotNull(obs);
    obs.subscribe();
    Mockito.verify(client).subscribe(Mockito.same(topics),
            Mockito.same(qos),
            Mockito.isNull(),
            actionListener.capture(),
            messageListener.capture());
    Assert.assertTrue(actionListener.getValue() instanceof SubscribeFactory.SubscribeActionListener);
    Assert.assertTrue(messageListener.getValue() instanceof SubscriberMqttMessageListener[]);
    Assert.assertEquals(2, messageListener.getValue().length);
}

@Test
public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled() throws Throwable {
    expectedException.expectCause(isA(MqttException.class));
    final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor.forClass(IMqttActionListener.class);
    final ArgumentCaptor<IMqttMessageListener[]> messageListener = ArgumentCaptor.forClass(IMqttMessageListener[].class);
    final String[] topics = new String[]{ "topic1", "topic2" };
    final int[] qos = new int[]{ 1, 2 };
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.subscribe(Mockito.same(topics),
            Mockito.same(qos),
            Mockito.isNull(),
            actionListener.capture(),
            messageListener.capture()))
            .thenThrow(new MqttException(MqttException.REASON_CODE_CLIENT_CONNECTED));
    final SubscribeFactory factory = new SubscribeFactory(client);
    final Flowable<MqttMessage> obs = factory.create(topics, qos, BackpressureStrategy.ERROR);
    obs.blockingFirst();
}

public void subscribe(String[] topicFilters, int[] qos, String invocationContext, String activityToken, IMqttMessageListener[] messageListeners) {
    service.traceDebug(TAG, "subscribe({" + Arrays.toString(topicFilters) + "}," + Arrays.toString(qos) + ",{"
            + invocationContext + "}, {" + activityToken + "}");
    final Bundle resultBundle = new Bundle();
    resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION, MqttServiceConstants.SUBSCRIBE_ACTION);
    resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN, activityToken);
    resultBundle.putString(MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, invocationContext);
    if((myClient != null) && (myClient.isConnected())){
        IMqttActionListener listener = new MqttConnectionListener(resultBundle);
        try {

            myClient.subscribe(topicFilters, qos,messageListeners);
        } catch (Exception e){
            handleException(resultBundle, e);
        }
    } else {
        resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE, NOT_CONNECTED);
        service.traceError("subscribe", NOT_CONNECTED);
        service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
    }
}

private void itCanPubAndSubToBroker(final String brokerUrl) throws Throwable {

        // Create async MQTT clients
        final MqttAsyncClient pubClient = new MqttAsyncClient(brokerUrl, CLIENT_ID + "-pub");
        AsyncPahoUtils.connect(pubClient);
        final MqttAsyncClient subClient = new MqttAsyncClient(brokerUrl, CLIENT_ID + "-sub");
        AsyncPahoUtils.connect(subClient);

        final CountDownLatch latch = new CountDownLatch(1);
        final AtomicReference<MqttMessage> msg = new AtomicReference<MqttMessage>();

        // Subscribe
        final IMqttMessageListener messageListener = new IMqttMessageListener() {

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                msg.set(message);
                latch.countDown();
            }
        };
        AsyncPahoUtils.subscribe(subClient, TOPIC, messageListener);

        // Publish the sensor data
        final byte[] expectedPayload = new byte[] { 'a', 'b', 'c' };
        AsyncPahoUtils.publish(pubClient, TOPIC, expectedPayload);

        // Await message publish and receipt
        latch.await();

        // Get the message received by the callback
        final MqttMessage receivedMessage = msg.get();
        Assert.assertNotNull(receivedMessage);
        Assert.assertNotNull(receivedMessage.getPayload());
        Assert.assertArrayEquals(expectedPayload, receivedMessage.getPayload());

        // Close the clients
        AsyncPahoUtils.disconnect(pubClient);
        AsyncPahoUtils.disconnect(subClient);

    }

/**
 * Auto close client
 * mqttConnectOptions.userName = it }$
 * mqttConnectOptions.password = it.toCharArray() }
 * @param client
 * @param topic
 * @return
 */
@NonNull
@CheckReturnValue
public static Observable<MqttMessage> remessage(@NonNull final MqttAndroidClient client,
                                                @NonNull final String topic) {
    final Observable<MqttMessage> msgObs =
            Observable.create(new ObservableOnSubscribe<MqttMessage>() {
        public void subscribe(
                @NonNull final ObservableEmitter<MqttMessage> emitter) throws Exception {
            client.subscribe(topic, 0, new IMqttMessageListener() {
                @Override
                public void messageArrived(
                        String topic2, @NonNull final MqttMessage message) throws Exception {
                    if (!emitter.isDisposed()) {
                        emitter.onNext(message);
                    }
                }
            });
        }
    });

    if (client.isConnected()) {
        return msgObs;
    } else {
        return reconnect(client).flatMapObservable(
                new Function<IMqttToken, ObservableSource<MqttMessage>>() {
            @Override
            public ObservableSource<MqttMessage> apply(IMqttToken token) throws Exception {
                return msgObs;
            }
        });
    }
}

protected synchronized void doSubscribe(final String topic, final int qos, final IMqttMessageListener listener) throws MqttException {
    if (client == null) {
        connect(); // Lazy connection.
    }
    if (client != null && client.isConnected()) {
        client.subscribe(topic, qos, listener);
    }
}

private void subscribe()
throws IOException {
    if (subscriber == null) {
        LOGGER.warn("MQTT subscriber instance is null");
        return;
    }

    final String subscribeTopic = configuration.getSubscriber().getTopic();
    final String publishTopic = configuration.getPublisher().getTopic();
    final Mapping mapping = configuration.getMapping();

    subscriber.subscribe(subscribeTopic, new IMqttMessageListener() {

        @Override
        public void messageArrived(String topic, MqttMessage message)
        throws Exception {
            LOGGER.info(
                String.format("MQTT message arrived for topic '%1$s': %2$s", topic, message));

            if (publisher == null) {
                LOGGER.warn("MQTT publisher instance is null");
                return;
            }

            try {
                MessageEnvelope messageEnvelope = MessageEnvelope.fromMqttMessage(message,
                    mapping);
                publisher.publish(publishTopic,
                    gson.toJson(messageEnvelope, MessageEnvelope.class));
            }
            catch (IllegalStateException e) {
                LOGGER.warn(
                    String.format("MQTT message was not published: %1$s", e.getMessage()));
            }
        }

    });

}

public void subscribe(String topic, IMqttMessageListener listener)
throws IOException {

    LOGGER.info(String.format("Subscribing for topic '%1$s' ...", topic));

    try {
        client.subscribe(topic, listener);

        LOGGER.info(String.format("Subscribed for topic '%1$s'", topic));
    }
    catch (MqttException e) {
        throw new IOException("Unable to subscribe for the MQTT topic", e);
    }
}

/**
 * <p>Subscribe the given listener to the given topic</p>
 *
 * <p> This method is used by the library to subscribe to each of the topic
 * where IBM Watson IoT Platform will send the DM requests</p>
 *
 * @param topic topic to be subscribed
 * @param qos Quality of Service for the subscription
 * @param listener The IMqttMessageListener for the given topic

 * @throws MqttException When subscription fails
 */
public void subscribe(String topic, int qos, IMqttMessageListener listener) throws MqttException {
    final String METHOD = "subscribe";
    LoggerUtility.fine(CLASS_NAME, METHOD, "Topic(" + topic + ")");
    if (isConnected()) {
        if (mqttAsyncClient != null) {
            mqttAsyncClient.subscribe(topic, qos, listener).waitForCompletion();
        } else if(mqttClient != null) {
            mqttClient.subscribe(topic, qos, listener);
        }
    } else {
        LoggerUtility.warn(CLASS_NAME, METHOD, "Will not subscribe to topic(" + topic +
                ") because MQTT client is not connected.");
    }
}

/**
 * <p>Subscribe the given listeners to the given topics</p>
 *
 * <p> This method is used by the library to subscribe to each of the topic
 * where IBM Watson IoT Platform will send the DM requests</p>
 *
 * @param topics List of topics to be subscribed
 * @param qos Quality of Service for the subscription
 * @param listeners The list of IMqttMessageListeners for the given topics
 * @throws MqttException When subscription fails
 */
public void subscribe(String[] topics, int[] qos, IMqttMessageListener[] listeners) throws MqttException {
    final String METHOD = "subscribe#2";
    LoggerUtility.fine(CLASS_NAME, METHOD, "Topics(" + topics + ")");
    if (isConnected()) {
        if (mqttAsyncClient != null) {
            mqttAsyncClient.subscribe(topics, qos, listeners).waitForCompletion();
        } else if(mqttClient != null) {
            mqttClient.subscribe(topics, qos, listeners);
        }
    } else {
        LoggerUtility.warn(CLASS_NAME, METHOD, "Will not subscribe to topics(" + topics +
                ") because MQTT client is not connected.");
    }
}

/**
 * <p>Subscribe the given listener to the given topic</p>
 * 
 * <p> This method is used by the library to subscribe to each of the topic 
 * where IBM Watson IoT Platform will send the DM requests</p>
 *  
 * @param topic topic to be subscribed
 * @param qos Quality of Service for the subscription
 * @param listener The IMqttMessageListener for the given topic
 * @throws MqttException
 */
private void subscribe(String topic, int qos, IMqttMessageListener listener) throws MqttException {
    final String METHOD = "subscribe";
    LoggerUtility.fine(CLASS_NAME, METHOD, "Topic(" + topic + ")");
    if (isConnected()) {
        if (mqttAsyncClient != null) {
            mqttAsyncClient.subscribe(topic, qos, listener);
        } else if(mqttClient != null) {
            mqttClient.subscribe(topic, qos, listener);
        }
    } else {
        LoggerUtility.warn(CLASS_NAME, METHOD, "Will not subscribe to topic(" + topic +
                ") because MQTT client is not connected.");
    }
}

/**
 * <p>Subscribe the given listeners to the given topics</p>
 * 
 * <p> This method is used by the library to subscribe to each of the topic 
 * where IBM Watson IoT Platform will send the DM requests</p>
 *  
 * @param topics List of topics to be subscribed
 * @param qos Quality of Service for the subscription
 * @param listeners The list of IMqttMessageListeners for the given topics
 * @throws MqttException
 */
private void subscribe(String[] topics, int[] qos, IMqttMessageListener[] listeners) throws MqttException {
    final String METHOD = "subscribe#2";
    LoggerUtility.fine(CLASS_NAME, METHOD, "Topics(" + topics + ")");
    if (isConnected()) {
        if (mqttAsyncClient != null) {
            mqttAsyncClient.subscribe(topics, qos, listeners);
        } else if(mqttClient != null) {
            mqttClient.subscribe(topics, qos, listeners);
        }
    } else {
        LoggerUtility.warn(CLASS_NAME, METHOD, "Will not subscribe to topics(" + topics +
                ") because MQTT client is not connected.");
    }
}

@Override
public void subscribe(String topic, int qos,
        IMqttMessageListener iMqttMessageListener) throws MqttException {
    dmClient.subscribe(topic, qos, iMqttMessageListener);
}

@Override
public void subscribe(String[] topics, int[] qos,
        IMqttMessageListener[] listener) throws MqttException {
    dmClient.subscribe(topics, qos, listener);
}

@Override
public void subscribe(String topic, int qos,
        IMqttMessageListener iMqttMessageListener) throws MqttException {
    gwClient.subscribe(topic, qos, iMqttMessageListener);
}

@Override
public void subscribe(String[] topics, int[] qos,
        IMqttMessageListener[] listener) throws MqttException {
    gwClient.subscribe(topics, qos, listener);
}

public void subscribe(String topic, int qos,
IMqttMessageListener iMqttMessageListener) throws MqttException;

public void subscribe(String[] topics, int[] qos,
IMqttMessageListener[] listener) throws MqttException;

/**
 * Subscribe to multiple topics, each of which may include wildcards.
 *
 * <p>Provides an optimized way to subscribe to multiple topics compared to
 * subscribing to each one individually.</p>
 *
 * @see #subscribe(String[], int[], Object, IMqttActionListener)
 *
 * @param topicFilters one or more topics to subscribe to, which can include wildcards
 * @param qos the maximum quality of service at which to subscribe. Messages
 * published at a lower quality of service will be received at the published
 * QoS.  Messages published at a higher quality of service will be received using
 * the QoS specified on the subscribe.
 * @param userContext optional object used to pass context to the callback. Use
 * null if not required.
 * @param callback optional listener that will be notified when subscribe
 * has completed
 * @param messageListeners an array of callbacks to handle incoming messages
 * @return token used to track and wait for the subscribe to complete. The token
 * will be passed to callback methods if set.
 * @throws MqttException if there was an error registering the subscription.
 */
public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback, IMqttMessageListener[] messageListeners) throws MqttException {
    IMqttToken token = new MqttTokenAndroid(this, userContext, callback, topicFilters);
    String activityToken = storeToken(token);
    mqttService.subscribe(clientHandle, topicFilters, qos, null, activityToken, messageListeners);

    return null;
}

/**
 * Subscribe using topic filters
 *
 * @param clientHandle
 *            identifies the MqttConnection to use
 * @param topicFilters
 *            a list of possibly wildcarded topicfilters
 * @param qos
 *            requested quality of service for each topic
 * @param invocationContext
 *            arbitrary data to be passed back to the application
 * @param activityToken
 *            arbitrary identifier to be passed back to the Activity
 * @param messageListeners a callback to handle incoming messages
 */
public void subscribe(String clientHandle, String[] topicFilters, int[] qos, String invocationContext, String activityToken, IMqttMessageListener[] messageListeners){
  MqttConnection client = getConnection(clientHandle);
  client.subscribe(topicFilters, qos, invocationContext, activityToken, messageListeners);
}

/**
 * Subscribe to a topic, which may include wildcards.
 *
 * @see #subscribe(String[], int[], Object, IMqttActionListener)
 *
 * @param topicFilter the topic to subscribe to, which can include wildcards.
 * @param qos the maximum quality of service at which to subscribe. Messages
 * published at a lower quality of service will be received at the published
 * QoS.  Messages published at a higher quality of service will be received using
 * the QoS specified on the subscribe.
 * @param userContext optional object used to pass context to the callback. Use
 * null if not required.
 * @param callback optional listener that will be notified when subscribe
 * has completed
 * @param messageListener a callback to handle incoming messages
 * @return token used to track and wait for the subscribe to complete. The token
 * will be passed to callback methods if set.
 * @throws MqttException if there was an error registering the subscription.
 */
public IMqttToken subscribe(String topicFilter, int qos, Object userContext, IMqttActionListener callback, IMqttMessageListener messageListener) throws MqttException {

    return subscribe(new String[] {topicFilter}, new int[] {qos}, userContext, callback, new IMqttMessageListener[] {messageListener});
}

/**
 * Subscribe to a topic, which may include wildcards.
 *
 * @see #subscribe(String[], int[], Object, IMqttActionListener)
 *
 * @param topicFilter the topic to subscribe to, which can include wildcards.
 * @param qos the maximum quality of service at which to subscribe. Messages
 * published at a lower quality of service will be received at the published
 * QoS.  Messages published at a higher quality of service will be received using
 * the QoS specified on the subscribe.
 * @param messageListener a callback to handle incoming messages
 * @return token used to track and wait for the subscribe to complete. The token
 * will be passed to callback methods if set.
 * @throws MqttException if there was an error registering the subscription.
 */
public IMqttToken subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException {

    return subscribe(topicFilter, qos, null, null, messageListener);
}

/**
 * Subscribe to multiple topics, each of which may include wildcards.
 *
 * <p>Provides an optimized way to subscribe to multiple topics compared to
 * subscribing to each one individually.</p>
 *
 * @see #subscribe(String[], int[], Object, IMqttActionListener)
 *
 * @param topicFilters one or more topics to subscribe to, which can include wildcards
 * @param qos the maximum quality of service at which to subscribe. Messages
 * published at a lower quality of service will be received at the published
 * QoS.  Messages published at a higher quality of service will be received using
 * the QoS specified on the subscribe.
 * @param messageListeners an array of callbacks to handle incoming messages
 * @return token used to track and wait for the subscribe to complete. The token
 * will be passed to callback methods if set.
 * @throws MqttException if there was an error registering the subscription.
 */
public IMqttToken subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException {

    return subscribe(topicFilters, qos, null, null, messageListeners);
}