public <T> CommonTopicResponse<T> publishWithReply(Class<T> clazz, String topic, CommonTopicRequest request,
                                                       int retryCount, long timeout) {
    AtomicInteger time = new AtomicInteger(0);
    boolean hasBid = StringUtils.hasText(request.getBid());
    request.setBid(hasBid ? request.getBid() : UUID.randomUUID().toString());
    // Retry
    while (time.getAndIncrement() <= retryCount) {
        this.publish(topic, request);

        // If the message is not received in 3 seconds then resend it again.
        CommonTopicResponse<T> receiver = Chan.getInstance(request.getTid(), true).get(request.getTid(), timeout);
        // Need to match tid and bid.
        if (Objects.nonNull(receiver)
                && receiver.getTid().equals(request.getTid())
                && receiver.getBid().equals(request.getBid())) {
            if (clazz.isAssignableFrom(receiver.getData().getClass())) {
                return receiver;
            }
            throw new TypeMismatchException(receiver.getData(), clazz);
        }
        // It must be guaranteed that the tid and bid of each message are different.
        if (!hasBid) {
            request.setBid(UUID.randomUUID().toString());
        }
        request.setTid(UUID.randomUUID().toString());
    }
    throw new CloudSDKException(CloudSDKErrorEnum.MQTT_PUBLISH_ABNORMAL, "No message reply received.");
}

public Result liveStart(LiveTypeDTO liveParam) {
      // Check if this lens is available live.
        Result<DeviceDTO> responseResult = this.checkBeforeLive(liveParam.getVideoId());
        if (CommonConstant.SC_OK_200 != responseResult.getCode() ) {
            return responseResult;
        }


        ILivestreamUrl url = LiveStreamProperty.get(liveParam.getUrlType());
        /**
         * 解析 agora url
         */
        if(liveParam.getUrlType()==UrlTypeEnum.AGORA){
            LivestreamAgoraUrl agoraUrl=new LivestreamAgoraUrl();
            Map<String,String> urlMap= HttpUtil.decodeParamMap(liveParam.getUrl(), Charset.forName("UTF-8"));
            agoraUrl.setSn(urlMap.get("sn"));
            agoraUrl.setChannel(urlMap.get("channel"));
            agoraUrl.setToken(urlMap.get("token"));
            agoraUrl.setUid(Integer.valueOf(urlMap.get("uid")));
            url=agoraUrl;
        }

        url = setExt(liveParam.getUrlType(), url, liveParam.getVideoId());

        TopicServicesResponse<ServicesReplyData<String>> response = abstractLivestreamService.liveStartPush(
                SDKManager.getDeviceSDK(responseResult.getResult().getDeviceSn()),
                new LiveStartPushRequest()
                        .setUrl(url)
                        .setUrlType(liveParam.getUrlType())
                        .setVideoId(liveParam.getVideoId())
                        .setVideoQuality(liveParam.getVideoQuality()));

        if (!response.getData().getResult().isSuccess()) {
            return Result.error(response.getData().getResult().getMessage());
        }

        LiveDTO live = new LiveDTO();

        switch (liveParam.getUrlType()) {
            case AGORA:
                break;
            case RTMP:
                live.setUrl(url.toString().replace("rtmp", "webrtc"));
                break;
            case GB28181:
                LivestreamGb28181Url gb28181 = (LivestreamGb28181Url) url;
                live.setUrl(new StringBuilder()
                        .append("webrtc://")
                        .append(gb28181.getServerIP())
                        .append("/live/")
                        .append(gb28181.getAgentID())
                        .append("@")
                        .append(gb28181.getChannel())
                        .toString());
                break;
            case RTSP:
                live.setUrl(response.getData().getOutput());
                break;
            case WHIP:
                live.setUrl(url.toString().replace("whip", "whep"));
                break;
            default:
                return Result.error(LiveErrorCodeEnum.URL_TYPE_NOT_SUPPORTED.getMessage());
        }

        return Result.ok(live);
}

public Result liveStart(LiveTypeDTO liveParam) {
        // Check if this lens is available live.
        Result<DeviceDTO> responseResult = this.checkBeforeLive(liveParam.getVideoId());
        if (CommonConstant.SC_OK_200 != responseResult.getCode()) {
            return responseResult;
        }

        TopicServicesResponse<ServicesReplyData<String>> response = abstractLivestreamService.liveStartPush(
                SDKManager.getDeviceSDK(responseResult.getResult().getDeviceSn()), new LiveStartPushRequest()
                        .setUrl(liveParam.getUrl())
                        .setUrlType(liveParam.getUrlType())
                        .setVideoId(liveParam.getVideoId())
                        .setVideoQuality(liveParam.getVideoQuality()));

        if (!response.getData().getResult().isSuccess()) {
            return Result.error(response.getData().getResult().getMessage());
        }

        LiveDTO live = new LiveDTO();

        return Result.OK(live);
}

