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

/**
 * get MqttClient by clientKey
 * @param clientKey
 * @return
 * @throws MqttException
 */
public static MqttClient getMqttClient(String serverURI, String clientId,StringRedisTemplate redisTemplate) 
                throws MqttException{
     String clientKey=serverURI.concat(clientId);
     if(clientMap.get(clientKey)==null){
         lock.lock();
             if(clientMap.get(clientKey)==null){
                 MqttClientPersistence persistence = new MemoryPersistence();

                 MqttClient client = new MqttClient(serverURI, clientId, persistence);
                 MqttConnectOptions connOpts = new MqttConnectOptions();

                 MqttCallback callback = new IMMqttCallBack(client,redisTemplate);
                 client.setCallback(callback);

                 connOpts.setCleanSession(true);
                 client.connect(connOpts);
                 clientMap.put(clientKey, client);
             }
          lock.unlock();
     }
      return clientMap.get(clientKey);
}

public MqttClientPersistence getPersistence() {
  if (StringUtils.isEmpty(type) || type.equals("memory")) {
    log.info("Initializing default memory persistence!");
    return new MemoryPersistence();
  } else if (type.equals("file")) {
    if (StringUtils.isEmpty(path)) {
      log.info("Initializing default file persistence!");
      return new MqttDefaultFilePersistence();
    } else {
      log.info("Initializing file persistence using directory: {}", path);
      return new MqttDefaultFilePersistence(path);
    }
  } else {
    log.error("Unknown persistence option: {}. Only 'memory' and 'file' are supported at the moment!", type);
    throw new IllegalArgumentException("Unknown persistence option: " + type + "!");
  }
}

public MqttClientPersistence getPersistence() {
    if (StringUtils.isEmpty(type) || type.equals("memory")) {
        log.info("Initializing default memory persistence!");
        return new MemoryPersistence();
    } else if (type.equals("file")) {
        if (StringUtils.isEmpty(path)) {
            log.info("Initializing default file persistence!");
            return new MqttDefaultFilePersistence();
        } else {
            log.info("Initializing file persistence using directory: {}", path);
            return new MqttDefaultFilePersistence(path);
        }
    } else {
        log.error("Unknown persistence option: {}. Only 'memory' and 'file' are supported at the moment!", type);
        throw new IllegalArgumentException("Unknown persistence option: " + type + "!");
    }
}

public MqttClientPersistence getPersistence() {
  if (StringUtils.isEmpty(type) || type.equals("memory")) {
    log.info("Initializing default memory persistence!");
    return new MemoryPersistence();
  } else if (type.equals("file")) {
    if (StringUtils.isEmpty(path)) {
      log.info("Initializing default file persistence!");
      return new MqttDefaultFilePersistence();
    } else {
      log.info("Initializing file persistence using directory: {}", path);
      return new MqttDefaultFilePersistence(path);
    }
  } else {
    log.error("Unknown persistence option: {}. Only 'memory' and 'file' are supported at the moment!", type);
    throw new IllegalArgumentException("Unknown persistence option: " + type + "!");
  }
}

/**
 * Constructor - create an MqttConnection to communicate with MQTT server
 * 
 * @param service
 *            our "parent" service - we make callbacks to it
 * @param serverURI
 *            the URI of the MQTT server to which we will connect
 * @param clientId
 *            the name by which we will identify ourselves to the MQTT
 *            server
 * @param persistence
 *            the persistence class to use to store in-flight message. If
 *            null then the default persistence mechanism is used
 * @param clientHandle
 *            the "handle" by which the activity will identify us
 */
MqttConnection(MqttService service, String serverURI, String clientId,
        MqttClientPersistence persistence, String clientHandle) {
    this.serverURI = serverURI;
    this.service = service;
    this.clientId = clientId;
    this.persistence = persistence;
    this.clientHandle = clientHandle;

    StringBuilder stringBuilder = new StringBuilder(this.getClass().getCanonicalName());
    stringBuilder.append(" ");
    stringBuilder.append(clientId);
    stringBuilder.append(" ");
    stringBuilder.append("on host ");
    stringBuilder.append(serverURI);
    wakeLockTag = stringBuilder.toString();
}

