<commit_before>/*
 * Copyright 2013 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package de.perdian.apps.downloader.core;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * An engine represents the central manager object, into which new jobs that are
 * supposed to download information from an external resource will be added and
 * from which information about the current state of the engine and it's jobs
 * can be requested.
 *
 * Client code should only change the state of the download queue and associated
 * processes using the methods provided in this API class.
 *
 * @author <NAME>
 */

public class DownloadEngine {

  private static final Logger log = LogManager.getLogger(DownloadEngine.class);

  private ExecutorService myExecutorService = Executors.newCachedThreadPool();
  private List<DownloadListener> myListeners = new CopyOnWriteArrayList<>();
  private Path myTargetDirectory = null;
  private int myBufferSize = 1024 * 4; // 4 KiB
  private int myNotificationSize = 1024 * 16; // 16 KiB
  private int myProcessorCount = 1;
  private Queue<DownloadJob> myWaitingJobs = new PriorityQueue<>(10, new DownloadJob.PriorityComparator());
  private List<DownloadJob> myActiveJobs = new ArrayList<>();

  /**
   * Create a new engine
   *
   * @param targetDirectory
   *   the directory into which the downloads will be written
   */
  public DownloadEngine(Path targetDirectory) {
    this.setTargetDirectory(Objects.requireNonNull(targetDirectory, "Parameter 'targetDirectory' must not be null!"));
  }

  /**
   * Submits a new download into this engine. It is up to the engine to decide
   * whether or not the job can be started immediately or if it needs to be put
   * in any kind of queue and await a free download slot.
   *
   * @param request
   *   the request containing all the information from which a download can be
   *   constructuted.
   * @return
   *   a {@link DownloadJob} if the engine has accepted the request and
   *   scheduled it for execution or {@code null} if the engine instance
   *   rejected the job and will not execute the transfer process.
   */
  public DownloadJob submit(DownloadRequest request) {
    if(request == null) {
      throw new NullPointerException("Parameter 'request' must not be null!");
    } else if(request.getTargetFileName() == null) {
      throw new NullPointerException("Property 'targetFileName' of request must not be null!");
    } else if(request.getContentFactory() == null) {
      throw new NullPointerException("Property 'contentFactory' of request must not be null!");
    } else {

      // First we contact all the validators and make sure the new request might
      // actually be processed
      if(!this.fireRequestSubmitted(request)) {
        return null;
      }

      log.info("Accepted request: {}", request);
      DownloadJob downloadJob = new DownloadJob(this);
      downloadJob.setRequest(request);
      downloadJob.setScheduleTime(System.currentTimeMillis());
      downloadJob.setStatus(DownloadStatus.SCHEDULED);
      synchronized(this) {
        if(!this.startJob(downloadJob, false)) {
          this.getWaitingJobs().add(downloadJob);
          this.fireJobScheduled(downloadJob);
        }
      }
      return downloadJob;

    }
  }

  /**
   * List all currently active jobs, which means the jobs for which processor
   * thread is currently transfering data from the source to a target.
   *
   * @return
   *   the list of active jobs at the time this method was called. Subsequent
   *   changes in the engines state (a download was completed or cancelled)
   *   will not be reflected into this result list. The list itself will
   *   therefore be immutable
   */
  public List<DownloadJob> listActiveJobs() {
    return Collections.unmodifiableList(new ArrayList<>(this.getActiveJobs()));
  }

  /**
   * List all currently waitinbg jobs, which means the jobs for which no
   * processor has been assigned yet. The job itself is therefore sitting in a
   * queue waiting to be picked up and to be executed.
   *
   * @return
   *   the list of waiting jobs at the time this method was called. Subsequent
   *   changes in the engines state (a download was picked up for execution
   *   or was cancelled) will not be reflected into this result list. The list
   *   itself will therefore be immutable
   */
  public List<DownloadJob> listWaitingJobs() {
    return Collections.unmodifiableList(new ArrayList<>(this.getWaitingJobs()));
  }

