content: package com.genersoft.iot.vmp.jt1078.session;
  
  import com.genersoft.iot.vmp.jt1078.bean.JTRecordDownloadCatch;
  import com.genersoft.iot.vmp.jt1078.event.FtpUploadEvent;
  import com.genersoft.iot.vmp.jt1078.proc.response.J9206;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.context.event.EventListener;
  import org.springframework.scheduling.annotation.Scheduled;
  import org.springframework.stereotype.Component;
  
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.DelayQueue;
  import java.util.concurrent.SynchronousQueue;
  import java.util.concurrent.TimeUnit;
  
  @Slf4j
  @Component
  public class FtpDownloadManager {
  
      private final Map<String, JTRecordDownloadCatch> downloadCatchMap = new ConcurrentHashMap<>();
      private final DelayQueue<JTRecordDownloadCatch> downloadCatchQueue = new DelayQueue<>();
  
      private final Map<String, SynchronousQueue<Object>> topicSubscribers = new ConcurrentHashMap<>();
  
      // 下载过期检查
      @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.SECONDS)
      public void downloadCatchCheck(){
          while (!downloadCatchQueue.isEmpty()) {
              try {
                  JTRecordDownloadCatch take = downloadCatchQueue.take();
                  downloadCatchMap.remove(take.getPath());
              } catch (InterruptedException e) {
                  log.error("[下载过期] ", e);
              }
          }
      }
  
      public void addCatch(String path, String phoneNumber, J9206 j9206) {
          JTRecordDownloadCatch downloadCatch = new JTRecordDownloadCatch();
          downloadCatch.setPhoneNumber(phoneNumber);
          downloadCatch.setPath(path);
          downloadCatch.setJ9206(j9206);
  
          // 10分钟临时地址无法访问则删除
          downloadCatch.setDelayTime(System.currentTimeMillis() + 10 * 60 * 1000L);
  
          downloadCatchMap.put(path, downloadCatch);
          downloadCatchQueue.add(downloadCatch);
      }
  
      public JTRecordDownloadCatch getCatch(String path) {
          return downloadCatchMap.get(path);
      }
  
      @EventListener
      public void onApplicationEvent(FtpUploadEvent event) {
          if (topicSubscribers.isEmpty()) {
              return;
          }
          topicSubscribers.keySet().forEach(key -> {
              if (!event.getFileName().contains(key)) {
                  return;
              }
              SynchronousQueue<Object> synchronousQueue = topicSubscribers.get(key);
              if (synchronousQueue != null) {
                  synchronousQueue.offer(null);
              }
          });
      }
  
  
      public Object runDownload(String path, long timeOut) {
          SynchronousQueue<Object> subscribe = subscribe(path);
          if (subscribe == null) {
              log.error("[JT-下载] 暂停进程失败");
              return null;
          }
          try {
              return subscribe.poll(timeOut, TimeUnit.SECONDS);
          } catch (InterruptedException e) {
              log.warn("[JT-下载] 暂停进程超时", e);
          } finally {
              this.unsubscribe(path);
              JTRecordDownloadCatch downloadCatch = getCatch(path);
              if (downloadCatch != null) {
                  downloadCatchMap.remove(path);
                  downloadCatchQueue.remove(downloadCatch);
              }
          }
          return null;
      }
  
      private SynchronousQueue<Object> subscribe(String key) {
          SynchronousQueue<Object> queue = null;
          if (!topicSubscribers.containsKey(key))
              topicSubscribers.put(key, queue = new SynchronousQueue<>());
          return queue;
      }
  
      private void unsubscribe(String key) {
          topicSubscribers.remove(key);
      }
  
  }
  