package cn.skcks.docking.gb28181.wvp.service.catalog;

import cn.skcks.docking.gb28181.common.json.JsonException;
import cn.skcks.docking.gb28181.common.xml.XmlUtils;
import cn.skcks.docking.gb28181.core.sip.gb28181.constant.CmdType;
import cn.skcks.docking.gb28181.core.sip.gb28181.constant.GB28181Constant;
import cn.skcks.docking.gb28181.core.sip.message.subscribe.GenericSubscribe;
import cn.skcks.docking.gb28181.core.sip.utils.SipUtil;
import cn.skcks.docking.gb28181.wvp.orm.mybatis.dynamic.model.WvpProxyDevice;
import cn.skcks.docking.gb28181.wvp.orm.mybatis.dynamic.model.WvpProxyDocking;
import cn.skcks.docking.gb28181.wvp.service.device.DeviceService;
import cn.skcks.docking.gb28181.wvp.service.docking.DockingService;
import cn.skcks.docking.gb28181.wvp.sip.message.message.catalog.dto.CatalogItemDTO;
import cn.skcks.docking.gb28181.wvp.sip.message.message.catalog.dto.CatalogRequestDTO;
import cn.skcks.docking.gb28181.wvp.sip.message.message.catalog.dto.CatalogResponseDTO;
import cn.skcks.docking.gb28181.wvp.sip.request.SipRequestBuilder;
import cn.skcks.docking.gb28181.wvp.sip.sender.SipSender;
import cn.skcks.docking.gb28181.wvp.sip.subscribe.SipSubscribe;
import gov.nist.javax.sip.message.SIPRequest;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Service
@RequiredArgsConstructor
public class CatalogService {
    private final SipSender sipSender;
    private final SipSubscribe sipSubscribe;
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    private final DockingService dockingService;

    private final DeviceService deviceService;

    @SneakyThrows
    public CompletableFuture<CatalogResponseDTO> getCatalog(String deviceId){
        CompletableFuture<CatalogResponseDTO> result = new CompletableFuture<>();
        WvpProxyDocking device = dockingService.getDeviceByDeviceCode(deviceId).orElse(null);
        if (device == null){
            throw new JsonException("设备不存在");
        }
        CatalogRequestDTO catalogRequestDTO = new CatalogRequestDTO();
        catalogRequestDTO.setDeviceId(deviceId);
        String sn = String.valueOf((int) (Math.random() * 9 + 1) * 100000);
        catalogRequestDTO.setSn(sn);
        String key = GenericSubscribe.Helper.getKey(CmdType.CATALOG, deviceId);
        sipSubscribe.getCatalogSubscribe().addPublisher(key);
        final ScheduledFuture<?>[] schedule = new ScheduledFuture<?>[1];
        Flow.Subscriber<SIPRequest> subscriber = catalog(key, device, schedule, result);
        // 60秒超时计时器
        schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, 60 , TimeUnit.SECONDS);
        // 添加订阅
        sipSubscribe.getCatalogSubscribe().addSubscribe(key, subscriber);
        sipSender.sendRequest((provider, ip, port)-> SipRequestBuilder.createMessageRequest(device,ip,port,SipRequestBuilder.getCSeq(), XmlUtils.toXml(catalogRequestDTO), SipUtil.generateViaTag(),
                SipUtil.generateFromTag(), provider.getNewCallId()));
        return result;
    }

    private Flow.Subscriber<SIPRequest> catalog(String key, WvpProxyDocking device, ScheduledFuture<?>[] schedule,CompletableFuture<CatalogResponseDTO> result){
        List<CatalogItemDTO> deviceList = new ArrayList<>();
        return new Flow.Subscriber<>() {
            Flow.Subscription subscription;
            CatalogResponseDTO dto;

            final AtomicLong getNum = new AtomicLong(0);

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                log.info("开始订阅 {}", key);
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(SIPRequest item) {
                CatalogResponseDTO responseDTO = XmlUtils.parse(item.getRawContent(), CatalogResponseDTO.class, GB28181Constant.CHARSET);
                dto = responseDTO;
                Long sumNum = responseDTO.getSumNum();
                log.info("{}",responseDTO);
                getNum.getAndAdd(responseDTO.getDeviceList().getDeviceList().size());
                deviceList.addAll(responseDTO.getDeviceList().getDeviceList());
                if(getNum.get() < sumNum){
                    subscription.request(1);
                } else{
                    onComplete();
                }
            }

            @Override
            public void onError(Throwable throwable) {
                onComplete();
            }

            @Override
            public void onComplete() {
                sipSubscribe.getCatalogSubscribe().delPublisher(key);
                schedule[0].cancel(true);
                log.info("订阅结束 {}", key);
                if(dto != null){
                    dto.getDeviceList().setDeviceList(deviceList);
                    scheduledExecutorService.execute(()->{
                        updateProxyDevice(dto);
                    });
                }
                result.complete(dto);
            }
        };
    }

    public void updateProxyDevice(CatalogResponseDTO dto){
        String gbDeviceId = dto.getDeviceId();
        List<WvpProxyDevice> deviceByGbDeviceId = deviceService.getDeviceByGbDeviceId(gbDeviceId);
        List<String> existChannels = deviceByGbDeviceId.stream().map(WvpProxyDevice::getGbDeviceChannelId).toList();
        List<CatalogItemDTO> deviceList = dto.getDeviceList().getDeviceList();
        List<String> catalogChannels = deviceList.stream().map(CatalogItemDTO::getDeviceId).toList();
        List<String> noexist = ListUtils.subtract(catalogChannels, existChannels);
        noexist.forEach(channel->{
            log.info("更新 设备 {}, 通道 {} 信息", gbDeviceId, channel);
            deviceService.autoUpdateDeviceByGbDeviceIdAndChannel(gbDeviceId,channel);
        });
    }
}