/**
 * Constructor - create an MqttConnection to communicate with MQTT server
 * 
 * @param service
 *            our "parent" service - we make callbacks to it
 * @param serverURI
 *            the URI of the MQTT server to which we will connect
 * @param clientId
 *            the name by which we will identify ourselves to the MQTT
 *            server
 * @param persistence the persistence class to use to store in-flight message. If null then the
 *          default persistence mechanism is used
 * @param clientHandle
 *            the "handle" by which the activity will identify us
 */
MqttConnection(MqttService service, String serverURI, String clientId,
        MqttClientPersistence persistence, String clientHandle) {
    this.serverURI = serverURI.toString();
    this.service = service;
    this.clientId = clientId;
    this.persistence = persistence;
    this.clientHandle = clientHandle;

    StringBuffer buff = new StringBuffer(this.getClass().getCanonicalName());
    buff.append(" ");
    buff.append(clientId);
    buff.append(" ");
    buff.append("on host ");
    buff.append(serverURI);
    wakeLockTag = buff.toString();
}

/**
 * Constructor - create an MqttConnection to communicate with MQTT server
 * 
 * @param service
 *            our "parent" service - we make callbacks to it
 * @param serverURI
 *            the URI of the MQTT server to which we will connect
 * @param clientId
 *            the name by which we will identify ourselves to the MQTT
 *            server
 * @param persistence the persistence class to use to store in-flight message. If null then the
 *          default persistence mechanism is used
 * @param clientHandle
 *            the "handle" by which the activity will identify us
 */
MqttConnection(MqttService service, String serverURI, String clientId,
        MqttClientPersistence persistence, String clientHandle) {
    this.serverURI = serverURI.toString();
    this.service = service;
    this.clientId = clientId;
    this.persistence = persistence;
    this.clientHandle = clientHandle;

    StringBuffer buff = new StringBuffer(this.getClass().getCanonicalName());
    buff.append(" ");
    buff.append(clientId);
    buff.append(" ");
    buff.append("on host ");
    buff.append(serverURI);
    wakeLockTag = buff.toString();
}

/**
 * @param persistence
 * @param client 
 * @param comms
 * @param options 
 * @param userToken  
 * @param userContext
 * @param userCallback
 */
public ConnectActionListener(
    MqttAsyncClient client,
    MqttClientPersistence persistence,
    ClientComms comms,
    MqttConnectOptions options,
    MqttToken userToken,
    Object userContext,
    IMqttActionListener userCallback) {
  this.persistence = persistence;
  this.client = client;
  this.comms = comms;
  this.options = options;
  this.userToken = userToken;
  this.userContext = userContext;
  this.userCallback = userCallback;
  this.originalMqttVersion = options.getMqttVersion();
}

protected ClientState(MqttClientPersistence persistence, CommsTokenStore tokenStore, 
        CommsCallback callback, ClientComms clientComms, MqttPingSender pingSender) throws MqttException {

    log.setResourceName(clientComms.getClient().getClientId());
    log.finer(CLASS_NAME, "<Init>", "" );

    inUseMsgIds = new Hashtable();
    pendingMessages = new Vector(this.maxInflight);
    pendingFlows = new Vector();
    outboundQoS2 = new Hashtable();
    outboundQoS1 = new Hashtable();
    inboundQoS2 = new Hashtable();
    pingCommand = new MqttPingReq();
    inFlightPubRels = 0;
    actualInFlight = 0;

    this.persistence = persistence;
    this.callback = callback;
    this.tokenStore = tokenStore;
    this.clientComms = clientComms;
    this.pingSender = pingSender;

    restoreState();
}

public MqttAndroidClient createClient(String id, String serverURI, String clientId) {
    MqttClientPersistence mqttClientPersistence = new MemoryPersistence();
    MqttAndroidClient client = new MqttAndroidClient(MyApplication.getContext(), serverURI, clientId, mqttClientPersistence);
    client.setCallback(new MqttCallback() {
        @Override
        public void connectionLost(Throwable cause) {
            LogUtil.e("connectionLost");
            EventBus.getDefault().post(new MQTTActionEvent(Constant.MQTTStatusConstant.CONNECTION_LOST, null, cause));

        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            LogUtil.d("topic is " + topic + ",message is " + message.toString() + ", qos is " + message.getQos());
            EventBus.getDefault().postSticky(new MessageEvent(new EmqMessage(topic, message)));

        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            LogUtil.d("deliveryComplete");


        }
    });

    mClients.put(id, client);

    return client;

}

