Java SDK Access Example

<!-- amqp 1.0 qpid client -->
 <dependency>
   <groupId>org.apache.qpid</groupId>
   <artifactId>qpid-jms-client</artifactId>
   <version>0.61.0</version>
 </dependency>

// Change the values of the following parameters as required.
AmqpClientOptions options = AmqpClientOptions.builder()
                .host(AmqpConstants.HOST)
                .port(AmqpConstants.PORT)
                .accessKey(AmqpConstants.ACCESS_KEY)
                .accessCode(AmqpConstants.ACCESS_CODE)
                .queuePrefetch(1000) // The SDK allocates the queue with the memory size set in this parameter to receive messages. If the client memory size is small, set this parameter to a smaller value.
                .build();
        AmqpClient amqpClient = new AmqpClient(options);
        amqpClient.initialize();

package com.iot.amqp;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionExtensions;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.apache.qpid.jms.JmsQueue;
import org.apache.qpid.jms.transports.TransportOptions;
import org.apache.qpid.jms.transports.TransportSupport;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

@Slf4j
public class AmqpClient {
    private final AmqpClientOptions options;
    private Connection connection;
    private Session session;
    private final Set<MessageConsumer> consumerSet = Collections.synchronizedSet(new HashSet<>());

    public AmqpClient(AmqpClientOptions options) {
        this.options = options;
    }

    public String getId() {
        return options.getClientId();
    }

    public void initialize() throws Exception {
        String connectionUrl = options.generateConnectUrl();
        log.info("connectionUrl={}", connectionUrl);
        JmsConnectionFactory cf = new JmsConnectionFactory(connectionUrl);
        // Trust the server.
        TransportOptions to = new TransportOptions();
        to.setTrustAll(true);
        cf.setSslContext(TransportSupport.createJdkSslContext(to));
        String userName = "accessKey=" + options.getAccessKey();
        cf.setExtension(JmsConnectionExtensions.USERNAME_OVERRIDE.toString(), (connection, uri) -> {
            String newUserName = userName;
            if (connection instanceof JmsConnection) {
                newUserName = ((JmsConnection) connection).getUsername();
            }
            if (StringUtils.isEmpty(options.getInstanceId())) {
            // userName of IoTDA is in the following format: accessKey=${accessKey}|timestamp=${timestamp}.
                return newUserName + "|timestamp=" + System.currentTimeMillis();
            } else {
            //If multiple Standard Editions are purchased in the same region, the format of userName is accessKey=${accessKey}|timestamp=${timestamp}|instanceId=${instanceId}.
                return newUserName + "|timestamp=" + System.currentTimeMillis() + "|instanceId=" + options.getInstanceId();
            }
        });
        // Create a connection.
        connection = cf.createConnection(userName, options.getAccessCode());
        // Create sessions Session.CLIENT_ACKNOWLEDGE and Session.AUTO_ACKNOWLEDGE. For Session.CLIENT_ACKNOWLEDGE, manually call message.acknowledge() after receiving a message. For Session.AUTO_ACKNOWLEDGE, the SDK automatically responds with an ACK message (recommended).
        session = connection.createSession(false, options.isAutoAcknowledge() ? Session.AUTO_ACKNOWLEDGE : Session.CLIENT_ACKNOWLEDGE);
        connection.start();
    }

    public MessageConsumer newConsumer(String queueName) throws Exception {
        if (connection == null || !(connection instanceof JmsConnection) || ((JmsConnection) connection).isClosed()) {
            throw new Exception("create consumer failed,the connection is disconnected.");
        }
        MessageConsumer consumer;

        consumer = session.createConsumer(new JmsQueue(queueName));
        if (consumer != null) {
            consumerSet.add(consumer);
        }
        return consumer;
    }

    public void close() {
        consumerSet.forEach(consumer -> {
            try {
                consumer.close();
            } catch (JMSException e) {
                log.warn("consumer close error,exception is ", e);
            }
        });

        if (session != null) {
            try {
                session.close();
            } catch (JMSException e) {
                log.warn("session close error,exception is ", e);
            }
        }

        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException e) {
                log.warn("connection close error,exception is", e);
            }
        }
    }
}

package com.iot.amqp;

import lombok.Builder;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;

import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

@Data
@Builder
public class AmqpClientOptions {
    private String host;
    @Builder.Default
    private int port = 5671;
    private String accessKey;
    private String accessCode;
    private String clientId;

