/* * Copyright (C) The Apache Software Foundation. All rights reserved. * * This software is published under the terms of the Apache Software License * version 1.1, a copy of which has been included with this distribution in * the LICENSE file. */ package org.apache.james.transport; import org.apache.avalon.cornerstone.services.threads.ThreadManager; import org.apache.avalon.excalibur.thread.ThreadPool; import org.apache.avalon.framework.activity.Disposable; import org.apache.avalon.framework.activity.Initializable; import org.apache.avalon.framework.component.ComponentException; import org.apache.avalon.framework.component.ComponentManager; import org.apache.avalon.framework.component.Composable; import org.apache.avalon.framework.component.DefaultComponentManager; import org.apache.avalon.framework.component.Component; 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.context.Context; import org.apache.avalon.framework.context.Contextualizable; import org.apache.avalon.framework.context.DefaultContext; import org.apache.avalon.framework.logger.AbstractLogEnabled; import org.apache.james.core.MailImpl; import org.apache.james.services.MailStore; import org.apache.james.services.SpoolRepository; import org.apache.mailet.*; import javax.mail.MessagingException; import java.util.HashMap; import java.util.Iterator; /** * @author Serge Knystautas * @author Federico Barbieri * * This is $Revision: 1.10 $ * Committed on $Date: 2002/06/03 16:07:07 $ by: $Author: hammant $ */ public class JamesSpoolManager extends AbstractLogEnabled implements Contextualizable, Composable, Configurable, Initializable, Runnable, Disposable, Component { private final static boolean DEEP_DEBUG = true; private DefaultComponentManager compMgr; //using implementation as we need put method. private Configuration conf; private Context context; private SpoolRepository spool; private MailetContext mailetcontext; private HashMap processors; private int threads; private ThreadPool workerPool; private ThreadManager threadManager; public void configure(Configuration conf) throws ConfigurationException { this.conf = conf; threads = conf.getChild("threads").getValueAsInteger(1); } public void contextualize(Context context) { this.context = new DefaultContext( context ); } public void compose(ComponentManager comp) throws ComponentException { threadManager = (ThreadManager)comp.lookup( ThreadManager.ROLE ); compMgr = new DefaultComponentManager(comp); } public void initialize() throws Exception { getLogger().info("JamesSpoolManager init..."); workerPool = threadManager.getThreadPool( "default" ); MailStore mailstore = (MailStore) compMgr.lookup("org.apache.james.services.MailStore"); spool = mailstore.getInboundSpool(); if (null == spool) { String exceptionMessage = "The mailstore's inbound spool is null. The mailstore is misconfigured"; if (getLogger().isErrorEnabled()) { getLogger().error( exceptionMessage ); } throw new ConfigurationException(exceptionMessage); } if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) { getLogger().debug("Got spool"); } mailetcontext = (MailetContext) compMgr.lookup("org.apache.mailet.MailetContext"); MailetLoader mailetLoader = new MailetLoader(); MatchLoader matchLoader = new MatchLoader(); try { mailetLoader.configure(conf.getChild("mailetpackages")); matchLoader.configure(conf.getChild("matcherpackages")); compMgr.put(Resources.MAILET_LOADER, mailetLoader); compMgr.put(Resources.MATCH_LOADER, matchLoader); } catch (ConfigurationException ce) { final String message = "Unable to configure mailet/matcher Loaders: " + ce.getMessage(); if (getLogger().isErrorEnabled()) { getLogger().error( message, ce ); } throw new RuntimeException( message ); } //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); // If this is the root processor, add the PostmasterAlias // mailet silently to the top if (processorName.equals("root")) { Matcher matcher = matchLoader.getMatcher("All", mailetcontext); Mailet mailet = mailetLoader.getMailet("PostmasterAlias", mailetcontext, null); processor.add(matcher, mailet); } 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, mailetcontext); //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 ); } System.err.println("Unable to init mailet " + matcherName); System.err.println("Check spool manager logs for more details."); ex.printStackTrace(); //System.exit(1); throw ex; } try { mailet = mailetLoader.getMailet(mailetClassName, mailetcontext, 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 ); } System.err.println("Unable to init mailet " + mailetClassName); System.err.println("Check spool manager logs for more details."); ex.printStackTrace(); //System.exit(1); throw ex; } //Add this pair to the proces 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(threads) .append(" Thread(s)"); getLogger().info(infoBuffer.toString()); } for ( int i = 0 ; i < threads ; i++ ) workerPool.execute(this); } /** * This routinely checks the message spool for messages, and processes * them as necessary */ public void run() { boolean infoEnabled = getLogger().isInfoEnabled(); if (infoEnabled) { getLogger().info("run JamesSpoolManager: " + Thread.currentThread().getName()); getLogger().info("spool=" + spool.getClass().getName()); } while(true) { try { String key = spool.accept(); MailImpl mail = spool.retrieve(key); if (infoEnabled) { StringBuffer infoBuffer = new StringBuffer(64) .append("==== Begin processing mail ") .append(mail.getName()) .append("===="); getLogger().info(infoBuffer.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)) { spool.remove(key); if (infoEnabled) { StringBuffer infoBuffer = new StringBuffer(64) .append("==== Removed from spool mail ") .append(mail.getName()) .append("===="); getLogger().info(infoBuffer.toString()); } } mail = null; } catch (Exception e) { e.printStackTrace(); if (getLogger().isErrorEnabled()) { getLogger().error("Exception in JamesSpoolManager.run " + e.getMessage()); } } } } /** * Process this mail message by the appropriate processor as designated * in the state of the Mail object. */ protected void process(MailImpl 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) { throw new MailetException("Unable to find processor " + processorName); } StringBuffer logMessageBuffer = null; if (getLogger().isInfoEnabled()) { logMessageBuffer = new StringBuffer(64) .append("Processing ") .append(mail.getName()) .append(" through ") .append(processorName); getLogger().info(logMessageBuffer.toString()); } processor.service(mail); return; } catch (Exception e) { // This is a strange error situation that shouldn't ordinarily // happen StringBuffer exceptionBuffer = new StringBuffer(64) .append("Exception in processor <") .append(processorName) .append(">"); System.err.println(exceptionBuffer.toString()); e.printStackTrace(); 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 error processor mail.setState(Mail.ERROR); mail.setErrorMessage(e.getMessage()); } } StringBuffer logMessageBuffer = new StringBuffer(128) .append("Processed") .append(mail.getName()) .append(" through ") .append(processorName); getLogger().info(logMessageBuffer.toString()); getLogger().info("Result was " + mail.getState()); } } // Shutdown processors public void dispose() { getLogger().info("JamesSpoolManager dispose..."); 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); } } }