xmlblasterimpl.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      XmlBlasterImpl.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Native Interface to xmlBlaster
------------------------------------------------------------------------------*/
package org.xmlBlaster.engine;

import java.util.logging.Logger;
import java.util.logging.Level;
import org.xmlBlaster.engine.qos.GetQosServer;
import org.xmlBlaster.engine.qos.AddressServer;
import org.xmlBlaster.engine.qos.EraseQosServer;
import org.xmlBlaster.engine.qos.SubscribeQosServer;
import org.xmlBlaster.engine.qos.UnSubscribeQosServer;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.util.def.MethodName;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.key.QueryKeyData;
import org.xmlBlaster.util.qos.QueryQosData;
import org.xmlBlaster.util.qos.QosData;
import org.xmlBlaster.util.MsgUnit;
import org.xmlBlaster.util.MsgUnitRaw;
import org.xmlBlaster.util.dispatch.DispatchStatistic;
import org.xmlBlaster.authentication.Authenticate;
import org.xmlBlaster.authentication.SessionInfo;
import org.xmlBlaster.authentication.plugins.CryptDataHolder;
import org.xmlBlaster.authentication.plugins.DataHolder;
import org.xmlBlaster.authentication.plugins.I_Session;
import org.xmlBlaster.authentication.plugins.I_Subject;
import org.xmlBlaster.authentication.plugins.SessionHolder;

/**
 * This is the native implementation of the xmlBlaster interface.
 * <p />
 * All protocol drivers access xmlBlaster through these methods.
 * <br />
 * All security checks are done here, and then the call is delegated
 * to RequestBroker for processing.
 * <br />
 * Most access methods are provided with varying arguments for your convenience.
 *
 * @see org.xmlBlaster.engine.RequestBroker
 * @see org.xmlBlaster.protocol.I_XmlBlaster
 * @author [e-mail]
 */
public class XmlBlasterImpl implements org.xmlBlaster.protocol.I_XmlBlaster
{
   private final String ME;
   private final RequestBroker requestBroker;
   private final Authenticate authenticate;
   private final AvailabilityChecker availabilityChecker;
   private final ServerScope glob;
   private static Logger log = Logger.getLogger(XmlBlasterImpl.class.getName());
   private final byte[] EMPTY_BYTEARR = null;

   /**
    * One instance of this represents one xmlBlaster server.
    * @param authenticate The authentication service
    */
   public XmlBlasterImpl(Authenticate authenticate) throws XmlBlasterException
   {
      this.authenticate = authenticate;
      this.glob = authenticate.getGlobal();
      this.ME = "XmlBlasterImpl" + this.glob.getLogPrefixDashed();

      this.requestBroker = new RequestBroker(authenticate);
      this.availabilityChecker = new AvailabilityChecker(this.glob);
   }

   /**
    * Subscribe to messages.
    * <p />
    * @see org.xmlBlaster.engine.RequestBroker
    */
   public final String subscribe(AddressServer addressServer, String sessionId,
                                 String xmlKey_literal, String qos_literal) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering subscribe(" + sessionId + ", key, qos)");

