eventplugin.java

java开源的企业总线.xmlBlaster

            else if (event.startsWith(this.engineGlob.getRunlevelManager().getContextNode().getRelativeName()+"/event/")) {
               log.fine("Register event = " + event);
               this.engineGlob.getRunlevelManager().addRunlevelListener(this);
               if (this.runlevelSet == null) this.runlevelSet = new TreeSet();
               this.runlevelSet.add(event);
            }
            else if (isCallbackStateEvent(event)) {
            //else if (event.endsWith("/event/callbackState") || event.endsWith("/event/callbackAlive") || event.endsWith("/event/callbackPolling") || event.endsWith("/event/callbackDead")) {
               // OK: "client/joe/session/1/event/callbackState"
               // Not yet supported: "client/joe/session/1/event/callbackAlive", "client/joe/session/1/event/callbackPolling"
               int index = event.lastIndexOf("/event/");
               String name = event.substring(0, index);
               SessionName sessionName = new SessionName(this.engineGlob, name);
               this.requestBroker.getAuthenticate().addClientListener(this);
               if (this.callbackSessionStateSet == null) this.callbackSessionStateSet = new TreeSet();
               if (event.startsWith(ContextNode.SUBJECT_MARKER_TAG+ContextNode.SEP+"*"+ContextNode.SEP) ||
                   event.endsWith(ContextNode.SESSION_MARKER_TAG+ContextNode.SEP+"*"+"/event/callbackState")) {
                  // "client/*/session/1/event/callbackState" or "client/joe/session/*/event/callbackState"
                  if (this.pendingCallbackSessionInfoSet == null) this.pendingCallbackSessionInfoSet = new TreeSet();
                  this.pendingCallbackSessionInfoSet.add(name);
                  this.callbackSessionStateSet.add(name);
                  SubjectInfo[] subs = null;
                  log.fine("Register existing wildcard callback session state event = " + event);
                  if (event.startsWith(ContextNode.SUBJECT_MARKER_TAG+ContextNode.SEP+"*"+ContextNode.SEP) &&
                		  !event.endsWith(ContextNode.SESSION_MARKER_TAG+ContextNode.SEP+"*"+"/event/callbackState")) {
	                  subs = this.requestBroker.getAuthenticate().getSubjectInfoArr();
	                  for (int sj=0; sj<subs.length; sj++) {
	                	  SubjectInfo subjectInfo = subs[sj];
	                	  if (!wildcardMatch(sessionName.getLoginName(), subjectInfo.getLoginName()))
                			  continue;
	                	  SessionInfo[] ses = subjectInfo.getSessions();
	                	  for (int se=0; se<ses.length; se++) {
	                		  if (!wildcardMatch(sessionName.getPublicSessionId(), ses[se].getPublicSessionId()))
	                			  continue;
	                		  DispatchManager mgr = ses[se].getDispatchManager();
	                          if (mgr != null) {
	                              mgr.addConnectionStatusListener(this);
	                           }
	                	  }
	                  }
                  }
               }
               else {
                  log.fine("Register callback session state event = " + event);
                  SessionInfo sessionInfo = this.requestBroker.getAuthenticate().getSessionInfo(sessionName);
                  DispatchManager mgr = null;
                  if (sessionInfo != null)
                     mgr = sessionInfo.getDispatchManager();
                  if (mgr != null) {
                     mgr.addConnectionStatusListener(this);
                  }
                  else {
                     if (this.pendingCallbackSessionInfoSet == null) this.pendingCallbackSessionInfoSet = new TreeSet();
                     this.pendingCallbackSessionInfoSet.add(sessionName.getAbsoluteName());
                  }
                  this.callbackSessionStateSet.add(sessionName.getRelativeName());
               }
            }
            else if (isQueueEvent(event)) {
               if (queueEventHandler == null) {
                  queueEventHandler = new QueueEventHandler(serverScope, this);
               }
               queueEventHandler.registerEventType(this, event);
            }
            else if (isPersistenceEvent(event)) {
               if (mapEventHandler == null) {
                  mapEventHandler = new MapEventHandler(serverScope, this);
               }
               mapEventHandler.registerEventType(this, event);
            }
            else if (event.startsWith(ContextNode.SUBJECT_MARKER_TAG+ContextNode.SEP)) {
               // REGEX: "client/.*/session/.*/event/.*"
               // "client/joe/session/1/event/connect", "client/*/session/*/event/disconnect"
               // "client/joe/session/1/event/subscribe"
               log.fine("Register login/logout event = " + event);
               if (event.endsWith("/event/subscribe") || event.endsWith("/event/unSubscribe"))
                  this.requestBroker.addSubscriptionListener(this);
               else if (event.endsWith("/event/remoteProperties"))
                  this.requestBroker.addRemotePropertiesListener(this); // I_RemotePropertiesListener
               else
                  this.requestBroker.getAuthenticate().addClientListener(this);
               if (this.clientSet == null) this.clientSet = new TreeSet();
               this.clientSet.add(event);
            }
            else if (event.startsWith(ContextNode.TOPIC_MARKER_TAG+ContextNode.SEP)) {
               // "topic/hello/event/alive", "topic/hello/event/subscribe" ...
               log.fine("Register topic event = " + event);
               if (event.endsWith("/event/subscribe") || event.endsWith("/event/unSubscribe"))
                  this.requestBroker.addSubscriptionListener(this);
               else
                  this.engineGlob.getTopicAccessor().addTopicListener(this);
               if (this.topicSet == null) this.topicSet = new TreeSet();
               this.topicSet.add(event);
            }
            else if (event.startsWith("heartbeat")) {
               // "heartbeat.360000
               log.fine("Register heartbeat event = " + event);
               int index = event.indexOf(".");
               if (index > 0 && index < event.length()-1)
                  this.heartbeatInterval = Long.valueOf(event.substring(index+1)).longValue();
               else
                  this.heartbeatInterval = Constants.DAY_IN_MILLIS;
               if (this.heartbeatInterval > 0) {
                  // send the first heartbeat directly after startup:
                  long initialInterval = (this.heartbeatInterval > 2000) ? 2000L : this.heartbeatInterval;
                  this.heartbeatTimeout = new Timeout("EventPlugin-HeartbeatTimer");
                  this.heartbeatTimeoutHandle = this.heartbeatTimeout.addTimeoutListener(new I_Timeout() {
                     public void timeout(Object userData) {
                        log.fine("Timeout happened " + userData + ": Sending now heartbeat");
                        newHeartbeatNotification((String)userData);
                        try {
                           heartbeatTimeout.addOrRefreshTimeoutListener(this, heartbeatInterval, userData, heartbeatTimeoutHandle);
                        }
                        catch (XmlBlasterException e) {
                           e.printStackTrace();
                        }

                     }
                   }, initialInterval, event);
               }
            }
            else {
               log.warning("Ignoring unknown '" + event
                     + "' from eventTypes='" + eventTypes + "'");
            }
         }
         catch (Throwable e) {
            e.printStackTrace();
            log.warning("Ignoring '" + event
                     + "' from eventTypes='" + eventTypes + "' because of " + e.toString());
         }
      }

      if (queueEventHandler != null)
         if (serverScope.getQueuePluginManager().setEventHandler(uniqueInstanceName, queueEventHandler) == false) {
         log.severe(ME+getType() + " Can't register queue threshold event, max one such event can be registered for '" + uniqueInstanceName + "'");
         }
      if (mapEventHandler != null)
         if (serverScope.getStoragePluginManager().setEventHandler(uniqueInstanceName, mapEventHandler) == false) {
            log.severe(ME+getType() + " Can't register msgUnitStore event, max one such event can be registered '" + uniqueInstanceName + "'");
         }
   }

   public static boolean isQueueEvent(String txt) {
      return matchesRegex(".*/queue/.*/event/threshold.*", txt);
   }

   public static boolean isPersistenceEvent(String txt) {
      return matchesRegex(".*/persistence/.*/event/threshold.*", txt);
   }

   public static boolean isCallbackStateEvent(String txt) {
	   return matchesRegex("client/.*/session/.*/event/callbackState", txt);
   }

	public final boolean isWildcard(String pattern) {
		return "*".equals(pattern);
	}

	public final boolean wildcardMatch(String pattern, String name) {
		if ("*".equals(pattern))
			return true;
		return pattern.equals(name);
	}

	/** SessionName.java parsed "client/joe/session/*" to pubSessionId=Long.MIN_VALUE */
	public final boolean wildcardMatch(long pattern, long pubSessionId) {
		if (pattern == Long.MIN_VALUE)
			return true;
		return pattern == pubSessionId;
	}

   private static boolean matchesRegex(String pattern, String txt) {
      try {
         RE regex = new RE(pattern, RE.REG_ICASE);
         return regex.isMatch(txt);
      }
      catch (REException ex) {
         ex.printStackTrace();
         return false;
      }
   }

   /**
    * Called when a client sends his remote properties, for example client side errors.
    * eventType == client/* /session/* /event/remoteProperties
    * Enforced by I_RemotePropertiesListener
    */
   public void update(SessionInfo sessionInfo, Map remoteProperties) {
      log.fine("Received new remote properties from client " + sessionInfo.getId());
      SessionName sessionName = sessionInfo.getSessionName();
      String relativeName = sessionName.getRelativeName();

      String event = ContextNode.SEP + ContextNode.EVENT_MARKER_TAG + ContextNode.SEP + "remoteProperties";
      String foundEvent = relativeName + event;  // "client/joe/session/1/event/remoteProperties"
      if (!this.clientSet.contains(foundEvent)) {
         // "client/joe/session/*/event/remoteProperties"
         foundEvent = ContextNode.SUBJECT_MARKER_TAG + ContextNode.SEP + sessionName.getLoginName() + ContextNode.SEP + ContextNode.SESSION_MARKER_TAG + ContextNode.SEP + "*" + event;
         if (!this.clientSet.contains(foundEvent)) {
            // "client/*/session/*/event/remoteProperties"
            foundEvent = ContextNode.SUBJECT_MARKER_TAG + ContextNode.SEP + "*" + ContextNode.SEP + ContextNode.SESSION_MARKER_TAG + ContextNode.SEP + "*" + event;
            if (!this.clientSet.contains(foundEvent)) {
               return;
            }
         }
      }

      try {
         String summary = "Remote properties change for client " + sessionName.getAbsoluteName();
         String description = summary;
         ClientProperty[] cp = sessionInfo.getRemotePropertyArr();
         for (int i=0; i<cp.length; i++)
            description += "\n   " + cp[i].toXml("", "remoteProperty").trim();
         String eventType = foundEvent;
         String errorCode = null;

         if (this.smtpDestinationHelper != null) {
            sendEmail(summary, description, eventType, null, false);
         }

         if (this.publishDestinationHelper != null) {
            sendMessage(summary, description,
                    eventType, errorCode, sessionName);
         }

         if (this.jmxDestinationHelper != null) {
            sendJmxNotification(summary, description, eventType, null, false);
         }
      } catch (Throwable e) {
         e.printStackTrace();
      }

   }

   /**
    * Send a heartbeat message/notification.
    * Called by timeout or by manual trigger (e.g. over jconsole)
    * @param eventType
    */
   protected void newHeartbeatNotification(String eventType) {
      try {
         ContextNode contextNode = this.engineGlob.getContextNode();
         int rl = this.engineGlob.getRunlevelManager().getCurrentRunlevel();
         String summary = "Heartbeat event from " + contextNode.getAbsoluteName() + ", runlevel=" + RunlevelManager.toRunlevelStr(rl) + " (" + rl + ")";
         String errorCode = null;
         String description = "Heartbeat event from " + contextNode.getAbsoluteName() + ", runlevel=" + RunlevelManager.toRunlevelStr(rl) + " (" + rl + ")";
         SessionName sessionName = null;

         if (this.smtpDestinationHelper != null) {
            // Ignores contentTemplate and forces the XML as last argument
            sendEmail(summary, description, eventType, null, false);
         }

         if (this.publishDestinationHelper != null) {
            // Uses XML as message content
            sendMessage(summary, description, eventType, errorCode, sessionName);
         }

         if (this.jmxDestinationHelper != null) {
            sendJmxNotification(summary, description, eventType, null, false);
         }
      } catch (Throwable e) {
         e.printStackTrace();
      }
   }

   /* (non-Javadoc)
    * @see org.xmlBlaster.engine.EventPluginMBean#triggerHeartbeatNotification()
    */
   public String triggerHeartbeatNotification() {
      String eventType = "heartbeat.manuallyTriggered";
      newHeartbeatNotification(eventType);
      return "Triggered event of type '" + eventType + "'";
   }