emailexecutor.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
 Name:      EmailExecutor.java
 Project:   xmlBlaster.org
 Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
 ------------------------------------------------------------------------------*/
package org.xmlBlaster.util.protocol.email;

import org.xmlBlaster.util.ReplaceVariable;
import org.xmlBlaster.util.Global;
import org.xmlBlaster.util.I_ResponseListener;
import org.xmlBlaster.util.SessionName;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.util.def.MethodName;
import org.xmlBlaster.util.plugin.I_PluginConfig;
import org.xmlBlaster.util.plugin.PluginInfo;
import org.xmlBlaster.util.protocol.RequestReplyExecutor;
import org.xmlBlaster.util.protocol.email.Pop3Driver;
import org.xmlBlaster.util.protocol.email.SmtpClient;
import org.xmlBlaster.util.protocol.email.EmailData;
import org.xmlBlaster.util.qos.address.AddressBase;
import org.xmlBlaster.util.xbformat.MsgInfo;
import org.xmlBlaster.util.xbformat.MsgInfoParserFactory;
import org.xmlBlaster.util.xbformat.XbfParser;
import org.xmlBlaster.util.MsgUnitRaw;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.sql.Timestamp;
import java.util.Map;
import java.util.TreeMap;
import java.util.zip.Deflater;
import java.util.zip.Inflater;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.mail.internet.AddressException;
import javax.mail.internet.InternetAddress;

/**
 * Base class to handle request/reply for emails.
 * <p>
 * 
 * @author <a href="mailto:[e-mail]">Marcel Ruff</a>
 */
public abstract class EmailExecutor extends  RequestReplyExecutor implements I_ResponseListener, EmailExecutorMBean {
   private String ME = "EmailExecutor";

   private static Logger log = Logger.getLogger(EmailExecutor.class.getName());
   
   private boolean isShutdown;

   private AddressBase addressBase;

   private I_PluginConfig pluginConfig;

   protected InternetAddress fromAddress;

   protected InternetAddress toAddress;
   
   protected String cc;

   protected String bcc;

   protected SmtpClient smtpClient;

   private String secretSessionId = "";
   
   private String emailSessionId = "";

   protected Pop3Driver pop3Driver;
   
   private Deflater compressor;
   
   private Inflater decompressor;
   
   /** Which message format parser to use */
   protected String msgInfoParserClassName;
   
   protected Map senderLoopProtectionMap = new TreeMap();
   /**
    * Use to protect against looping messages
    */
   protected class LoopProtection {
      LoopProtection(String key, long lastRequestId, long lastPingRequestId) {
         this.key = key;
         this.lastRequestId = lastRequestId;
         this.lastPingRequestId = lastPingRequestId;
      }
      public String key; // email.from address
      //protected String lastSessionId;
      /** Use to protect against looping messages, is a monotonous ascending timestamp */
      public long lastRequestId=-1;
      /** Ping runs in another thread, we need to protect seperately (an update can legally overtake a ping and vice versa) */
      public long lastPingRequestId=-1;
   }
   
   protected final String BOUNCE_MESSAGEID_KEY = "bounce:messageId";
   public static final String BOUNCE_MAILTO_KEY = "mail.to";
   public static final String BOUNCE_MAILFROM_KEY = "mail.from";

   /** 'messageId.mid' */
   protected String messageIdFileName = "messageId" + EmailData.MESSAGEID_EXTENSION;
   
   /** The extension is added later to for example "xmlBlasterMessage.xfb" */
   protected String payloadFileNamePrefix = "xmlBlasterMessage";
   
   protected String subjectTemplate;
   
   protected final String SUBJECT_MESSAGEID_TOKEN = "$_{xmlBlaster/email/messageId}";

   /**
    * This init() is called after the init(Global, PluginInfo)
    * 
    * @param addressBase
    *           Contains the email TO: address
    */
   public void init(Global glob, AddressBase addressBase, PluginInfo pluginConfig)
         throws XmlBlasterException {
      this.addressBase = addressBase;
      this.pluginConfig = pluginConfig;
      this.compressor = new Deflater(Deflater.BEST_COMPRESSION);
      this.decompressor = new Inflater();

      // Add
      //   CbProtocolPlugin[email][1.0]=org.xmlBlaster.protocol.email.CallbackEmailDriver,mail.user=xmlBlaster,mail.password=xmlBlaster,compress/type=zlib:stream
      //   ClientCbServerProtocolPlugin[email][1.0]=org.xmlBlaster.client.protocol.email.EmailCallbackImpl,mail.user=demo,mail.password=demo,mail.pop3.url=pop3://demo:demo@localhost/INBOX,compress/type=zlib:stream
      //   ClientProtocolPlugin[email][1.0]=org.xmlBlaster.client.protocol.email.EmailConnection,mail.user=demo,mail.password=demo,mail.pop3.url=pop3://demo:demo@localhost/INBOX,pop3PollingInterval=1000,compress/type=zlib:stream
      // settings to the clients Address configuration
      if (this.pluginConfig != null)
         this.addressBase.setPluginInfoParameters(this.pluginConfig.getParameters());

      super.initialize(glob, addressBase);
      
      this.secretSessionId = addressBase.getSecretSessionId();

      // The email address to reach the xmlBlaster server (if client side)
      setTo(this.glob.get("mail.smtp.to", "xmlBlaster@localhost", null,
            pluginConfig));
      String to = addressBase.getRawAddress();
      if (to != null && to.length() > 0) {
         // The xmlBlaster address is given on client side
         // but not on server side
         setTo(to);
      }

      Object serverScope = glob.getObjectEntry(Constants.OBJECT_ENTRY_ServerScope); // e.g. in cluster environment
      String fromm = (serverScope != null) ? ((Global)serverScope).getContextNode().getInstanceNameNotNull() : "unknown"; 
      // from="xmlBlaster@localhost"
      setFrom(this.glob.get("mail.smtp.from",
            fromm+"@localhost", null, this.pluginConfig));

      setCc(this.glob.get("mail.smtp.cc",
            "", null, this.pluginConfig));

      setBcc(this.glob.get("mail.smtp.bcc",
            "", null, this.pluginConfig));
      
      // if template contains SUBJECT_MESSAGEID_TOKEN = "${xmlBlaster/email/messageId}"
      // this will be replaced by the current messageId
      this.subjectTemplate = this.glob.get("mail.subject",
            "XmlBlaster Generated Email "+SUBJECT_MESSAGEID_TOKEN, null, this.pluginConfig);


      this.isShutdown = false;
      if (log.isLoggable(Level.FINE)) log.fine("Initialized email connector from=" + this.fromAddress.toString() + " to=" + to);
   }

