publisher.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      Publisher.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE filep
------------------------------------------------------------------------------*/

package org.xmlBlaster.client.filepoller;

import java.util.Set;

import java.util.logging.Logger;
import java.util.logging.Level;
import org.xmlBlaster.client.I_XmlBlasterAccess;
import org.xmlBlaster.client.key.PublishKey;
import org.xmlBlaster.client.qos.ConnectQos;
import org.xmlBlaster.client.qos.DisconnectQos;
import org.xmlBlaster.util.Global;
import org.xmlBlaster.util.I_Timeout;
import org.xmlBlaster.util.MsgUnit;
import org.xmlBlaster.util.Timeout;
import org.xmlBlaster.util.Timestamp;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.util.plugin.I_PluginConfig;
import org.xmlBlaster.util.qos.ConnectQosData;


/**
 * Publisher
 * @author <a href="mailto:[e-mail]">Michele Laghi</a>
 * @deprectated it is now replaced by the corresponding class in org.xmlBlaster.contrib.filewatcher
 */
public class Publisher implements I_Timeout {

   private String ME = "Publisher";
   private Global global;
   private static Logger log = Logger.getLogger(Publisher.class.getName());
   private DirectoryManager directoryManager;
   private I_XmlBlasterAccess access;
   private String publishKey;
   private String publishQos;
   private ConnectQos connectQos;

   private long pollInterval;
   private long maximumFileSize;
   private String fileFilter;
   private String filterType;
   private String directoryName;
   private boolean copyOnMove;
   private String sent;
   private String discarded;
   private String lockExtention;
   private long delaySinceLastFileChange;
   
   public static final String USE_REGEX = "regex";

   private Timestamp timeoutHandle;
   private static Timeout timeout = new Timeout("FileSystem-Poller");
   
   /** used to identify if it has shut down (to get a new global) */
   private boolean isShutdown;
   /** used to break the loop in doPublish when shutting down */
   private boolean forceShutdown;
   
   /** only used as a default login name and logging */
   private String name;
   
   private boolean isActive;
   
   
   // private I_PluginConfig pluginConfig;
   
   public Publisher(Global globOrig, String name, I_PluginConfig pluginConfig) throws XmlBlasterException {
      ME += "-" + name;
      this.name = name;
      this.isShutdown = false;
      this.global = globOrig.getClone(globOrig.getNativeConnectArgs()); // sets session.timeout to 0 etc.
      // this.pluginConfig = pluginConfig;

      if (log.isLoggable(Level.FINER)) 
         log.finer(ME+": constructor");
      // retrieve all necessary properties:
      String tmp = null;
      tmp = this.global.get("publishKey", (String)null, null, pluginConfig);
      String topicName =  this.global.get("topicName", (String)null, null, pluginConfig);
      if (tmp != null) {
         // this.publishKey = new PublishKey(this.global, new MsgKeyData(this.global, tmp));
         this.publishKey = tmp;
         if (topicName != null)
            log.warning(ME+": constructor: since 'publishKey' is defined, 'topicName' will be ignored");
      }
      else {
         if (topicName == null)
            throw new XmlBlasterException(this.global, ErrorCode.USER_CONFIGURATION, ME, "at least one of the properties 'topicName' or 'publishKey' must be defined");
         this.publishKey = (new PublishKey(this.global, topicName)).toXml(); 
      }
      
      this.publishQos = this.global.get("publishQos", "<qos/>", null, pluginConfig);

      tmp  = this.global.get("connectQos", (String)null, null, pluginConfig);
      if (tmp != null) {
         ConnectQosData data = this.global.getConnectQosFactory().readObject(tmp);
         this.connectQos = new ConnectQos(this.global, data);
      }
      else {
         String userId = this.global.get("loginName", "_" + this.name, null, pluginConfig);
         String password = this.global.get("password", (String)null, null, pluginConfig);
         this.connectQos = new ConnectQos(this.global, userId, password);
         this.global.addObjectEntry(Constants.OBJECT_ENTRY_ServerScope, globOrig.getObjectEntry(Constants.OBJECT_ENTRY_ServerScope));
      }

      this.fileFilter =  this.global.get("fileFilter", (String)null, null, pluginConfig);
      this.directoryName = this.global.get("directoryName", (String)null, null, pluginConfig);
      if (directoryName == null)
         throw new XmlBlasterException(this.global, ErrorCode.USER_CONFIGURATION, ME, "constructor: 'directoryName' is mandatory");
      
      this.maximumFileSize = this.global.get("maximumFileSize", 10000000L, null, pluginConfig);
      this.delaySinceLastFileChange = this.global.get("delaySinceLastFileChange", 10000L, null, pluginConfig);
      this.pollInterval = this.global.get("pollInterval", 2000L, null, pluginConfig);

      this.sent =  this.global.get("sent", (String)null, null, pluginConfig);
      this.discarded =  this.global.get("discarded", (String)null, null, pluginConfig);
      this.lockExtention =  this.global.get("lockExtention", (String)null, null, pluginConfig);
     
      // this would throw an exception and act as a validation if something is not OK in configuration
      new MsgUnit(this.publishKey, (byte[])null, this.publishQos);
      this.filterType = this.global.get("filterType", "simple", null, pluginConfig);
      this.copyOnMove = this.global.get("copyOnMove", true, null, pluginConfig);
      
      createDirectoryManager();
   }
   