  /**
   * Removes all currently queued jobs
   *
   * @return
   *   the jobs that were cleared from this engine
   */
  public synchronized List<DownloadJob> clearWaitingJobs() {
    List<DownloadJob> resultList = new ArrayList<>(this.getWaitingJobs());
    this.getWaitingJobs().clear();
    return Collections.unmodifiableList(resultList);
  }

  /**
   * Wait until all jobs currently executing and waiting inside this executor
   * have been completed
   */
  public void waitUntilAllDownloadsComplete() {
    synchronized(this) {
      if(!this.isBusy()) {
        return;
      }
    }
    try {
      final CountDownLatch latch = new CountDownLatch(1);
      this.addListener(new DownloadListenerSkeleton() {
        @Override public void onJobCompleted(DownloadJob job) {
          synchronized(DownloadEngine.this) {
            if(!DownloadEngine.this.isBusy()) {
              DownloadEngine.this.removeListener(this);
              latch.countDown();
            }
          }
        }
      });
      latch.await();
    } catch(InterruptedException e) {
      // Ignore here
    }
  }

  // ---------------------------------------------------------------------------
  // --- Status transitions ----------------------------------------------------
  // ---------------------------------------------------------------------------

  synchronized boolean startJob(final DownloadJob job, boolean ignoreSlots) {
    if(ignoreSlots || this.getProcessorCount() > this.getActiveJobs().size()) {

      // Make sure we remove the job from the waiting list, so no matter from
      // where we come from we always leave in a consistent state
      this.getWaitingJobs().remove(job);

      // Now make sure we have the right satus upon the job itself
      job.setStartTime(System.currentTimeMillis());
      job.setStatus(DownloadStatus.ACTIVE);
      DownloadEngine.this.getActiveJobs().add(job);

      this.getExecutorService().submit(new Runnable() {
        @Override public void run() {
          DownloadEngine.this.runJob(job);
        }
        @Override public String toString() {
          return DownloadJob.class.getSimpleName() + "-Processor[" + job + "]";
        }
      });
      return true;

    } else {
      return false;
    }
  }

  synchronized boolean cancelJob(DownloadJob job, String reason) {
    if(DownloadStatus.CANCELLED.equals(job.getStatus())) {

      // The download is already cancelled, so there must be someone else
      // calling the cancel method twice.
      return true;

    } else if(!this.getActiveJobs().contains(job) && !this.getWaitingJobs().remove(job)) {

      // The job could not be found in the list of active jobs, which means we
      // have no way of handling him at all - so we just exit
      return false;

    } else {

      log.debug("Cancelling job {} with reason: {}", job, reason);
      job.setStatus(DownloadStatus.CANCELLED);
      job.setCancelTime(System.currentTimeMillis());
      job.setCancelReason(reason);
      this.fireJobCancelled(job);
      this.checkWaitingJobs();
      return true;

    }
  }

  synchronized void checkWaitingJobs() {
    Queue<DownloadJob> queue = this.getWaitingJobs();
    int maxJobsToRemove = this.getProcessorCount() - this.getActiveJobs().size();
    if(maxJobsToRemove > 0) {
      List<DownloadJob> removedJobs = new ArrayList<>(maxJobsToRemove);
      for(int i=0; i < maxJobsToRemove && !queue.isEmpty(); i++) {
        removedJobs.add(queue.remove());
      }
      for(DownloadJob job : removedJobs) {
        this.startJob(job, false);
      }
    }
  }

  // ---------------------------------------------------------------------------
  // --- Job execution ---------------------------------------------------------
  // ---------------------------------------------------------------------------

