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

private Builder builderWithMocks(final String expectedClientId) {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.getClientId()).thenReturn(expectedClientId);
    final CloseFactory closeFactory = Mockito.mock(CloseFactory.class);
    final ConnectFactory connectFactory = Mockito.mock(ConnectFactory.class);
    final DisconnectFactory disconnectFactory = Mockito.mock(DisconnectFactory.class);
    final PublishFactory publishFactory = Mockito.mock(PublishFactory.class);
    final SubscribeFactory subscribeFactory = Mockito.mock(SubscribeFactory.class);
    final UnsubscribeFactory unsubscribeFactory = Mockito.mock(UnsubscribeFactory.class);
    return new PahoObservableMqttClient.Builder(client)
            .setCloseFactory(closeFactory)
            .setConnectFactory(connectFactory)
            .setDisconnectFactory(disconnectFactory)
            .setPublishFactory(publishFactory)
            .setSubscribeFactory(subscribeFactory)
            .setUnsubscribeFactory(unsubscribeFactory);
}

@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();
}

@Test
public void whenCreateIsCalledThenAnObservableIsReturned() throws Exception {
    // Given
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final PublishFactory factory = new PublishFactory(client);
    final String topic = "topic1";
    final MqttMessage msg = MqttMessage.create(0, new byte[] { 'a',  'b',  'c' }, 1, true);
    final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor.forClass(IMqttActionListener.class);

    // When
    final Single<PublishToken> obs = factory.create(topic, msg);

    // Then
    Assert.assertNotNull(obs);
    obs.subscribe();
    Mockito.verify(client).publish(Mockito.same(topic), 
            Mockito.same(msg.getPayload()), Mockito.anyInt(),
            Mockito.anyBoolean(), Mockito.any(),
            actionListener.capture());
    Assert.assertTrue(actionListener.getValue() instanceof PublishFactory.PublishActionListener);
}

@Test
public void whenCreateIsCalledThenAnObservableIsReturned() throws Exception {
    // Given
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final UnsubscribeFactory factory = new UnsubscribeFactory(client);
    final String[] topics = new String[]{ "topic1", "topic2" };
    final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor.forClass(IMqttActionListener.class);

    // When
    final Completable obs = factory.create(topics);

    // Then
    Assert.assertNotNull(obs);
    obs.subscribe();
    Mockito.verify(client).unsubscribe(Mockito.same(topics), Mockito.isNull(),
            actionListener.capture());
    Assert.assertTrue(actionListener.getValue() instanceof UnsubscribeFactory.UnsubscribeActionListener);
}

@Test
public void whenCreateIsCalledThenAnObservableIsReturned() throws Exception {
    // Given
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final DisconnectFactory factory = new DisconnectFactory(client);
    final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor.forClass(IMqttActionListener.class);

    // When
    final Completable obs = factory.create();

    // Then
    Assert.assertNotNull(obs);
    obs.subscribe();
    Mockito.verify(client).disconnect(Mockito.isNull(),
            actionListener.capture());
    Assert.assertTrue(actionListener.getValue() instanceof DisconnectFactory.DisconnectActionListener);
}

@Stop
public void stop() {
    LOG.info("Mqttv3Client stoping!");

    if (clnt != null) {
        for (Entry<String, ObjectPool<IMqttAsyncClient>> entry : clnt.entrySet()) {
            try {
                entry.getValue().close();
            } catch (Exception e) {
                LOG.error(e.getMessage(), e.fillInStackTrace());
            }
        }
        clnt = null;
    }

    LOG.info("Mqttv3Client finish stoping!");
}

public MqttCallBackImpl (IMqttAsyncClient  client,String clientID,String logLevel,String encoding) {
    super(encoding);        
    this.clientID = clientID;
    this.logLevel = logLevel;
    this.client = client;
    this.encoding = encoding;
}

public Builder(final IMqttAsyncClient client) {
    this.client = client;
    this.connectOptions = new MqttConnectOptions();
    this.closeFactory = new CloseFactory(client);
    this.connectFactory = new ConnectFactory(this.client, this.connectOptions);
    this.disconnectFactory = new DisconnectFactory(client);
    this.publishFactory = new PublishFactory(client);
    this.subscribeFactory = new SubscribeFactory(client);
    this.unsubscribeFactory = new UnsubscribeFactory(client);
    this.backpressureStrategy = BackpressureStrategy.BUFFER;
}