   /**
    * Create the file checker instance with the current configuration. 
    * @throws XmlBlasterException
    */
   private void createDirectoryManager() throws XmlBlasterException {
      boolean isTrueRegex = USE_REGEX.equalsIgnoreCase(filterType);
      this.directoryManager = new DirectoryManager(this.global,
            this.name, this.directoryName, this.delaySinceLastFileChange, 
            this.fileFilter, this.sent, this.discarded, this.lockExtention, isTrueRegex,
            this.copyOnMove);
   }

   /**
    * Useful for JMX invocations
    */
   private void reCreateDirectoryManager() {
      try {
         createDirectoryManager();
      } catch (XmlBlasterException e) {
         throw new IllegalArgumentException(e.getMessage());
      }
   }

   public String toString() {
      return "FilePoller " + this.filterType + " directoryName=" + this.directoryName + " fileFilter='" + this.fileFilter + "'";
   }
   
   /**
    * Connects to the xmlBlaster.
    * 
    * @throws XmlBlasterException
    */
   public synchronized void init() throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) 
         log.finer(ME+": init");
      if (this.isShutdown) { // on a second init
         this.global = this.global.getClone(null);
      }
      this.isShutdown = false;
      this.forceShutdown = false;
      this.access = this.global.getXmlBlasterAccess();
      // no callback listener (we are not subscribing and don't want ptp)
      this.access.connect(this.connectQos, null);
      this.isActive = true;
      if (this.pollInterval >= 0)
         this.timeoutHandle = timeout.addTimeoutListener(this, this.pollInterval, null);
   }
   
   /**
    * If an exception occurs it means it could not publish the entry
    * @throws XmlBlasterException
    */
   public void shutdown() throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) 
         log.finer(ME+": shutdown");
      timeout.removeTimeoutListener(this.timeoutHandle);
      this.isActive = false;
      this.forceShutdown = true; // in case doPublish is looping due to an exception
      synchronized (this) {
         this.isShutdown = false;
         this.access.disconnect(new DisconnectQos(this.global));
         this.global.shutdown();
      }
   }
   
   /**
    * Fail-safe sending files. 
    * @return Comman separated list of send file names
    */
   public synchronized void publish() {
      while (true) {
         try {
            doPublish();
            break;
         }
         catch (XmlBlasterException ex) {
            log.severe(ME+": publish: exception " + ex.getMessage());
            try {
               Thread.sleep(this.pollInterval);
            }
            catch  (Exception e) {}      
         }
         if (this.forceShutdown)
            break;
      }
   }

   /**
    * Create a comma separated list of file names. 
    * @param infos
    * @param max Max file names to collect
    * @return
    */
   public String toString(FileInfo[] infos, int max) {
      StringBuffer sb = new StringBuffer();
      if (max <= 0) max = infos.length;
      if (max > infos.length) max = infos.length;
      for (int i=0; i<max; i++) {
         if (i>0) sb.append(",");
         sb.append(infos[i].getRelativeName());
      }
      return sb.toString();
   }
   
   /**
    * Publish file to xmlBlaster. 
    * @return An empty string if nothing was sent, is never null
    * @throws XmlBlasterException
    */
   private FileInfo[] doPublish() throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) 
         log.finer(ME+": doPublish");
      Set entries = this.directoryManager.getEntries();
      if (entries == null || entries.size() < 1)
         return new FileInfo[0];
      FileInfo[] infos = (FileInfo[])entries.toArray(new FileInfo[entries.size()]);
      for (int i=0; i < infos.length; i++) {
         if (this.maximumFileSize <= 0L || infos[i].getSize() <= this.maximumFileSize) {
            if (infos[i].getSize() > Integer.MAX_VALUE) {
               log.severe(ME+": doPublish: sizes bigger than '" + Integer.MAX_VALUE + "' are currently not implemented");
            }