What's new? 1. Add license for dock. 2. Modify the logic corresponding to t...

sean.zhou

2023-02-24 a7aaeabc7873a0eafb4a7ecad7f65b018b7a9bc9

 src/main/java/com/dji/sample/component/mqtt/service/impl/MessageSenderServiceImpl.java

@@ -12,7 +12,6 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;

/**
@@ -32,7 +31,7 @@

    public void publish(String topic, CommonTopicResponse response) {
        try {

            log.info("send topic: {}, payload: {}", topic, response.toString());
            messageGateway.publish(topic, mapper.writeValueAsBytes(response));
        } catch (JsonProcessingException e) {
            log.info("Failed to publish the message. {}", response.toString());
@@ -49,28 +48,29 @@
        }
    }

    public Optional<ServiceReply> publishWithReply(String topic, CommonTopicResponse response) {
    public ServiceReply publishWithReply(String topic, CommonTopicResponse response) {
        return this.publishWithReply(ServiceReply.class, topic, response, 2);
    }

    public <T> Optional<T> publishWithReply(Class<T> clazz, String topic, CommonTopicResponse response, int retryTime) {
    public <T> T publishWithReply(Class<T> clazz, String topic, CommonTopicResponse response, int retryTime) {
        log.info("send topic: {}, payload: {}", topic, response.toString());
        AtomicInteger time = new AtomicInteger(0);
        // Retry three times
        while (time.getAndIncrement() < retryTime) {
        while (time.getAndIncrement() <= retryTime) {
            this.publish(topic, response);

            Chan<CommonTopicReceiver<T>> chan = Chan.getInstance();
            // If the message is not received in 0.5 seconds then resend it again.
            CommonTopicReceiver<T> receiver = chan.get(response.getMethod());
            CommonTopicReceiver<T> receiver = chan.get(response.getTid());
            if (receiver == null) {
                continue;
            }
            // Need to match tid and bid.
            if (receiver.getTid().equals(response.getTid()) &&
                    receiver.getBid().equals(response.getBid())) {
                return Optional.ofNullable(receiver.getData());
                return receiver.getData();
            }
        }
        return Optional.empty();
        throw new RuntimeException("No message reply received.");
    }
}