@Test
public void whenAValidBackpressureStrategyThenTheAccessorReturnsIt() throws MqttException {
    BackpressureStrategy expected = BackpressureStrategy.BUFFER;
    Builder builder = PahoObservableMqttClient.builder(Mockito.mock(IMqttAsyncClient.class))
    .setBackpressureStrategy(expected);
    Assert.assertNotNull(builder);
    Assert.assertNotNull(builder.getBackpressureStrategy());
    Assert.assertEquals(expected, builder.getBackpressureStrategy());
}

@Test
public void whenGetClientIdIsCalledItReturnsPahoClientId() {
    final String expectedClientId = "clientId";
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.getClientId()).thenReturn(expectedClientId);
    final Builder builder = new PahoObservableMqttClient.Builder(client);
    final PahoObservableMqttClient target = builder.build();
    Assert.assertEquals(expectedClientId, target.getClientId());
}

@Test
public void whenGetBrokerUriIsCalledItReturnsPahoServerUrl() {
    final String expectedBrokerUri = "brokerUri";
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.getServerURI()).thenReturn(expectedBrokerUri);
    final Builder builder = new PahoObservableMqttClient.Builder(client);
    final PahoObservableMqttClient target = builder.build();
    Assert.assertEquals(expectedBrokerUri, target.getBrokerUri());
}

@Test
public void whenThePahoClientIsConnectedIsConnectedReturnsTrue() {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.isConnected()).thenReturn(true);
    final Builder builder = new PahoObservableMqttClient.Builder(client);
    final PahoObservableMqttClient target = builder.build();
    Assert.assertEquals(true, target.isConnected());
}

@Test
public void whenCreateIsCalledThenAnObservableIsReturned() throws Exception {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final MqttConnectOptions options = Mockito.mock(MqttConnectOptions.class);
    final ConnectFactory factory = new ConnectFactory(client, options);
    final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor.forClass(IMqttActionListener.class);
    final Completable obs = factory.create();
    Assert.assertNotNull(obs);
    obs.subscribe();
    Mockito.verify(client).connect(Mockito.same(options), Mockito.isNull(),
            actionListener.capture());
    Assert.assertTrue(actionListener.getValue() instanceof ConnectFactory.ConnectActionListener);
}

@Test
public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled() throws Throwable {
    expectedException.expectCause(isA(MqttException.class));
    final MqttConnectOptions options = Mockito.mock(MqttConnectOptions.class);
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.connect(Mockito.same(options), Mockito.isNull(),
            Mockito.any(ConnectFactory.ConnectActionListener.class)))
            .thenThrow(new MqttException(MqttException.REASON_CODE_CLIENT_CONNECTED));
    final ConnectFactory factory = new ConnectFactory(client, options);
    final Completable obs = factory.create();
    obs.blockingAwait();
}

@Test(expected=NullPointerException.class)
public void whenANullTopicsIsSuppliedThenAnExceptionIsThrown() {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final SubscribeFactory factory = new SubscribeFactory(client);
    final String[] topics = null;
    final int[] qos = new int[]{ 1, 2 };
    factory.create(topics, qos, BackpressureStrategy.ERROR);
}

@Test(expected=NullPointerException.class)
public void whenANullQoSIsSuppliedThenAnExceptionIsThrown() {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final SubscribeFactory factory = new SubscribeFactory(client);
    final String[] topics = new String[]{ "topic1", "topic2" };;
    final int[] qos = null;
    factory.create(topics, qos, BackpressureStrategy.ERROR);
}

@Test(expected=NullPointerException.class)
public void whenANullBackpressureStrategyIsSuppliedThenAnExceptionIsThrown() {
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final SubscribeFactory factory = new SubscribeFactory(client);
    final String[] topics = new String[]{ "topic1", "topic2" };
    final int[] qos = new int[]{ 1, 2 };
    factory.create(topics, qos, null);
}