/**
 * @author sean
 * @version 1.7
 * @date 2023/5/23
 */
public class LiveStartPushRequest extends BaseModel {

    @NotNull
    private UrlTypeEnum urlType;

    /**
     * RTMP: (rtmp://xxxxxxx) Example: rtmp://192.168.1.1:8080/live
     * RTSP：(uerName&password&port) Example: userName=dji-cloud-api&password=123456&port=8080
     * GB28181：(serverIP&serverPort&serverID&agentID&agentPassword&localPort&channel)
     *  Example: serverIP=192.168.1.1&serverPort=8080&serverID=34000000000000000000&agentID=
     *  300000000010000000000&agentPassword=0000000&localPort=7060&channel=340000000000000000000
     * AGORA：(channel&sn&token&uid)
     *  Example: channel=1ZNDH360010162_39-0-7&sn=1ZNDH360010162&token=006dca67721582a48768ec4d8
     *  17b7b25a86IAB4cw2JgN6iX8BpTPdc3e4S1Iendz94IFJ56aSXKvzAJei27MqF2zyCIgCLIIoBt41+YAQAAQC3jX
     *  5gAgC3jX5gAwC3jX5gBAC3jX5g&uid=50000
     *  Notice: The token generated by Shengwang may have special characters such as '+' ' ',
     *  and need to do url encode, otherwise there will be a parsing error on the PILOT side
     */
    @NotBlank
    private String url;

    /**
     * The format is #{uav_sn}/#{camera_id}/#{video_index},
     * drone serial number/payload and mounted location enumeration value/payload lens numbering
     */
    @NotNull
    private VideoId videoId;

    @NotNull
    private VideoQualityEnum videoQuality;

    public LiveStartPushRequest() {
    }

    @Override
    public String toString() {
        return "LiveStartPushRequest{" +
                "urlType=" + urlType +
                ", url='" + url + '\'' +
                ", videoId=" + videoId +
                ", videoQuality=" + videoQuality +
                '}';
    }

    public UrlTypeEnum getUrlType() {
        return urlType;
    }

    public LiveStartPushRequest setUrlType(UrlTypeEnum urlType) {
        this.urlType = urlType;
        return this;
    }

    public String getUrl() {
        return url;
    }

    public LiveStartPushRequest setUrl(String url) {
        this.url = url;
        return this;
    }

    public VideoId getVideoId() {
        return videoId;
    }

    public LiveStartPushRequest setVideoId(VideoId videoId) {
        this.videoId = videoId;
        return this;
    }

    public VideoQualityEnum getVideoQuality() {
        return videoQuality;
    }

    public LiveStartPushRequest setVideoQuality(VideoQualityEnum videoQuality) {
        this.videoQuality = videoQuality;
        return this;
    }
}

/**
 * 一键起飞
 */
@RequestMapping("/takeoffToPoint")
private HttpResultResponse<Boolean> takeoffToPoint(@RequestBody JSONObject param){
    String sn=param.getString("sn");

    // 设置飞行参数
    TakeoffToPointParam takeParam=new TakeoffToPointParam();
    takeParam.setTargetLongitude(param.getDouble("target_longitude"));
    takeParam.setTargetLatitude(param.getDouble("target_latitude"));
    takeParam.setTargetLatitude(param.getDouble("target_height"));
    takeParam.setSecurityTakeoffHeight(param.getDouble("security_takeoff_height"));
    takeParam.setRthAltitude(param.getDouble("rth_altitude"));
    takeParam.setMaxSpeed(param.getDouble("max_speed"));
    takeParam.setRcLostAction(RcLostActionEnum.RETURN_HOME);
    takeParam.setExitWaylineWhenRcLost(ExitWaylineWhenRcLostEnum.EXECUTE_RC_LOST_ACTION);
    takeParam.setRthMode(RthModeEnum.SMART_HEIGHT);
    takeParam.setCommanderModeLostAction(CommanderModeLostActionEnum.EXECUTE_RC_LOST_ACTION);
    takeParam.setCommanderFlightMode(CommanderFlightModeEnum.SMART_HEIGHT);
    takeParam.setCommanderFlightHeight(param.getFloat("commander_flight_height"));

    return controlService.takeoffToPoint(sn, takeParam);
}

/**
 * 返航控制
 */

@RequestMapping("/returnHome")
private HttpResultResponse<Boolean> returnHome(@RequestParam(value = "sn")String sn){
    return controlService.controlDockDebug(sn, RemoteDebugMethodEnum.RETURN_HOME,null);
}