MailQueueConsumer.java

Go to the documentation of this file.

/*
 * Turró i Cutiller Foundation. License notice.
 * Copyright (C) 2022 Lluis Turró Cutiller <http://www.turro.org/>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
package org.turro.mail.pool;
 
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import org.turro.action.IElephantCloseable;
import org.turro.annotation.ElephantCloseable;
import org.turro.log.WebLoggers;
import org.turro.mail.message.MailMessage;
 
@ElephantCloseable
public class MailQueueConsumer  implements Runnable, IElephantCloseable {
 
  public void addToQueue(Collection<MailMessage> pool) {
    addToQueue(pool, null);
  }
  
  public void addToQueue(Collection<MailMessage> pool, Consumer onFinish) {
    if(pool == null) return;
    int index = 0, size = pool.size();
    for(MailMessage mail : pool) {
      if(++index == size) {
        addToQueue(mail, onFinish);
      } else {
        addToQueue(mail);
      }
    }
  }
  
  public void addToQueue(MailMessage mail) {
    addToQueue(mail, null);
  }
  
  public void addToQueue(MailMessage mail, Consumer onFinish) {
    try {
      if(!QUEUE.offer(new MailQueueItem(mail, onFinish), 60L, TimeUnit.MINUTES)) {
        WebLoggers.warning(this).message("Too long attempt").log();
      }
    } catch (InterruptedException ex) {
      WebLoggers.severe(this).exception(ex).log();
    }
  }
  
  public int size() {
    return QUEUE.size();
  }
  
  private void addTrigger() {
    QUEUE.add(new MailQueueItem(null, null));
  }
 
  /* Sending */
  
  public void run() {
    try {
      running.set(true);
      while (running.get()) {
        MailQueueItem item = QUEUE.take();
        if(!running.get()) break;
        try {
          item.send();
        } catch(Exception ex) {
          WebLoggers.severe(this).exception(ex).log();
        }
        Thread.sleep(200);
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }
 
  /* Pool */
  
  private static final BlockingQueue<MailQueueItem> QUEUE = new LinkedBlockingQueue<>(500);
  
  /* Sender instance */
  
  private static MailQueueConsumer INSTANCE;
  private static Thread MAIL_QUEUE;
  private static final AtomicBoolean running = new AtomicBoolean(false);
 
  public static MailQueueConsumer getInstance() {
    if(INSTANCE == null) {
      INSTANCE = new MailQueueConsumer();
      MAIL_QUEUE = new Thread(INSTANCE, "ElephantMailQueue");
      MAIL_QUEUE.start();  
    }
    return INSTANCE;
  }
 
  public static void destroy() {
    if(INSTANCE != null) {
      running.set(false);
      getInstance().addTrigger();
      INSTANCE = null;
    }
  }
 
  @Deprecated(forRemoval = false)
  public MailQueueConsumer() {
  }
  
  /* Closeable */
  
  @Override
  public void closeElephant() {
    MailQueueConsumer.destroy();
  }
 
}