package com.atlassian.core.task;
   
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   
   import javax.mail.MessagingException;
   import java.io.Serializable;
   import java.util.ArrayList;
   import java.util.List;
  import java.util.concurrent.atomic.AtomicInteger;
  
  public class AbstractErrorQueuedTaskQueue extends AbstractTaskQueue implements TaskQueueWithErrorQueue {
      private static final transient Logger log = LoggerFactory.getLogger(AbstractErrorQueuedTaskQueue.class);
  
      private final TaskQueue errorQueue;
  
      private int retryCount = 5;
  
      private List<Task> failed;
  
      public AbstractErrorQueuedTaskQueue(TaskQueue errorQueue, FifoBuffer<Task> buffer) {
          super(buffer);
          this.errorQueue = errorQueue;
      }
  
      public void flush() {
          failed = new ArrayList<Task>();
          super.flush();
          for (Task task : failed) {
              addTask(task);
          }
      }
  
      protected void handleException(Task task, Exception rootException) {
          TaskDecorator theTask = (TaskDecorator) task;
  
          if (theTask.getExecutionCount() > retryCount) {
  
              errorQueue.addTask(theTask.getTask());
          } else {
  
              failed.add(task);
          }
          if (rootException instanceof MessagingException) {
              Exception e = rootException;
              while (e instanceof MessagingException) {
                  MessagingException me = (MessagingException) e;
                  log.error(me.getMessage(), me);
                  e = me.getNextException();
              }
          } else
              log.error("Failed to execute task", rootException);
      }
  
      public void addTask(Task task) {
          if (task instanceof TaskDecorator) {
              super.addTask(task);
          } else {
              super.addTask(new TaskDecorator(task));
          }
      }
  
      public TaskQueue getErrorQueue() {
          return errorQueue;
      }
  
      public int getRetryCount() {
          return retryCount;
      }
  
      public void setRetryCount(int retryCount) {
          this.retryCount = retryCount;
      }
  
      public static class TaskDecorator implements Task, Serializable {
          private final Task task;
          private final AtomicInteger executionCount = new AtomicInteger();
  
          public TaskDecorator(Task task) {
              this.task = task;
          }
  
          public void execute() throws Exception {
              executionCount.incrementAndGet();
              task.execute();
          }
  
          public int getExecutionCount() {
              return executionCount.get();
          }
  
          public Task getTask() {
              return task;
          }
      }
  }