/**
 * Default constructor
 * 
 * @param serverUri the Server URI to connect to
 * @param clientId the Client ID to connect as
 * @param connectionType the {@link MqttClientConnectType} this instance will be used as
 * @param clientPersistence TODO: add description
 * @throws IllegalArgumentException if the {@code serverUri} is blank or null, the
 *             {@code clientId} is blank or null, or if the {@code connectionType} value is null
 * @throws MqttException if the underlying {@link MqttAsyncClient} instance cannot be created
 */
public PahoAsyncMqttClientService(final String serverUri, final String clientId,
    final MqttClientConnectionType connectionType,
    final MqttClientPersistence clientPersistence)
    throws MqttException
{
    super(connectionType);
    Assert.hasText(serverUri, "'serverUri' must be set!");
    Assert.hasText(clientId, "'clientId' must be set!");
    this.clientPersistence = clientPersistence;
    mqttClient = new MqttAsyncClient(serverUri, clientId, this.clientPersistence);
    mqttClient.setCallback(this);
}

/**
 * Change the default Mqtt persistence settings
 *
 * @param _persistence A custom persistence setting
 * @return the Client instance
 */
public Client setMqttPersistence(MqttClientPersistence _persistence) {
    if (mqttClient != null) {
        throw new RuntimeException("Can not be called while client is running");
    }
    persistence = _persistence;
    return this;
}

/**
 * get MqttClient by clientKey
 * 
 * @param clientKey
 * @return
 * @throws MqttException
 * @throws NoSuchAlgorithmException
 */
private MqttClient getMqttClient(String clientId) throws MqttException {
    MqttClientPersistence persistence = new MemoryPersistence();

    MqttClient client = new MqttClient(broker_address, clientId,
            persistence);

    MqttConnectOptions connOpts = new MqttConnectOptions();
    MqttCallback callback = new MqttCallback() {
        public void messageArrived(String topic, MqttMessage message)
                throws Exception {
            long arriveTime = System.currentTimeMillis();
            String msgID = message.toString();
            for(MsgHandleInterface handle : handleList)
                handle.handle(msgID,topic);
            Object[] str = {msgID,arriveTime};
            arriveQueue.put(str);
        }

        public void deliveryComplete(IMqttDeliveryToken token) {

        }

        public void connectionLost(Throwable cause) {
        }
    };
    client.setCallback(callback);
    connOpts.setKeepAliveInterval(3600);
    connOpts.setCleanSession(true);
    client.connect(connOpts);
    return client;
}

private static MqttClientPersistence newPersistenceProvider(String className) {
    Class<?> clazz = null;
    try {
        clazz = Class.forName(className);
        return (MqttClientPersistence) clazz.newInstance();
    }
    catch (Exception e) {
        throw new IllegalArgumentException(e.getLocalizedMessage(), e);
    }
}

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

protected MqttClientPersistence resolvePersistence() {
    if (persistence ==  PahoPersistence.MEMORY) {
        return new MemoryPersistence();
    } else {
        if (filePersistenceDirectory != null) {
            return new MqttDefaultFilePersistence(filePersistenceDirectory);
        } else {
            return new MqttDefaultFilePersistence();
        }
    }
}

@Override
protected void configure() {
    bind(MqttClientPersistence.class).to(MemoryPersistence.class).in(Singleton.class);

    bind(Registry.class).to(SimpleRegistry.class).in(Singleton.class);

    bind(Engine.class).in(Singleton.class);
    bind(MessagePublisher.class).to(Engine.class);

    expose(Registry.class);
    expose(MessagePublisher.class);
    expose(Engine.class);
}

@Inject
public Engine(final Gson gson, final Registry registry, final Config config,
        final MqttClientPersistence persistence) throws MqttException {
    String topicPrefix = config.getString(Settings.TOPIC_PREFIX).replaceFirst("^/+", "");
    if (!topicPrefix.isEmpty() && !topicPrefix.endsWith("/")) {
        topicPrefix += "/";
    }
    this.topicPrefix = topicPrefix;

    this.reconnectInterval = Duration.ofSeconds(config.getLong(Settings.RECONNECT_INTERVAL));

    this.gson = Objects.requireNonNull(gson);
    this.registry = Objects.requireNonNull(registry);

    this.client = new MqttAsyncClient(
        config.getString(Settings.BROKER_URL),
        config.getString(Settings.CLIENT_ID),
        persistence
    );
    this.client.setCallback(this);

    {
        final String username = config.getString(Settings.BROKER_USERNAME);
        if (username != null && !username.isEmpty()) {
            this.options.setUserName(username);
        }
    }
    {
        final String password = config.getString(Settings.BROKER_PASSWORD);
        if (password != null && !password.isEmpty()) {
            this.options.setPassword(password.toCharArray());
        }
    }

    this.options.setCleanSession(true);
}