@Test
public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled() throws Throwable {
    expectedException.expectCause(isA(MqttException.class));
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.publish(Mockito.any(String.class), 
            Mockito.any(byte[].class), Mockito.any(int.class), 
            Mockito.any(boolean.class), Mockito.isNull(),
            Mockito.any(PublishFactory.PublishActionListener.class)))
    .thenThrow(new MqttException(MqttException.REASON_CODE_CLIENT_CONNECTED));
    final PublishFactory factory = new PublishFactory(client);
    final Single<PublishToken> obs = factory.create("topic1", Mockito.mock(MqttMessage.class));
    obs.blockingGet();
}

@Test
public void whenOnSuccessIsCalledThenObserverOnNextAndOnCompletedAreCalled() throws Exception {
    @SuppressWarnings("unchecked")
    final SingleEmitter<MqttToken> observer = Mockito.mock(SingleEmitter.class);
    final PublishActionListener listener = new PublishFactory.PublishActionListener(observer);
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.getClientId()).thenReturn("client_id");
    final IMqttToken iMqttDeliveryToken = Mockito.mock(IMqttToken.class);
    Mockito.when(iMqttDeliveryToken.getClient()).thenReturn(client);
    Mockito.when(iMqttDeliveryToken.getMessageId()).thenReturn(123);
    Mockito.when(iMqttDeliveryToken.getSessionPresent()).thenReturn(false);
    Mockito.when(iMqttDeliveryToken.getGrantedQos()).thenReturn(new int[0]);
    Mockito.when(iMqttDeliveryToken.getTopics()).thenReturn(new String[]{"topic"});
    final ArgumentCaptor<MqttToken> publishToken = ArgumentCaptor.forClass(MqttToken.class);
    listener.onSuccess(iMqttDeliveryToken);
    Mockito.verify(observer).onSuccess(publishToken.capture());
    Assert.assertNotNull(iMqttDeliveryToken);
    Assert.assertNotNull(publishToken);
    Assert.assertNotNull(publishToken.getValue().getClientId());
    Assert.assertEquals(iMqttDeliveryToken.getClient().getClientId(), publishToken.getValue().getClientId());
    Assert.assertNotNull(publishToken.getValue().getMessageId());
    Assert.assertEquals(iMqttDeliveryToken.getMessageId(), publishToken.getValue().getMessageId());
    Assert.assertNotNull(publishToken.getValue().getTopics());
    Assert.assertArrayEquals(iMqttDeliveryToken.getTopics(), publishToken.getValue().getTopics());
    Assert.assertNotNull(publishToken.getValue().getMessageId());
    Assert.assertEquals(iMqttDeliveryToken.getMessageId(), publishToken.getValue().getMessageId());
}

@Test
public void whenCreateIsCalledThenAnObservableIsReturned() throws Exception {
    // Given
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    final CloseFactory factory = new CloseFactory(client);

    // When
    final Completable obs = factory.create();

    // Then
    Assert.assertNotNull(obs);
    obs.subscribe();
    Mockito.verify(client).close();
}

@Test
public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled() throws Throwable {
    expectedException.expectCause(isA(MqttException.class));
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.doThrow(new MqttException(MqttException.REASON_CODE_CLIENT_CONNECTED)).when(client).close();
    final CloseFactory factory = new CloseFactory(client);
    final Completable obs = factory.create();
    obs.blockingAwait();
}

@Test
public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled() throws Throwable {
    expectedException.expectCause(isA(MqttException.class));
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.unsubscribe(Mockito.any(String[].class), Mockito.isNull(),
            Mockito.any(UnsubscribeFactory.UnsubscribeActionListener.class)))
    .thenThrow(new MqttException(MqttException.REASON_CODE_CLIENT_CONNECTED));
    final UnsubscribeFactory factory = new UnsubscribeFactory(client);
    final Completable obs = factory.create(new String[]{ "topic1", "topic2" });
    obs.blockingAwait();
}

@Test
public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled() throws Throwable {
    expectedException.expectCause(isA(MqttException.class));
    final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
    Mockito.when(client.disconnect(Mockito.isNull(),
            Mockito.any(DisconnectFactory.DisconnectActionListener.class)))
    .thenThrow(new MqttException(MqttException.REASON_CODE_CLIENT_CONNECTED));
    final DisconnectFactory factory = new DisconnectFactory(client);
    final Completable obs = factory.create();
    obs.blockingAwait();
}

@NonNull
public static IMqttAsyncClient client(
        @NonNull final String url,
        @NonNull final String id,
        @NonNull final MqttClientPersistence persistence) throws MqttException {
    return new MqttAsyncClient(url, id, persistence);
}

