content: package org.araymond.joal.core.ttorrent.client.announcer.request;
  
  import com.google.common.util.concurrent.ThreadFactoryBuilder;
  import lombok.Getter;
  import lombok.RequiredArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.araymond.joal.core.torrent.torrent.InfoHash;
  import org.araymond.joal.core.ttorrent.client.announcer.Announcer;
  import org.araymond.joal.core.ttorrent.client.announcer.exceptions.TooManyAnnouncesFailedInARowException;
  import org.araymond.joal.core.ttorrent.client.announcer.response.AnnounceResponseCallback;
  
  import java.util.*;
  import java.util.concurrent.*;
  
  import static java.util.Optional.ofNullable;
  import static java.util.stream.Collectors.toSet;
  
  @Slf4j
  public class AnnouncerExecutor {
  
      private final AnnounceResponseCallback announceResponseCallback;
      private final ThreadPoolExecutor executorService;
      private final Map<InfoHash, AnnouncerWithFuture> currentlyRunning;
  
      public AnnouncerExecutor(final AnnounceResponseCallback announceResponseCallback) {
          this.announceResponseCallback = announceResponseCallback;
          // From javadoc :
          //   Unbounded queues. Using an unbounded queue (for example a LinkedBlockingQueue without a predefined capacity) will cause new tasks to wait in
          //   the queue when all corePoolSize threads are busy. Thus, no more than corePoolSize threads will ever be created. (And the value of the
          //   maximumPoolSize therefore doesn't have any effect.) This may be appropriate when each task is completely independent of others, so tasks
          //   cannot affect each others execution
          final int corePoolSize = 3;
          final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("annnouncer-%d").build();
          this.executorService = new ThreadPoolExecutor(corePoolSize, 3, 40, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), threadFactory);
          this.currentlyRunning = new ConcurrentHashMap<>();
      }
  
      public void execute(final AnnounceRequest request) {
          final Runnable task = () -> {
              try {
                  announceResponseCallback.onAnnounceWillAnnounce(request.getEvent(), request.getAnnouncer());
                  final SuccessAnnounceResponse result = request.getAnnouncer().announce(request.getEvent());
                  announceResponseCallback.onAnnounceSuccess(request.getEvent(), request.getAnnouncer(), result);
              } catch (final TooManyAnnouncesFailedInARowException e) {
                  announceResponseCallback.onTooManyAnnounceFailedInARow(request.getEvent(), request.getAnnouncer(), e);
              } catch (final Throwable throwable) {
                  announceResponseCallback.onAnnounceFailure(request.getEvent(), request.getAnnouncer(), throwable);
              } finally {
                  this.currentlyRunning.remove(request.getAnnouncer().getTorrentInfoHash());
              }
          };
  
          final Future<?> taskFuture = this.executorService.submit(task);
          this.currentlyRunning.put(
                  request.getAnnouncer().getTorrentInfoHash(),
                  new AnnouncerWithFuture(request.getAnnouncer(), taskFuture)
          );
      }
  
      public Optional<Announcer> deny(final InfoHash infoHash) {
          return ofNullable(this.currentlyRunning.remove(infoHash)).map(announcerFuture -> {
              announcerFuture.getFuture().cancel(true);
              return announcerFuture.getAnnouncer();
          });
      }
  
      public Set<Announcer> denyAll() {
          return new HashSet<>(this.currentlyRunning.keySet()).stream()
                  .map(this::deny)
                  .flatMap(Optional::stream)
                  .collect(toSet());
      }
  
      public void awaitForRunningTasks() {
          this.executorService.shutdown();
          try {
              if (!this.executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                  log.warn("AnnouncerExecutor timed out after 10s");
              }
          } catch (final InterruptedException e) {
              log.warn("AnnouncerExecutor interrupt", e);
          }
      }
  
      @RequiredArgsConstructor
      @Getter
      private static final class AnnouncerWithFuture {
          private final Announcer announcer;
          private final Future<?> future;
      }
  }
  