    /**
     * Specifies the instance ID. This parameter is required when multiple instances of the Standard Edition are purchased in the same region.
     */
    private String instanceId;

    /**
     * Only true is supported.
     */
    @Builder.Default
    private boolean useSsl = true;

    /**
     * IoTDA supports default only.
     */
    @Builder.Default
    private String vhost = "default";

    /**
     * IoTDA supports PLAIN only.
     */
    @Builder.Default
    private String saslMechanisms = "PLAIN";

    /**
     * true: The SDK automatically responds with an ACK message (default).
     * false: After receiving a message, manually call message.acknowledge().
     */
    @Builder.Default
    private boolean isAutoAcknowledge = true;

    /**
     * Reconnection delay (ms)
     */
    @Builder.Default
    private long reconnectDelay = 3000L;

    /**
     * Maximum reconnection delay (ms). The reconnection delay increases with the number of reconnection times.
     */
    @Builder.Default
    private long maxReconnectDelay = 30 * 1000L;

    /**
     * Maximum number of reconnection times. The default value is –1, indicating that the number of reconnection times is not limited.
     */
    @Builder.Default
    private long maxReconnectAttempts = -1;

    /**
     * Idle timeout interval. If the peer end does not send an AMQP frame within the interval, the connection will be cut off. The default value is 30000, in milliseconds.
     */
    @Builder.Default
    private long idleTimeout = 30 * 1000L;

    /**
     * The values below control how many messages the remote peer can send to the client and be held in a pre-fetch buffer for each consumer instance.
     */
    @Builder.Default
    private int queuePrefetch = 1000;

    /**
     * Extended parameters
     */
    private Map<String, String> extendedOptions;

    public String generateConnectUrl() {
        String uri = MessageFormat.format("{0}://{1}:{2}", (useSsl ? "amqps" : "amqp"), host, String.valueOf(port));
        Map<String, String> uriOptions = new HashMap<>();
        uriOptions.put("amqp.vhost", vhost);
        uriOptions.put("amqp.idleTimeout", String.valueOf(idleTimeout));
        uriOptions.put("amqp.saslMechanisms", saslMechanisms);

        Map<String, String> jmsOptions = new HashMap<>();
        jmsOptions.put("jms.prefetchPolicy.queuePrefetch", String.valueOf(queuePrefetch));
        if (StringUtils.isNotEmpty(clientId)) {
            jmsOptions.put("jms.clientID", clientId);
        } else {
            jmsOptions.put("jms.clientID", UUID.randomUUID().toString());
        }
        jmsOptions.put("failover.reconnectDelay", String.valueOf(reconnectDelay));
        jmsOptions.put("failover.maxReconnectDelay", String.valueOf(maxReconnectDelay));
        if (maxReconnectAttempts > 0) {
            jmsOptions.put("failover.maxReconnectAttempts", String.valueOf(maxReconnectAttempts));
        }
        if (extendedOptions != null) {
            for (Map.Entry<String, String> option : extendedOptions.entrySet()) {
                if (option.getKey().startsWith("amqp.") || option.getKey().startsWith("transport.")) {
                    uriOptions.put(option.getKey(), option.getValue());
                } else {
                    jmsOptions.put(option.getKey(), option.getValue());
                }
            }
        }
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(uriOptions.entrySet().stream()
                .map(option -> MessageFormat.format("{0}={1}", option.getKey(), option.getValue()))
                .collect(Collectors.joining("&", "failover:(" + uri + "?", ")")));
        stringBuilder.append(jmsOptions.entrySet().stream()
                .map(option -> MessageFormat.format("{0}={1}", option.getKey(), option.getValue()))
                .collect(Collectors.joining("&", "?", "")));
        return stringBuilder.toString();
    }
}

package com.iot.amqp;


public interface AmqpConstants {
    /**
     * AMQP access domain name
     * eg: "****.iot-amqps.cn-north-4.myhuaweicloud.com";
     */
    String HOST = "127.0.0.1";   

    /**
     * AMQP access port
     */
    int PORT = 5671;

    /**
     * Access key
     * A timestamp does not need to be combined.
     */
    String ACCESS_KEY = "accessKey";

    /**
     * Access code
     */
    String ACCESS_CODE = "accessCode";

    /**
     * Default queue
     */
    String DEFAULT_QUEUE = "DefaultQueue";
}