  void runJob(DownloadJob job) {
    try {
      log.info("Running job: {}", job);
      this.runJobTransfer(job);
      job.setEndTime(System.currentTimeMillis());
      job.setStatus(DownloadStatus.COMPLETED);
      log.info("Job completed: {} in {} ms", job, (job.getEndTime() - job.getStartTime()));<commit_msg>Log level<commit_after>/*
 * Copyright 2013 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package de.perdian.apps.downloader.core;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * An engine represents the central manager object, into which new jobs that are
 * supposed to download information from an external resource will be added and
 * from which information about the current state of the engine and it's jobs
 * can be requested.
 *
 * Client code should only change the state of the download queue and associated
 * processes using the methods provided in this API class.
 *
 * @author <NAME>
 */

public class DownloadEngine {

  private static final Logger log = LogManager.getLogger(DownloadEngine.class);

  private ExecutorService myExecutorService = Executors.newCachedThreadPool();
  private List<DownloadListener> myListeners = new CopyOnWriteArrayList<>();
  private Path myTargetDirectory = null;
  private int myBufferSize = 1024 * 4; // 4 KiB
  private int myNotificationSize = 1024 * 16; // 16 KiB
  private int myProcessorCount = 1;
  private Queue<DownloadJob> myWaitingJobs = new PriorityQueue<>(10, new DownloadJob.PriorityComparator());
  private List<DownloadJob> myActiveJobs = new ArrayList<>();

  /**
   * Create a new engine
   *
   * @param targetDirectory
   *   the directory into which the downloads will be written
   */
  public DownloadEngine(Path targetDirectory) {
    this.setTargetDirectory(Objects.requireNonNull(targetDirectory, "Parameter 'targetDirectory' must not be null!"));
  }

  /**
   * Submits a new download into this engine. It is up to the engine to decide
   * whether or not the job can be started immediately or if it needs to be put
   * in any kind of queue and await a free download slot.
   *
   * @param request
   *   the request containing all the information from which a download can be
   *   constructuted.
   * @return
   *   a {@link DownloadJob} if the engine has accepted the request and
   *   scheduled it for execution or {@code null} if the engine instance
   *   rejected the job and will not execute the transfer process.
   */
  public DownloadJob submit(DownloadRequest request) {
    if(request == null) {
      throw new NullPointerException("Parameter 'request' must not be null!");
    } else if(request.getTargetFileName() == null) {
      throw new NullPointerException("Property 'targetFileName' of request must not be null!");
    } else if(request.getContentFactory() == null) {
      throw new NullPointerException("Property 'contentFactory' of request must not be null!");
    } else {

      // First we contact all the validators and make sure the new request might
      // actually be processed
      if(!this.fireRequestSubmitted(request)) {
        return null;
      }

      log.info("Accepted request: {}", request);
      DownloadJob downloadJob = new DownloadJob(this);
      downloadJob.setRequest(request);
      downloadJob.setScheduleTime(System.currentTimeMillis());
      downloadJob.setStatus(DownloadStatus.SCHEDULED);
      synchronized(this) {
        if(!this.startJob(downloadJob, false)) {
          this.getWaitingJobs().add(downloadJob);
          this.fireJobScheduled(downloadJob);
        }
      }
      return downloadJob;

    }
  }

  /**
   * List all currently active jobs, which means the jobs for which processor
   * thread is currently transfering data from the source to a target.
   *
   * @return
   *   the list of active jobs at the time this method was called. Subsequent
   *   changes in the engines state (a download was completed or cancelled)
   *   will not be reflected into this result list. The list itself will
   *   therefore be immutable
   */
  public List<DownloadJob> listActiveJobs() {
    return Collections.unmodifiableList(new ArrayList<>(this.getActiveJobs()));
  }

  /**
   * List all currently waitinbg jobs, which means the jobs for which no
   * processor has been assigned yet. The job itself is therefore sitting in a
   * queue waiting to be picked up and to be executed.
   *
   * @return
   *   the list of waiting jobs at the time this method was called. Subsequent
   *   changes in the engines state (a download was picked up for execution
   *   or was cancelled) will not be reflected into this result list. The list
   *   itself will therefore be immutable
   */
  public List<DownloadJob> listWaitingJobs() {
    return Collections.unmodifiableList(new ArrayList<>(this.getWaitingJobs()));
  }

  /**
   * Removes all currently queued jobs
   *
   * @return
   *   the jobs that were cleared from this engine
   */
  public synchronized List<DownloadJob> clearWaitingJobs() {
    List<DownloadJob> resultList = new ArrayList<>(this.getWaitingJobs());
    this.getWaitingJobs().clear();
    return Collections.unmodifiableList(resultList);
  }

  /**
   * Wait until all jobs currently executing and waiting inside this executor
   * have been completed
   */
  public void waitUntilAllDownloadsComplete() {
    synchronized(this) {
      if(!this.isBusy()) {
        return;
      }
    }
    try {
      final CountDownLatch latch = new CountDownLatch(1);
      this.addListener(new DownloadListenerSkeleton() {
        @Override public void onJobCompleted(DownloadJob job) {
          synchronized(DownloadEngine.this) {
            if(!DownloadEngine.this.isBusy()) {
              DownloadEngine.this.removeListener(this);
              latch.countDown();
            }
          }
        }
      });
      latch.await();
    } catch(InterruptedException e) {
      // Ignore here
    }
  }

  // ---------------------------------------------------------------------------
  // --- Status transitions ----------------------------------------------------
  // ---------------------------------------------------------------------------

  synchronized boolean startJob(final DownloadJob job, boolean ignoreSlots) {
    if(ignoreSlots || this.getProcessorCount() > this.getActiveJobs().size()) {

      // Make sure we remove the job from the waiting list, so no matter from
      // where we come from we always leave in a consistent state
      this.getWaitingJobs().remove(job);

      // Now make sure we have the right satus upon the job itself
      job.setStartTime(System.currentTimeMillis());
      job.setStatus(DownloadStatus.ACTIVE);
      DownloadEngine.this.getActiveJobs().add(job);

      this.getExecutorService().submit(new Runnable() {
        @Override public void run() {
          DownloadEngine.this.runJob(job);
        }
        @Override public String toString() {
          return DownloadJob.class.getSimpleName() + "-Processor[" + job + "]";
        }
      });
      return true;

    } else {
      return false;
    }
  }

  synchronized boolean cancelJob(DownloadJob job, String reason) {
    if(DownloadStatus.CANCELLED.equals(job.getStatus())) {

      // The download is already cancelled, so there must be someone else
      // calling the cancel method twice.
      return true;

    } else if(!this.getActiveJobs().contains(job) && !this.getWaitingJobs().remove(job)) {

      // The job could not be found in the list of active jobs, which means we
      // have no way of handling him at all - so we just exit
      return false;

    } else {

      log.debug("Cancelling job {} with reason: {}", job, reason);
      job.setStatus(DownloadStatus.CANCELLED);
      job.setCancelTime(System.currentTimeMillis());
      job.setCancelReason(reason);
      this.fireJobCancelled(job);
      this.checkWaitingJobs();
      return true;

    }
  }

  synchronized void checkWaitingJobs() {
    Queue<DownloadJob> queue = this.getWaitingJobs();
    int maxJobsToRemove = this.getProcessorCount() - this.getActiveJobs().size();
    if(maxJobsToRemove > 0) {
      List<DownloadJob> removedJobs = new ArrayList<>(maxJobsToRemove);
      for(int i=0; i < maxJobsToRemove && !queue.isEmpty(); i++) {
        removedJobs.add(queue.remove());
      }
      for(DownloadJob job : removedJobs) {
        this.startJob(job, false);
      }
    }
  }

  // ---------------------------------------------------------------------------
  // --- Job execution ---------------------------------------------------------
  // ---------------------------------------------------------------------------

  void runJob(DownloadJob job) {
    try {
      log.debug("Running job: {}", job);
      this.runJobTransfer(job);
      job.setEndTime(System.currentTimeMillis());
      job.setStatus(DownloadStatus.COMPLETED);
      log.info("Job completed: {} in {} ms", job, (job.getEndTime() - job.getStartTime()));
{'fraction_non_alphanumeric': 0.13369783875793056, 'fraction_numerical': 0.0030948573786558, 'mean_word_length': 2.669190007570023, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 108, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6964357', 'n_tokens_mistral': 5112, 'n_tokens_neox': 5055, 'n_words': 1682}