   /**
    * Access the Pop3Driver. 
    * @return never null
    */
   public Pop3Driver getPop3Driver() throws XmlBlasterException {
      if (this.pop3Driver == null) {
         this.pop3Driver = (Pop3Driver) glob.getObjectEntry(Pop3Driver.OBJECTENTRY_KEY); // Global.instance()
         
         if (this.pop3Driver == null) {
            if (this.glob.isServerSide()) {
               // On server side the Pop3Driver is created by the runlevel manager as configured in xmlBlasterPlugins.xml
               String text = "Please register a Pop3Driver in xmlBlasterPlugins.xml to have 'email' support";
               // If the session was loaded on startup from persistent store we shouldn't to a log.warn
               // but how to detect this? We don't have connectQosServer.isFromPersistenceRecovery(true);
               log.warning(text);
               // Throw a communication exception to go to polling until Pop3Driver is available
               throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, text);
            }
            // On client side we create it dynamically as configured in xmlBlaster.properties
            this.pop3Driver = Pop3Driver.getPop3Driver(glob, this.pluginConfig);
         }
      }
      return this.pop3Driver;
   }
   
   /**
    * TODO: Put into engine.Global and util.Global (see EventPlugin.java)
    * @return
    * @throws XmlBlasterException
    */
   public SmtpClient getSmtpClient() throws XmlBlasterException {
      if (this.smtpClient == null) {
         this.smtpClient = (SmtpClient) glob.getObjectEntry(SmtpClient.OBJECTENTRY_KEY);

         if (this.smtpClient == null) {
            if (this.glob.isServerSide()) {
               // On server side the SmtpClient is created by the runlevel manager as configured in xmlBlasterPlugins.xml
               String text = "Please register a SmtpClient in xmlBlasterPlugins.xml to have 'email' support";
               // If the session was loaded on startup from persistent store we shouldn't to a log.warn
               // but how to detect this?
               log.warning(text);
               throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, text);
            }
            // On client side we create it dynamically as configured in xmlBlaster.properties
            this.smtpClient = SmtpClient.getSmtpClient(this.glob, this.pluginConfig);
         }
      }
      return this.smtpClient;
   }
   
  // public String getType() {
  //    return "email";
  // }
   
   /**
    * Defaults to one day. 
    */
   public long getDefaultResponseTimeout() {
      return Constants.DAY_IN_MILLIS;
   }
   
   /**
    * Defaults to one day. 
    */
   public long getDefaultUpdateResponseTimeout() {
      return Constants.DAY_IN_MILLIS;
   }
   
   /**
    * Which parser to use. 
    * The 'email' protocol uses as a default setting the XbfParser
    * but usig the XmlScriptParser may be convenient as well. 
    * <p />
    * The environment setting 'parserClass=' is checked.
    * @return The class name of the parser, "org.xmlBlaster.util.xbformat.XbfParser"
    */
   public String getMsgInfoParserClassName() {
      if (this.msgInfoParserClassName == null) {
         this.msgInfoParserClassName = this.addressConfig.getEnv("parserClass", XbfParser.class.getName()).getValue();
      }
      return this.msgInfoParserClassName; //XbfParser.class.getName();
   }

   public Object sendEmail(String qos, MethodName methodName,
         boolean expectingResponse) throws XmlBlasterException {
      MsgUnitRaw[] msgArr = { new MsgUnitRaw(null, (byte[])null, qos) };
      return sendEmail(msgArr, methodName,
            expectingResponse);
   }

   public Object sendEmail(String key, String qos, MethodName methodName,
         boolean expectingResponse) throws XmlBlasterException {
      MsgUnitRaw[] msgArr = { new MsgUnitRaw(key, (byte[])null, qos) };
      return sendEmail(msgArr, methodName,
            expectingResponse);
   }

   public Object sendEmail(MsgUnitRaw msgUnit, MethodName methodName,
         boolean expectingResponse) throws XmlBlasterException {
      MsgUnitRaw[] msgArr = { msgUnit };
      return sendEmail(msgArr, methodName, expectingResponse);
   }

   /**
    * This sends the update to the client.
    * 
    * @param methodName
    *           MethodName.UPDATE and others
    * @param withResponse
    *           one of SocketExecutor.WAIT_ON_RESPONSE or SocketExecutor.ONEWAY
    */
   public Object sendEmail(MsgUnitRaw[] msgArr, MethodName methodName,
         boolean expectingResponse) throws XmlBlasterException {
      if (msgArr == null || msgArr.length < 1)
         throw new XmlBlasterException(glob,
               ErrorCode.INTERNAL_ILLEGALARGUMENT, ME, "Illegal sendEmail("