/**
 * close iMqttAsyncClient quietly.
 * @param iMqttAsyncClient mqtt async client interface.
 */
public static void closeMqttClientQuite(IMqttAsyncClient iMqttAsyncClient) {
    if (iMqttAsyncClient != null) {
        try {
            iMqttAsyncClient.disconnectForcibly();
        } catch (MqttException e) {
            logger.info("Close Mqtt Client quite.", e);
        }
    }
}

@Override
public IMqttAsyncClient create() throws Exception {
    String broker = conf.getConf(id, MqttClientConf.F_mqtt_broker);
    MqttAsyncClient mqttClient = new MqttAsyncClient(broker, createClientId(), createPersistence());
    mqttClient.connect(createConnectOptions()).waitForCompletion();
    return mqttClient;
}

@Override
public void destroyObject(PooledObject<IMqttAsyncClient> p) throws Exception {
    IMqttAsyncClient client = p.getObject();
    try {
        client.disconnect();
    } finally {
        client.close();
    }
}

public void init(InputStream file) throws Exception {
    MqttClientConf props = new MqttClientConf();
    props.init(file);

    String[] ids = props.getGroupIds();
    clnt = new HashMap<>(ids.length, 1);
    for (String id : ids) {
        GenericObjectPoolConfig config = new GenericObjectPoolConfig();
        config.setMaxTotal(props.getConfInt(id, MqttClientConf.F_pool_maxTotal, "100"));
        config.setMaxIdle(props.getConfInt(id, MqttClientConf.F_pool_maxIdle, "10"));
        config.setMinIdle(props.getConfInt(id, MqttClientConf.F_pool_minIdle, "1"));
        clnt.put(id, new GenericObjectPool<IMqttAsyncClient>(new MqttAsyncClientFactory(id, props), config));
    }
}

@Override
public IMqttAsyncClient borrowMqttClient(String id) {
    try {
        return clnt.get(id).borrowObject();
    } catch (Exception e) {
        LOG.error(e.getMessage(), e.fillInStackTrace());
    }
    return null;
}

@Override
public void returnMqttClient(String id, IMqttAsyncClient mqttClient) {
    try {
        clnt.get(id).returnObject(mqttClient);
    } catch (Exception e) {
        LOG.error(e.getMessage(), e.fillInStackTrace());
    }
}

/**
 * @param client
 * @throws MqttException
 */
public static void disconnectAndCloseClient(IMqttAsyncClient client) throws MqttException {
  if (client != null) {
    if (client.isConnected()) {
      IMqttToken token = client.disconnect(null, null);
      token.waitForCompletion();
    }
    client.close();
  }
}

/**
 * @param mqttClient
 * @param reportStream
 */
public MqttV3Receiver(IMqttAsyncClient mqttClient, PrintStream reportStream) {
  String methodName = Utility.getMethodName();
  log.entering(className, methodName);

  this.reportStream = reportStream;
  connected = true;

  clientId = mqttClient.getClientId();

  log.exiting(className, methodName);
}

/**
 * Creates a new ClientComms object, using the specified module to handle
 * the network calls.
 */
public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, MqttPingSender pingSender) throws MqttException {
    this.conState = DISCONNECTED;
    this.client     = client;
    this.persistence = persistence;
    this.pingSender = pingSender;
    this.pingSender.init(this);

    this.tokenStore = new CommsTokenStore(getClient().getClientId());
    this.callback   = new CommsCallback(this);
    this.clientState = new ClientState(persistence, tokenStore, this.callback, this, pingSender);

    callback.setClientState(clientState);
    log.setResourceName(getClient().getClientId());
}

public ConnectFactory(final IMqttAsyncClient client, final MqttConnectOptions options) {
    super(client);
    this.options = Objects.requireNonNull(options);
}

public PublishFactory(final IMqttAsyncClient client) {
    super(client);
}

public IMqttAsyncClient getClient() {
    return this.client;
}

public static Builder builder(final IMqttAsyncClient client) {
    return new Builder(client);
}

public CloseFactory(final IMqttAsyncClient client) {
    super(client);
}

public UnsubscribeFactory(final IMqttAsyncClient client) {
    super(client);
}