clusternode.java

java开源的企业总线.xmlBlaster

   /**
    * Set the filter rules to determine the master of a message.
    */
   public void addDomainInfo(NodeDomainInfo domainInfo) {
      // How to avoid duplicates? key = domainInfo.getQuery() does not help because of subtags
      this.domainInfoMap.put(""+domainInfo.getCount(), domainInfo);
   }

   /**
    * Check if we have currently a functional connection to this node.
    * <p />
    * Note: A call to this check does try to login if the connection
    *       was not initialized before. This is sometimes an unwanted behavior.
    *       On the other hand, without trying to login it is difficult to
    *       determine the connection state.
    */
   public boolean isConnected() throws XmlBlasterException {
      if (isLocalNode())
         return true;
      I_XmlBlasterAccess con = getXmlBlasterAccess();
      if (con != null)
         return con.isConnected();
      return false;
   }

   /**
    * Check if we are currently polling for a connection to this node.
    * <p />
    * Note: A call to this check does try to login if the connection
    *       was not initialized before. This is sometimes an unwanted behavior.
    *       On the other hand, without trying to login it is difficult to
    *       determine the connection state.
    */
   public boolean isPolling() throws XmlBlasterException {
      if (isLocalNode())
         return false;
      I_XmlBlasterAccess con = getXmlBlasterAccess();
      if (con != null)
         return con.isPolling();
      return false;
   }

   /**
    * Check if we currently have an open connection to this node.
    * @return Always true for the local node
    */
   public boolean isAlive() throws XmlBlasterException {
      if (isLocalNode())
         return true;
      I_XmlBlasterAccess con = getXmlBlasterAccess();
      if (con != null)
         return con.isAlive();
      return false;
   }

   /**
    * Is this node usable.
    * @return true if we are logged in or are polling for the node<br />
    *         false if the node should not be used
    */
   public boolean isAllowed() throws XmlBlasterException {
      if (isLocalNode())
         return true;
      return isAllowed;
   }

   /**
    * Returns the current connection state to the node.
    * @return 0 -> We are logged in<br />
    *         1 -> We are polling for this node<br />
    *         2 -> The node is not allowed to use<br />
    */
   public int getConnectionState() throws XmlBlasterException {
      if (!isConnected()) // connect() was not yet called
         return 2;

      if (isAlive())
         return 0;
      else if (isPolling())
         return 1;
      else
         return 2;
   }

   /**
    * Check if we have currently a functional connection to this node.
    */
   public boolean isLocalNode() {
      return getId().equals(this.fatherGlob.getId());
   }

   /**
    * Set the connection status
    */
   public void setAvailable(boolean available) {
      this.available = available;
   }

   /**
    * Needed for TreeSet and MapSet, implements Comparable.
    */
   public int compareTo(Object obj)  {
      ClusterNode n = (ClusterNode)obj;
      return getId().compareTo(n.getId());
   }

   /**
    * This is the callback method invoked from I_XmlBlasterAccess
    * informing the client in an asynchronous mode if the connection was established.
    * <p />
    * This method is enforced through interface I_ConnectionStateListener
    */
   public void reachedAlive(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
      this.available = true;
      if (connection.getQueue().getNumOfEntries() > 0) {
         log.info("Connected to xmlBlaster node '" + getId() + "', sending " + connection.getQueue().getNumOfEntries() + " tailback messages ...");
      }
      else {
         log.info("Connected to " + getId() + ", no backup messages to flush");
      }
   }

   /**
    * This is the callback method invoked from I_XmlBlasterAccess
    * informing the client in an asynchronous mode if the connection was lost.
    * <p />
    * This method is enforced through interface I_ConnectionStateListener
    */
   public void reachedPolling(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
      this.available = false;
      log.warning("I_ConnectionStateListener: No connection to xmlBlaster node '" + getId() + "', we are polling ...");
   }

   /**
    * This is the callback method invoked from I_XmlBlasterAccess
    * informing the client in an asynchronous mode if the connection was lost.
    * <p />
    * This method is enforced through interface I_ConnectionStateListener
    */
   public void reachedDead(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
      this.available = false;
      log.severe("I_ConnectionStateListener: No connection to xmlBlaster node '" + getId() + "', state=DEAD, giving up.");
   }

   /**
    * This is the callback method invoked from xmlBlaster
    * delivering us a new asynchronous message.
    * @see org.xmlBlaster.client.I_Callback#update(String, UpdateKey, byte[], UpdateQos)
    */
   public String update(String cbSessionId, UpdateKey updateKey, byte[] content, UpdateQos updateQos) throws XmlBlasterException {
      if (isLocalNode()) {
         log.severe("Receiving unexpected update of message oid=" + updateKey.getOid() + " from xmlBlaster node '" + getId() + "' sessionId=" + cbSessionId);
         Thread.dumpStack();
      }
      else {
         if (log.isLoggable(Level.FINER)) log.finer("Receiving update of message oid=" + updateKey.getOid() + " from xmlBlaster node '" + getId() + "' sessionId=" + cbSessionId);
      }

      // Important: Do authentication of sender:
      if (!getNodeInfo().getCbSessionId().equals(cbSessionId)) {
         log.warning("The callback sessionId '" + cbSessionId + "' is invalid, no access to " + this.remoteGlob.getId());
         throw new XmlBlasterException(updateKey.getGlobal(), ErrorCode.USER_SECURITY_AUTHENTICATION_ACCESSDENIED, ME,
                     "Your callback sessionId is invalid, no access to " + this.remoteGlob.getId());
      }

      // Publish messages to our RequestBroker WITHOUT ANY FURTHER SECURITY CHECKS:

      String ret = this.fatherGlob.getRequestBroker().update(sessionInfo, updateKey, content, updateQos.getData());
      if (ret == null || ret.length() < 1)
         return Constants.RET_FORWARD_ERROR;   // OK like this?
      return Constants.RET_OK;
   }

   public void shutdown() {
      resetXmlBlasterAccess(false);
   }

   /**
    * Dump state of this object into a XML ASCII string.
    */
   public final String toXml() {
      return toXml((String)null, (Properties)null);
   }

   /**
    * Dump state of this object into a XML ASCII string.
    * @param extraOffset indenting of tags for nice output
    */
   public final String toXml(String extraOffset, Properties props) {
      StringBuffer sb = new StringBuffer(512);
      if (extraOffset == null) extraOffset = "";
      String offset = Constants.OFFSET + extraOffset;

      sb.append(offset).append("<clusternode id='").append(getId()).append("'>");

      sb.append(getNodeInfo().toXml(extraOffset + Constants.INDENT, props));

      if (getDomainInfoMap() != null) {
         Iterator it = getDomainInfoMap().values().iterator();
         while (it.hasNext()) {
            NodeDomainInfo info = (NodeDomainInfo)it.next();
            sb.append(info.toXml(extraOffset + Constants.INDENT));
         }
      }

      sb.append(getNodeStateInfo().toXml(extraOffset + Constants.INDENT));

      sb.append(offset).append("</clusternode>");

      return sb.toString();
   }

   public boolean isAvailable() {
      return available;
   }

   /**
    * @return Returns the remoteGlob.
    */
   public org.xmlBlaster.util.Global getRemoteGlob() {
      return this.remoteGlob;
   }
}