/*****************************************************************
    * Licensed to the Apache Software Foundation (ASF) under one   *
    * or more contributor license agreements.  See the NOTICE file *
    * distributed with this work for additional information        *
    * regarding copyright ownership.  The ASF licenses this file   *
    * to you 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 org.apache.james.transport;
  
  import org.apache.avalon.framework.activity.Disposable;
  import org.apache.avalon.framework.activity.Initializable;
  import org.apache.avalon.framework.configuration.Configurable;
  import org.apache.avalon.framework.configuration.Configuration;
  import org.apache.avalon.framework.configuration.ConfigurationException;
  import org.apache.avalon.framework.container.ContainerUtil;
  import org.apache.avalon.framework.logger.AbstractLogEnabled;
  import org.apache.avalon.framework.service.DefaultServiceManager;
  import org.apache.avalon.framework.service.ServiceException;
  import org.apache.avalon.framework.service.ServiceManager;
  import org.apache.avalon.framework.service.Serviceable;
  import org.apache.james.services.MailetLoader;
  import org.apache.james.services.MatcherLoader;
  import org.apache.james.services.SpoolRepository;
  import org.apache.mailet.Mail;
  import org.apache.mailet.Mailet;
  import org.apache.mailet.MailetException;
  import org.apache.mailet.Matcher;
  
  import javax.mail.MessagingException;
  
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.Iterator;
  
  /***
   * Manages the mail spool.  This class is responsible for retrieving
   * messages from the spool, directing messages to the appropriate
   * processor, and removing them from the spool when processing is
   * complete.
   *
   * @version CVS $Revision: 494012 $ $Date: 2007-01-08 10:23:58 +0000 (Mon, 08 Jan 2007) $
   */
  public class JamesSpoolManager
      extends AbstractLogEnabled
      implements Serviceable, Configurable, Initializable, Runnable, Disposable {
  
      /***
       * System component manager
       */
      private DefaultServiceManager compMgr;
  
      /***
       * The configuration object used by this spool manager.
       */
      private Configuration conf;
  
      /***
       * The spool that this manager will process
       */
      private SpoolRepository spool;
  
      /***
       * The map of processor names to processors
       */
      private HashMap processors;
  
      /***
       * The number of threads used to move mail through the spool.
       */
      private int numThreads;
  
      /***
       * The ThreadPool containing worker threads.
       *
       * This used to be used, but for threads that lived the entire
       * lifespan of the application.  Currently commented out.  In
       * the future, we could use a thread pool to run short-lived
       * workers, so that we have a smaller number of readers that
       * accept a message from the spool, and dispatch to a pool of
       * worker threads that process the message.
       */
      // private ThreadPool workerPool;
  
      /***
       * The ThreadManager from which the thread pool is obtained.
       */
      // private ThreadManager threadManager;
 
     /***
      * Number of active threads
      */
     private int numActive;
 
     /***
      * Spool threads are active
      */
     private boolean active;
 
     /***
      * Spool threads
      */
     private Collection spoolThreads;
 
     /***
      * @see org.apache.avalon.framework.service.Serviceable#service(ServiceManager)
      */
     public void service(ServiceManager comp) throws ServiceException {
         // threadManager = (ThreadManager) comp.lookup(ThreadManager.ROLE);
         compMgr = new DefaultServiceManager(comp);
     }
 
     /***
      * @see org.apache.avalon.framework.configuration.Configurable#configure(Configuration)
      */
     public void configure(Configuration conf) throws ConfigurationException {
         this.conf = conf;
         numThreads = conf.getChild("threads").getValueAsInteger(1);
     }
 
     /***
      * @see org.apache.avalon.framework.activity.Initializable#initialize()
      */
     public void initialize() throws Exception {
 
         getLogger().info("JamesSpoolManager init...");
         spool = (SpoolRepository) compMgr.lookup(SpoolRepository.ROLE);
 
         MailetLoader mailetLoader
         = (MailetLoader) compMgr.lookup(MailetLoader.ROLE);
         MatcherLoader matchLoader
         = (MatcherLoader) compMgr.lookup(MatcherLoader.ROLE);
 
         //A processor is a Collection of
         processors = new HashMap();
 
         final Configuration[] processorConfs = conf.getChildren( "processor" );
         for ( int i = 0; i < processorConfs.length; i++ )
         {
             Configuration processorConf = processorConfs[i];
             String processorName = processorConf.getAttribute("name");
             try {
                 LinearProcessor processor = new LinearProcessor();
                 setupLogger(processor, processorName);
                 processor.setSpool(spool);
                 processor.initialize();
                 processors.put(processorName, processor);
 
                 final Configuration[] mailetConfs
                     = processorConf.getChildren( "mailet" );
                 // Loop through the mailet configuration, load
                 // all of the matcher and mailets, and add
                 // them to the processor.
                 for ( int j = 0; j < mailetConfs.length; j++ )
                 {
                     Configuration c = mailetConfs[j];
                     String mailetClassName = c.getAttribute("class");
                     String matcherName = c.getAttribute("match");
                     Mailet mailet = null;
                     Matcher matcher = null;
                     try {
                         matcher = matchLoader.getMatcher(matcherName);
                         //The matcher itself should log that it's been inited.
                         if (getLogger().isInfoEnabled()) {
                             StringBuffer infoBuffer =
                                 new StringBuffer(64)
                                         .append("Matcher ")
                                         .append(matcherName)
                                         .append(" instantiated.");
                             getLogger().info(infoBuffer.toString());
                         }
                     } catch (MessagingException ex) {
                         // **** Do better job printing out exception
                         if (getLogger().isErrorEnabled()) {
                             StringBuffer errorBuffer =
                                 new StringBuffer(256)
                                         .append("Unable to init matcher ")
                                         .append(matcherName)
                                         .append(": ")
                                         .append(ex.toString());
                             getLogger().error( errorBuffer.toString(), ex );
                 if (ex.getNextException() != null) {
                 getLogger().error( "Caused by nested exception: ", ex.getNextException());
                 }
                         }
                         System.err.println("Unable to init matcher " + matcherName);
                         System.err.println("Check spool manager logs for more details.");
                         //System.exit(1);
                         throw ex;
                     }
                     try {
                         mailet = mailetLoader.getMailet(mailetClassName, c);
                         if (getLogger().isInfoEnabled()) {
                             StringBuffer infoBuffer =
                                 new StringBuffer(64)
                                         .append("Mailet ")
                                         .append(mailetClassName)
                                         .append(" instantiated.");
                             getLogger().info(infoBuffer.toString());
                         }
                     } catch (MessagingException ex) {
                         // **** Do better job printing out exception
                         if (getLogger().isErrorEnabled()) {
                             StringBuffer errorBuffer =
                                 new StringBuffer(256)
                                         .append("Unable to init mailet ")
                                         .append(mailetClassName)
                                         .append(": ")
                                         .append(ex.toString());
                             getLogger().error( errorBuffer.toString(), ex );
                 if (ex.getNextException() != null) {
                 getLogger().error( "Caused by nested exception: ", ex.getNextException());
                 }
                         }
                         System.err.println("Unable to init mailet " + mailetClassName);
                         System.err.println("Check spool manager logs for more details.");
                         //System.exit(1);
                         throw ex;
                     }
                     //Add this pair to the processor
                     processor.add(matcher, mailet);
                 }
 
                 // Close the processor matcher/mailet lists.
                 //
                 // Please note that this is critical to the proper operation
                 // of the LinearProcessor code.  The processor will not be
                 // able to service mails until this call is made.
                 processor.closeProcessorLists();
 
                 if (getLogger().isInfoEnabled()) {
                     StringBuffer infoBuffer =
                         new StringBuffer(64)
                                 .append("Processor ")
                                 .append(processorName)
                                 .append(" instantiated.");
                     getLogger().info(infoBuffer.toString());
                 }
             } catch (Exception ex) {
                 if (getLogger().isErrorEnabled()) {
                     StringBuffer errorBuffer =
                        new StringBuffer(256)
                                .append("Unable to init processor ")
                                .append(processorName)
                                .append(": ")
                                .append(ex.toString());
                     getLogger().error( errorBuffer.toString(), ex );
                 }
                 throw ex;
             }
         }
         if (getLogger().isInfoEnabled()) {
             StringBuffer infoBuffer =
                 new StringBuffer(64)
                     .append("Spooler Manager uses ")
                     .append(numThreads)
                     .append(" Thread(s)");
             getLogger().info(infoBuffer.toString());
         }
 
         active = true;
         numActive = 0;
         spoolThreads = new java.util.ArrayList(numThreads);
         for ( int i = 0 ; i < numThreads ; i++ ) {
             Thread reader = new Thread(this, "Spool Thread #" + i);
             spoolThreads.add(reader);
             reader.start();
         }
     }
 
     /***
      * This routinely checks the message spool for messages, and processes
      * them as necessary
      */
     public void run() {
 
         if (getLogger().isInfoEnabled())
         {
             getLogger().info("Run JamesSpoolManager: "
                              + Thread.currentThread().getName());
             getLogger().info("Spool=" + spool.getClass().getName());
         }
 
         numActive++;
         while(active) {
             String key = null;
             try {
                 Mail mail = (Mail)spool.accept();
                 key = mail.getName();
                 if (getLogger().isDebugEnabled()) {
                     StringBuffer debugBuffer =
                         new StringBuffer(64)
                                 .append("==== Begin processing mail ")
                                 .append(mail.getName())
                                 .append("====");
                     getLogger().debug(debugBuffer.toString());
                 }
                 process(mail);
                 // Only remove an email from the spool is processing is
                 // complete, or if it has no recipients
                 if ((Mail.GHOST.equals(mail.getState())) ||
                     (mail.getRecipients() == null) ||
                     (mail.getRecipients().size() == 0)) {
                     ContainerUtil.dispose(mail);
                     spool.remove(key);
                     if (getLogger().isDebugEnabled()) {
                         StringBuffer debugBuffer =
                             new StringBuffer(64)
                                     .append("==== Removed from spool mail ")
                                     .append(key)
                                     .append("====");
                         getLogger().debug(debugBuffer.toString());
                     }
                 }
                 else {
                     // spool.remove() has a side-effect!  It unlocks the
                     // message so that other threads can work on it!  If
                     // we don't remove it, we must unlock it!
                     spool.store(mail);
                     ContainerUtil.dispose(mail);
                     spool.unlock(key);
                     // Do not notify: we simply updated the current mail
                     // and we are able to reprocess it now.
                 }
                 mail = null;
             } catch (InterruptedException ie) {
                 getLogger().info("Interrupted JamesSpoolManager: " + Thread.currentThread().getName());
             } catch (Throwable e) {
                 if (getLogger().isErrorEnabled()) {
                     getLogger().error("Exception processing " + key + " in JamesSpoolManager.run "
                                       + e.getMessage(), e);
                 }
                 /* Move the mail to ERROR state?  If we do, it could be
                  * deleted if an error occurs in the ERROR processor.
                  * Perhaps the answer is to resolve that issue by
                  * having a special state for messages that are not to
                  * be processed, but aren't to be deleted?  The message
                  * would already be in the spool, but would not be
                  * touched again.
                 if (mail != null) {
                     try {
                         mail.setState(Mail.ERROR);
                         spool.store(mail);
                     }
                 }
                 */
             }
         }
         if (getLogger().isInfoEnabled())
         {
             getLogger().info("Stop JamesSpoolManager: " + Thread.currentThread().getName());
         }
         numActive--;
     }
 
     /***
      * Process this mail message by the appropriate processor as designated
      * in the state of the Mail object.
      *
      * @param mail the mail message to be processed
      */
     protected void process(Mail mail) {
         while (true) {
             String processorName = mail.getState();
             if (processorName.equals(Mail.GHOST)) {
                 //This message should disappear
                 return;
             }
             try {
                 LinearProcessor processor
                     = (LinearProcessor)processors.get(processorName);
                 if (processor == null) {
                     StringBuffer exceptionMessageBuffer =
                         new StringBuffer(128)
                             .append("Unable to find processor ")
                             .append(processorName)
                             .append(" requested for processing of ")
                             .append(mail.getName());
                     String exceptionMessage = exceptionMessageBuffer.toString();
                     getLogger().debug(exceptionMessage);
                     mail.setState(Mail.ERROR);
                     throw new MailetException(exceptionMessage);
                 }
                 StringBuffer logMessageBuffer = null;
                 if (getLogger().isDebugEnabled()) {
                     logMessageBuffer =
                         new StringBuffer(64)
                                 .append("Processing ")
                                 .append(mail.getName())
                                 .append(" through ")
                                 .append(processorName);
                     getLogger().debug(logMessageBuffer.toString());
                 }
                 processor.service(mail);
                 if (getLogger().isDebugEnabled()) {
                     logMessageBuffer =
                         new StringBuffer(128)
                                 .append("Processed ")
                                 .append(mail.getName())
                                 .append(" through ")
                                 .append(processorName);
                     getLogger().debug(logMessageBuffer.toString());
                     getLogger().debug("Result was " + mail.getState());
                 }
                 return;
             } catch (Throwable e) {
                 // This is a strange error situation that shouldn't ordinarily
                 // happen
                 StringBuffer exceptionBuffer = 
                     new StringBuffer(64)
                             .append("Exception in processor <")
                             .append(processorName)
                             .append(">");
                 getLogger().error(exceptionBuffer.toString(), e);
                 if (processorName.equals(Mail.ERROR)) {
                     // We got an error on the error processor...
                     // kill the message
                     mail.setState(Mail.GHOST);
                     mail.setErrorMessage(e.getMessage());
                 } else {
                     //We got an error... send it to the requested processor
                     if (!(e instanceof MessagingException)) {
                         //We got an error... send it to the error processor
                         mail.setState(Mail.ERROR);
                     }
                     mail.setErrorMessage(e.getMessage());
                 }
             }
             if (getLogger().isErrorEnabled()) {
                 StringBuffer logMessageBuffer =
                     new StringBuffer(128)
                             .append("An error occurred processing ")
                             .append(mail.getName())
                             .append(" through ")
                             .append(processorName);
                 getLogger().error(logMessageBuffer.toString());
                 getLogger().error("Result was " + mail.getState());
             }
         }
     }
 
     /***
      * The dispose operation is called at the end of a components lifecycle.
      * Instances of this class use this method to release and destroy any
      * resources that they own.
      *
      * This implementation shuts down the LinearProcessors managed by this
      * JamesSpoolManager
      *
      * @throws Exception if an error is encountered during shutdown
      */
     public void dispose() {
         getLogger().info("JamesSpoolManager dispose...");
         active = false; // shutdown the threads
         for (Iterator it = spoolThreads.iterator(); it.hasNext(); ) {
             ((Thread) it.next()).interrupt(); // interrupt any waiting accept() calls.
         }
 
         long stop = System.currentTimeMillis() + 60000;
         // give the spooler threads one minute to terminate gracefully
         while (numActive != 0 && stop > System.currentTimeMillis()) {
             try {
                 Thread.sleep(1000);
             } catch (Exception ignored) {}
         }
         getLogger().info("JamesSpoolManager thread shutdown completed.");
 
         Iterator it = processors.keySet().iterator();
         while (it.hasNext()) {
             String processorName = (String)it.next();
             if (getLogger().isDebugEnabled()) {
                 getLogger().debug("Processor " + processorName);
             }
             LinearProcessor processor = (LinearProcessor)processors.get(processorName);
             processor.dispose();
             processors.remove(processor);
         }
     }
 
 }