protected synchronized void doConnect(final String clientId, final MqttConnectOptions options, final MqttClientPersistence persistence, final boolean retry) throws MqttException {
    if (client == null) {
        client = new MqttClient(options.getServerURIs()[0], clientId, persistence);
        client.setCallback(new MqttCallback(clientId));
        final Thread starter = new MqttClientStartThread(options, retry);
        starter.start();
        try {
            starter.join();
        }
        catch (final InterruptedException e) {
            logger.log(Level.FINE, e.toString(), e);
            Thread.currentThread().interrupt();
        }
    }
}

/**
 * Get an MqttConnection object to represent a connection to a server
 * 
 * @param serverURI specifies the protocol, host name and port to be used to connect to an MQTT server
 * @param clientId specifies the name by which this connection should be identified to the server
 * @param contextId specifies the app conext info to make a difference between apps
 * @return a string to be used by the Activity as a "handle" for this
 *         MqttConnection
 */
public String getClient(String serverURI, String clientId, String contextId,MqttClientPersistence persistence) {
  String clientHandle = serverURI + ":" + clientId+":"+contextId;
  if (!connections.containsKey(clientHandle)) {
    MqttConnection client = new MqttConnection(this, serverURI,
        clientId, persistence, clientHandle);
    connections.put(clientHandle, client);
  }
  return clientHandle;
}

/**
 * Get an MqttConnection object to represent a connection to a server
 * 
 * @param serverURI specifies the protocol, host name and port to be used to connect to an MQTT server
 * @param clientId specifies the name by which this connection should be identified to the server
 * @param contextId specifies the app conext info to make a difference between apps
 * @return a string to be used by the Activity as a "handle" for this
 *         MqttConnection
 */
public String getClient(String serverURI, String clientId, String contextId,MqttClientPersistence persistence) {
  String clientHandle = serverURI + ":" + clientId+":"+contextId;
  if (!connections.containsKey(clientHandle)) {
    MqttConnection client = new MqttConnection(this, serverURI,
        clientId, persistence, clientHandle);
    connections.put(clientHandle, client);
  }
  return clientHandle;
}

/**
 * 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 Builder(final String brokerUri, final String clientId, final MqttClientPersistence persistence)
        throws MqttException {
    this(new MqttAsyncClient(brokerUri, clientId, persistence));
}

public static Builder builder(final String brokerUri, final String clientId,
        final MqttClientPersistence persistence) throws MqttException {
    return builder(new MqttAsyncClient(brokerUri, clientId, persistence));
}

public MQClient(String serverURI, String clientId, MqttClientPersistence persistence) throws MqttException {
    super(serverURI, clientId, persistence);
}

public MqttClientPing(String serverURI, String clientId, MqttClientPersistence persistence, MqttPingSender pingSender) throws MqttException {
    super(serverURI, clientId, persistence);
    aClient = new MqttAsyncClient(serverURI, clientId, persistence, pingSender);
}

public MqttAsyncClientEx(String serverURI, String clientId, MqttClientPersistence persistence) throws MqttException {
    super(serverURI, clientId, persistence);
}

public MqttAsyncClientEx(String serverURI, String clientId, MqttClientPersistence persistence, MqttPingSender pingSender) throws MqttException {
    super(serverURI, clientId, persistence, pingSender);
}

public Jdk16MqttWebSocketAsyncClient(String serverURI, String clientId,
        MqttClientPersistence persistence, MqttPingSender pingSender,
        String loggerName) throws MqttException {
    super(serverURI, clientId, persistence, pingSender, loggerName);
}

public Jdk16MqttWebSocketAsyncClient(String serverURI, String clientId,
        MqttClientPersistence persistence, String loggerName)
        throws MqttException {
    super(serverURI, clientId, persistence, loggerName);
}

public Jdk16MqttWebSocketAsyncClient(String serverURI, String clientId,
        MqttClientPersistence persistence) throws MqttException {
    super(serverURI, clientId, persistence);
}

private MqttClientPersistence createPersistence() throws Exception {
    return (MqttClientPersistence) Class.forName(conf.getConf(id, MqttClientConf.F_mqtt_persistence)).newInstance();
}

public MqttClientPersistence getClientPersistence() {
    return cp;
}

/**
 * Get an MqttConnection object to represent a connection to a server
 *
 * @param serverURI specifies the protocol, host name and port to be used to connect to an MQTT server
 * @param clientId specifies the name by which this connection should be identified to the server
 * @param contextId specifies the app conext info to make a difference between apps
 * @param persistence specifies the persistence layer to be used with this client
 * @return a string to be used by the Activity as a "handle" for this
 *         MqttConnection
 */
