eventplugin.java

java开源的企业总线.xmlBlaster

      EmailData createEmailData() {
         EmailData emailData = new EmailData(this.to, this.from,
               "", "");
         emailData.setCc(this.cc);
         emailData.setBcc(this.bcc);
         // emailData.setExpiryTime(expiryTimestamp);
         // emailData.addAttachment(new AttachmentHolder(payloadFileName,
         // payloadMimetype, payload));
         return emailData;
      }
   } // end of helper class SmtpDestination

   protected SmtpClient smtpClient;
   protected final Object smtpDestinationMonitor = new Object();
   protected SmtpDestinationHelper smtpDestinationHelper;
   protected String smtpDestinationConfiguration;
   protected Timeout smtpTimeout;
   protected Timestamp smtpTimeoutHandle;
   protected EmailData currentEmailData;

   protected long heartbeatInterval;
   protected Timeout heartbeatTimeout;
   protected Timestamp heartbeatTimeoutHandle;

   /**
    * Helper class to publish messages.
    */
   class PublishDestinationHelper {
      String destination;
      String key, qos;
      String contentTemplate;
      public PublishDestinationHelper(String destination) throws XmlBlasterException {
         Map map = StringPairTokenizer.parseLineToProperties(destination);
         if (map.containsKey("publish.key"))
            this.key = (String) map.get("publish.key");
         if (map.containsKey("publish.qos"))
            this.qos = (String) map.get("publish.qos");
         if (map.containsKey("publish.content"))
            this.contentTemplate = (String) map.get("publish.content");
         else
            this.contentTemplate = "$_{eventType}";
      }
      MsgKeyData getPublishKey(String summary, String description,
            String eventType, String errorCode) throws XmlBlasterException {
         if (this.key != null) {
            return engineGlob.getMsgKeyFactory().readObject(this.key);
         }
         //PublishKey publishKey = new PublishKey(engineGlob, Constants.EVENT_OID_LOGIN/*"__sys__Login"*/, "text/plain");
         // TODO: invent an oid depending on the eventType:
         PublishKey publishKey = new PublishKey(engineGlob, "__sys__Event", "text/plain", "1.0");
         publishKey.setClientTags("<org.xmlBlaster><event/></org.xmlBlaster>");
         return publishKey.getData();
      }
      MsgQosData getPublishQos(String summary, String description,
            String eventType, String errorCode, SessionName sessionName) throws XmlBlasterException {
         MsgQosData msgQosData = null;
         if (this.qos != null) {
            msgQosData = engineGlob.getMsgQosFactory().readObject(this.qos);
         }
         else {
            PublishQos publishQos = new PublishQos(engineGlob);
            publishQos.setLifeTime(-1L);
            publishQos.setForceUpdate(true);
            // TODO: Configure history depth to 0 only on first publish
            TopicProperty topicProperty = new TopicProperty(engineGlob);
            HistoryQueueProperty historyQueueProperty = new HistoryQueueProperty(engineGlob, engineGlob.getId());
            historyQueueProperty.setMaxEntriesCache(2);
            historyQueueProperty.setMaxEntries(2);
            topicProperty.setHistoryQueueProperty(historyQueueProperty);
            publishQos.setTopicProperty(topicProperty);
            msgQosData = publishQos.getData();
         }
         if (summary != null && summary.length() > 0)
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_SUMMARY, summary); // "_summary"
         if (description != null && description.length() > 0)
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_DESCRIPTION, description);
         if (eventType != null && eventType.length() > 0)
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_EVENTTYPE, eventType);
         if (errorCode != null && errorCode.length() > 0)
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_ERRORCODE, errorCode);
         if (sessionName != null) {
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_PUBSESSIONID,
                  sessionName.getPublicSessionId());
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_SUBJECTID,
                  sessionName.getLoginName());
            msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_ABSOLUTENAME,
                  sessionName.getAbsoluteName());
            /*
            // To be backwards compatible with loginEvent=true setting:
            // deprecated:
            msgQosData.addClientProperty("__publicSessionId",
                  sessionName.getPublicSessionId());
            msgQosData.addClientProperty("__subjectId",
                  sessionName.getLoginName());
            msgQosData.addClientProperty("__absoluteName",
                  sessionName.getAbsoluteName());
            // TODO: backwards compatible?
            //msgUnit.setContent(sessionName.getLoginName().getBytes());
            // To be backwards compatible with loginEvent=true setting:
            */
         }
         msgQosData.addClientProperty(Constants.EVENTPLUGIN_PROP_NODEID, engineGlob.getId());

         return msgQosData;
      }
      MsgUnit getMsgUnit(String summary, String description,
            String eventType, String errorCode, SessionName sessionName) throws XmlBlasterException {
         String content = replaceTokens(
            this.contentTemplate, summary, description, eventType, errorCode);
         return new MsgUnit(
               getPublishKey(summary, description, eventType, errorCode),
               content.getBytes(),
               getPublishQos(summary, description, eventType, errorCode, sessionName));

      }
   }
   protected PublishDestinationHelper publishDestinationHelper;
   protected String publishDestinationConfiguration;

   /**
    * Helper class to send a JMX notification.
    */
   class JmxDestinationHelper {
      String destination;
      String contentTemplate;
      public JmxDestinationHelper(String destination) throws XmlBlasterException {
         Map map = StringPairTokenizer.parseLineToProperties(destination);
         if (map.containsKey("jmx.content"))
            this.contentTemplate = (String) map.get("jmx.content");
         else
            this.contentTemplate = "$_{eventType}: $_{summary}";
      }
      String getMessage(String summary, String description,
            String eventType, String errorCode, SessionName sessionName) throws XmlBlasterException {
         String content = replaceTokens(
            this.contentTemplate, summary, description, eventType, errorCode);
         return content;

      }
   }
   protected JmxDestinationHelper jmxDestinationHelper;
   protected String jmxDestinationConfiguration;

   public EventPlugin() {
      synchronized (EventPlugin.class) {
         staticInstanceCounter++;
         this.instanceCounter = staticInstanceCounter;
         log.fine("instance #" + instanceCounter + " created");
      }
   }

   /**
    * Initializes the plugin
    *
    * @see org.xmlBlaster.util.plugin.I_Plugin#init(org.xmlBlaster.util.Global,
    *      org.xmlBlaster.util.plugin.PluginInfo)
    */
   public void init(org.xmlBlaster.util.Global utilGlob, PluginInfo pluginInfo)
         throws XmlBlasterException {
      this.pluginConfig = pluginInfo;
      this.uniqueInstanceName = pluginInfo.getId() + "-" + instanceCounter;
      this.glob = utilGlob.getClone(utilGlob.getNativeConnectArgs());
      this.glob.addObjectEntry(Constants.OBJECT_ENTRY_ServerScope, utilGlob // "ServerNodeScope"
            .getObjectEntry(Constants.OBJECT_ENTRY_ServerScope));

      this.engineGlob = (org.xmlBlaster.engine.ServerScope) utilGlob
            .getObjectEntry(Constants.OBJECT_ENTRY_ServerScope);
      this.requestBroker = engineGlob.getRequestBroker();
      this.sessionInfo = requestBroker.getInternalSessionInfo();

      // For JMX instanceName may not contain ","
      this.contextNode = new ContextNode(ContextNode.SERVICE_MARKER_TAG,
            "EventPlugin[" + getType() + "]", this.engineGlob.getScopeContextNode());
      this.mbeanHandle = this.engineGlob.registerMBean(this.contextNode, this);

      this.eventTypes = this.glob.get("eventTypes", "", null,
            this.pluginConfig);
      if (this.eventTypes == null || this.eventTypes.trim().length() == 0) {
         log.warning("Please configure an attribute 'eventTypes', there is nothing to do for us.");
         return;
      }
      this.eventTypes = this.eventTypes.trim();

      StringBuffer destLogStr = new StringBuffer(1024);

      // Sending the events with email?
      this.smtpDestinationConfiguration = this.glob.get("destination.smtp", "", null,
            this.pluginConfig);
      if (this.smtpDestinationConfiguration != null && this.smtpDestinationConfiguration.trim().length() > 0) {
         this.smtpDestinationConfiguration = this.smtpDestinationConfiguration.trim();
         setupSmtpSink(this.smtpDestinationConfiguration);
         if (destLogStr.length() > 0) destLogStr.append(",");
         destLogStr.append("destination.smtp");
         if (this.smtpDestinationConfiguration.length() > 0)
            destLogStr.append("(").append(this.smtpDestinationConfiguration).append(")");
      }

      // Sending the events with publish()?
      this.publishDestinationConfiguration = this.glob.get("destination.publish", (String)null, null,
            this.pluginConfig);
      if (this.publishDestinationConfiguration != null) {
         this.publishDestinationConfiguration = this.publishDestinationConfiguration.trim();
         this.publishDestinationHelper = new PublishDestinationHelper(this.publishDestinationConfiguration);
         if (destLogStr.length() > 0) destLogStr.append(",");
         destLogStr.append("destination.publish");
         if (this.publishDestinationConfiguration.length() > 0)
            destLogStr.append("(").append(this.publishDestinationConfiguration).append(")");
         if (this.eventTypes.indexOf("logging/severe/*") != -1 || this.eventTypes.indexOf("log/error/*") != -1)
            log.warning("The combination of 'destination.publish' with 'logging/severe/*' is dangerous as it could loop forever, it is supressed.");
         else if (this.eventTypes.indexOf("logging/severe/") != -1 || this.eventTypes.indexOf("log/error/") != -1)
            log.warning("The combination of 'destination.publish' with 'logging/severe/xyz' is dangerous as it could loop forever.");
         if (this.eventTypes.indexOf("logging/warning") != -1 || this.eventTypes.indexOf("logging/warn") != -1)
            log.warning("The combination of 'destination.publish' with 'logging/warning' is very dangerous as it could loop forever, it is supressed.");
      }

      // Sending the events as a JMX notification?
      this.jmxDestinationConfiguration = this.glob.get("destination.jmx", (String)null, null,
            this.pluginConfig);
      if (this.jmxDestinationConfiguration != null) {
         this.jmxDestinationConfiguration = this.jmxDestinationConfiguration.trim();
         this.jmxDestinationHelper = new JmxDestinationHelper(this.jmxDestinationConfiguration);
         if (destLogStr.length() > 0) destLogStr.append(",");
         destLogStr.append("destination.jmx");
         if (this.jmxDestinationConfiguration.length() > 0)
            destLogStr.append("(").append(this.jmxDestinationConfiguration).append(")");
      }

      if (destLogStr.length() < 1) {
         log.warning("Please configure a data sink attribute 'destination.*', there is nothing to do for us.");
         return;
      }

      this.isActive = true;

      registerEventTypes(this.eventTypes);

      log.info("Configured to send core events of type '" + this.eventTypes.trim()
            + "' to '" + destLogStr.toString() + "'");
   } // init()

   /**
    * Find out which events to listen.
    * @param eventTypes A commas seperated list of supported events, e.g. <code>logging/severe/*,logging/warning/*</code>
    */
   private void registerEventTypes(String eventTypes) throws XmlBlasterException {
      String[] eventTypeArr = StringPairTokenizer.parseLine(eventTypes);

      ServerScope serverScope = requestBroker.getServerScope();
      QueueEventHandler queueEventHandler = null;
      MapEventHandler mapEventHandler = null;

      for (int i = 0; i < eventTypeArr.length; i++) {
         String event = eventTypeArr[i].trim();
         if (event.length() < 1) continue; // Allow ',' at end

         try {
            // "logging/severe/*"
            // TODO: support specific channels as "logging/severe/*" -> "logging/severe/core"
            if (event.startsWith(ContextNode.LOGGING_MARKER_TAG+"/severe/")
                  || event.startsWith("logging/error/")) {
               // We want to be notified if a log.error() is called, this will
               // notify our LogableDevice.log() method
               XbNotifyHandler.instance().register(Level.SEVERE.intValue(), this);
               if (this.loggingSet == null) this.loggingSet = new TreeSet();
               this.loggingSet.add(event);
            }
            // "logging/warning/*"
            else if (event.startsWith(ContextNode.LOGGING_MARKER_TAG+"/warning/")
                  || event.startsWith("logging/warn/")) {
               XbNotifyHandler.instance().register(Level.WARNING.intValue(), this);
               if (this.loggingSet == null) this.loggingSet = new TreeSet();
               this.loggingSet.add(event);
            }
            // "service/RunlevelManager/event/startupRunlevel8", "service/RunlevelManager/event/shutdownRunlevel7"