      try {
         // authentication security check
         SessionInfo sessionInfo = authenticate.check(sessionId);

         // import and authorize message
         MsgUnit msgUnit = importAndAuthorize(sessionInfo, addressServer,
                                       new MsgUnitRaw(xmlKey_literal, EMPTY_BYTEARR, qos_literal),
                                       MethodName.SUBSCRIBE);

         SubscribeQosServer subscribeQos = new SubscribeQosServer(glob, (QueryQosData)msgUnit.getQosData());

         // Invoke xmlBlaster
         String ret = requestBroker.subscribe(sessionInfo, (QueryKeyData)msgUnit.getKeyData(), subscribeQos);


         sessionInfo.getDispatchStatistic().incrNumSubscribe(1);

         // export (encrypt) return value
         MsgUnitRaw in = new MsgUnitRaw(null, (byte[])null, ret);
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.SUBSCRIBE, in);
         dataHolder.setReturnValue(true);
         return sessionInfo.getSecuritySession().exportMessage(dataHolder).getQos();
      }
      catch (Throwable e) {
         throw this.availabilityChecker.checkException(MethodName.SUBSCRIBE, e);
      }
   }

   /**
    * Unsubscribe from messages.
    * <p />
    * @see org.xmlBlaster.engine.RequestBroker
    */
   public final String[] unSubscribe(AddressServer addressServer, String sessionId,
                                     String xmlKey_literal, String qos_literal) throws XmlBlasterException
   {
      if (log.isLoggable(Level.FINER)) log.finer("Entering unSubscribe(" + sessionId + ", key, qos)");

      try {
         // authentication and authorization security checks
         SessionInfo sessionInfo = authenticate.check(sessionId);

         // import and authorize message
         MsgUnit msgUnit = importAndAuthorize(sessionInfo, addressServer,
                                       new MsgUnitRaw(xmlKey_literal, EMPTY_BYTEARR, qos_literal),
                                       MethodName.UNSUBSCRIBE);

         UnSubscribeQosServer unSubscribeQosServer = new UnSubscribeQosServer(glob, (QueryQosData)msgUnit.getQosData());

         // Invoke xmlBlaster
         String [] retArr = requestBroker.unSubscribe(sessionInfo, (QueryKeyData)msgUnit.getKeyData(), unSubscribeQosServer);

         sessionInfo.getDispatchStatistic().incrNumUnSubscribe(1);

         // export (encrypt) return value
         I_Session sec = sessionInfo.getSecuritySession();
         for (int ii=0; ii<retArr.length; ii++) {
            CryptDataHolder dataHolder = new CryptDataHolder(MethodName.UNSUBSCRIBE, new MsgUnitRaw(null, (byte[])null, retArr[ii]));
            dataHolder.setReturnValue(true);
            retArr[ii] = sec.exportMessage(dataHolder).getQos();
         }
         return retArr;

      }
      catch (Throwable e) {
         throw this.availabilityChecker.checkException(MethodName.UNSUBSCRIBE, e);
      }
   }

   /**
    * Publish a message.
    * <p />
    * @see org.xmlBlaster.engine.RequestBroker
    */
   public final String publish(AddressServer addressServer, String sessionId, MsgUnitRaw msgUnitRaw) throws XmlBlasterException
   {
      if (log.isLoggable(Level.FINER)) log.finer("Entering publish()");

      try {
         // authentication and authorization security checks
         SessionInfo sessionInfo = authenticate.check(sessionId);

         MsgUnit msgUnit = importAndAuthorize(sessionInfo, addressServer, msgUnitRaw, MethodName.PUBLISH);

         String ret = requestBroker.publish(sessionInfo, msgUnit);

         sessionInfo.getDispatchStatistic().incrNumPublish(1);

         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.PUBLISH, new MsgUnitRaw(null, (byte[])null, ret));
         dataHolder.setReturnValue(true);
         return sessionInfo.getSecuritySession().exportMessage(dataHolder).getQos();
      }
      catch (Throwable e) {
         throw this.availabilityChecker.checkException(MethodName.PUBLISH, e);
      }
   }

   /**
    * Publish messages.
    * <p />
    * @see org.xmlBlaster.engine.RequestBroker
    */
   public final String[] publishArr(AddressServer addressServer, String sessionId,
                                    MsgUnitRaw[] msgUnitArr) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering publishArr()");

      try {
         // authentication and authorization security checks
         SessionInfo sessionInfo = authenticate.check(sessionId);
         I_Session sec = sessionInfo.getSecuritySession();

         // How to guarantee complete transaction?
         DispatchStatistic statistic = sessionInfo.getDispatchStatistic();
         String[] returnArr = new String[msgUnitArr.length];
         for (int ii=0; ii<msgUnitArr.length; ii++) {
            // TODO: Implement native PUBLISH_ARR
            MsgUnit msgUnit = importAndAuthorize(sessionInfo, addressServer, msgUnitArr[ii], MethodName.PUBLISH);
            String ret = requestBroker.publish(sessionInfo, msgUnit);
            statistic.incrNumPublish(1);
            CryptDataHolder dataHolder = new CryptDataHolder(MethodName.PUBLISH_ARR, new MsgUnitRaw(null, (byte[])null, ret));
            dataHolder.setReturnValue(true);
            returnArr[ii] = sec.exportMessage(dataHolder).getQos();
         }

         return returnArr;
      }
      catch (Throwable e) {
         throw this.availabilityChecker.checkException(MethodName.PUBLISH_ARR, e);
      }
   }

   /**
    * Publish messages.
    * <p />
    * @see org.xmlBlaster.engine.RequestBroker