public String getClient(String serverURI, String clientId, String contextId, MqttClientPersistence persistence) {
  String clientHandle = serverURI + ":" + clientId+":"+contextId;
  if (!connections.containsKey(clientHandle)) {
    MqttConnection client = new MqttConnection(this, serverURI,
        clientId, persistence, clientHandle);
    connections.put(clientHandle, client);
  }
  return clientHandle;
}

/**
 * Constructor- create an MqttAndroidClient that can be used to communicate
 * with an MQTT server on android
 * 
 * @param context
 *            used to pass context to the callback.
 * @param serverURI
 *            specifies the protocol, host name and port to be used to
 *            connect to an MQTT server
 * @param clientId
 *            specifies the name by which this connection should be
 *            identified to the server
 * @param persistence
 *            the persistence class to use to store in-flight message. If
 *            null then the default persistence mechanism is used
 * @param ackType
 *            how the application wishes to acknowledge a message has been
 *            processed.
 */
public MqttAndroidClient(Context context, String serverURI,
        String clientId, MqttClientPersistence persistence, Ack ackType) {
    myContext = context;
    this.serverURI = serverURI;
    this.clientId = clientId;
    this.persistence = persistence;
    messageAck = ackType;
}

/**
 * constructor
 *
 * @param context
 *            used to pass context to the callback.
 * @param serverURI
 *            specifies the protocol, host name and port to be used to
 *            connect to an MQTT server
 * @param clientId
 *            specifies the name by which this connection should be
 *            identified to the server
 * @param persistence
 *          the persistence class to use to store in-flight message. If null then the
 *          default persistence mechanism is used
 * @param ackType
 *            how the application wishes to acknowledge a message has been processed.
 */
public MqttAndroidClient(Context context, String serverURI,
                         String clientId, MqttClientPersistence persistence, Ack ackType) {
    myContext = context;
    this.serverURI = serverURI;
    this.clientId = clientId;
    this.persistence = persistence;
    messageAck = ackType;
}

/**
 * constructor
 * 
 * @param context
 *            used to pass context to the callback.
 * @param serverURI
 *            specifies the protocol, host name and port to be used to
 *            connect to an MQTT server
 * @param clientId
 *            specifies the name by which this connection should be
 *            identified to the server
 * @param persistence
 *          the persistence class to use to store in-flight message. If null then the
 *          default persistence mechanism is used
 * @param ackType 
 *            how the application wishes to acknowledge a message has been processed.
 */
public MqttAndroidClient(Context context, String serverURI,
        String clientId, MqttClientPersistence persistence, Ack ackType) {
    myContext = context;
    this.serverURI = serverURI;
    this.clientId = clientId;
    this.persistence = persistence;
    messageAck = ackType;
}

/**
 * Constructor - create an MqttAndroidClient that can be used to communicate
 * with an MQTT server on android
 * 
 * @param ctx
 *            Application's context
 * @param serverURI
 *            specifies the protocol, host name and port to be used to
 *            connect to an MQTT server
 * @param clientId
 *            specifies the name by which this connection should be
 *            identified to the server
 * @param persistence
 *            The object to use to store persisted data
 */
public MqttAndroidClient(Context ctx, String serverURI, String clientId, MqttClientPersistence persistence) {
    this(ctx, serverURI, clientId, persistence, Ack.AUTO_ACK);
}

/**
 * Get the QoS 1 and 2 in-flight message persistence handler.
 * @return the value
 */
public MqttClientPersistence getPersistence() {
    return persistence;
}

/**
 * QoS 1 and 2 in-flight message persistence.
 * <p>
 * optional. default: use memory persistence.
 * @param persistence
 */
public void setPersistence(MqttClientPersistence persistence) {
    this.persistence = persistence;
}