xmlblasterimpl.java

java开源的企业总线.xmlBlaster

    */
   public final void publishOneway(AddressServer addressServer, String sessionId, MsgUnitRaw[] msgUnitArr) {
      if (log.isLoggable(Level.FINER)) log.finer("Entering publishOneway()");
      try {
         publishArr(addressServer, sessionId, msgUnitArr);
      }
      catch (Throwable e) {
	    	MsgUnitRaw msgUnit = (msgUnitArr.length > 0) ? msgUnitArr[0] : null;
	    	String context = "";
	    	if (msgUnit != null) context = "key=" + msgUnit.getKey() + " qos=" + msgUnit.getQos();
	      log.warning("Caught exception on publishOneway which can't be delivered back to client because of 'oneway' mode '" + context + "', message is lost: " + e.getMessage());
      }
   }

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

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

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

         EraseQosServer eraseQosServer = new EraseQosServer(glob, (QueryQosData)msgUnit.getQosData());

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

         sessionInfo.getDispatchStatistic().incrNumErase(1);

         // export (encrypt) return value
         I_Session sec = sessionInfo.getSecuritySession();
         for (int ii=0; ii<retArr.length; ii++) {
            CryptDataHolder dataHolder = new CryptDataHolder(MethodName.ERASE, 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.ERASE, e);
      }
   }

   /**
    * Synchronous access a message.
    * <p />
    * @see org.xmlBlaster.engine.RequestBroker
    */
   public final MsgUnitRaw[] get(AddressServer addressServer, String sessionId,
                                 String xmlKey_literal, String qos_literal) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering get()");

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

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

         // Parse XML key and XML QoS
         GetQosServer getQosServer = new GetQosServer(glob, (QueryQosData)msgUnit.getQosData());

         // Invoke xmlBlaster
         MsgUnit[] msgUnitArr = requestBroker.get(sessionInfo, (QueryKeyData)msgUnit.getKeyData(), getQosServer);

         sessionInfo.getDispatchStatistic().incrNumGet(1);

         // export (encrypt) return value
         MsgUnitRaw[] msgUnitRawArr = new MsgUnitRaw[msgUnitArr.length];
         I_Session sec = sessionInfo.getSecuritySession();
         for (int ii=0; ii<msgUnitArr.length; ii++) {
            CryptDataHolder dataHolder = new CryptDataHolder(MethodName.GET, msgUnitArr[ii].getMsgUnitRaw());
            dataHolder.setReturnValue(true);
            msgUnitRawArr[ii] = sec.exportMessage(dataHolder);
         }

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

   /**
    * Dump state of RequestBroker into a XML ASCII string.
    * <br>
    * @return internal state of the RequestBroker as a XML ASCII string
    */
   public final String toXml() throws XmlBlasterException {
      return requestBroker.toXml();
   }

   /**
    * Dump state of RequestBroker into a XML ASCII string.
    * <br>
    * @param extraOffset indenting of tags for nice output
    * @return internal state of the RequestBroker as a XML ASCII string
    */
   public final String toXml(String extraOffset) throws XmlBlasterException {
      return requestBroker.toXml(extraOffset);
   }

   /**
    * Check message via security plugin.
    * <p/>
    * <ul>
    *   <li>First we import the message using the interceptor</li>
    *   <li>Then we do the authorization</li>
    * </ul>
    * @param sessionInfo The sessionInfo (we are already authenticated)
    * @param addressServer The server side protocol plugin information (like IP and port)
    * @param MsgUnit The message, probably encrypted
    * @param String actionKey (eg. PUBLISH, GET, ...)
    * @return The message decrypted (readable) of type MsgUnit (ready parsed)
    * @exception XmlBlasterException Thrown if seal/signature checks fail, the identity in unknown
    *                                or the message format has errors.<br />
    *            Throws "NotAuthorized" if client may not do the action with this message
    */
   private MsgUnit importAndAuthorize(SessionInfo sessionInfo, AddressServer addressServer,
                       MsgUnitRaw msgUnitRaw, MethodName action) throws XmlBlasterException {
      I_Session sessionSecCtx = sessionInfo.getSecuritySession();
      if (sessionSecCtx==null) { // assert
         throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME+".accessDenied", "unknown session - internal error.");
      }

      // check the message, if it was treated with confidentiality and integrity
      CryptDataHolder ctyptDataHolder = new CryptDataHolder(action, msgUnitRaw, null);
      msgUnitRaw = sessionSecCtx.importMessage(ctyptDataHolder);

      // Parse XML key and XML QoS
      MsgUnit msgUnit = new MsgUnit(glob, msgUnitRaw.getKey(), msgUnitRaw.getContent(), msgUnitRaw.getQos(), action);
      QosData qosData = msgUnit.getQosData();

      // Currently we have misused used the clientProperty to transport this information
      if (qosData.getClientProperty(Constants.PERSISTENCE_ID) != null)
         qosData.isFromPersistenceRecovery(true);

      // Check if server is ready (throws XmlBlasterException otherwise)
      this.availabilityChecker.checkServerIsReady(sessionInfo.getSessionName(), addressServer, msgUnit, action);

      // Protect against faked sender name
      if (sessionInfo.getConnectQos().isClusterNode()) {
         if (qosData.getSender() == null) // In cluster routing don't overwrite the original sender
            qosData.setSender(sessionInfo.getSessionName());
      }
      else {
         if (qosData.getSender() == null) {
            qosData.setSender(sessionInfo.getSessionName());
         }
         else if (!sessionInfo.getSessionName().equalsAbsolute(qosData.getSender())) {
            //if (! publishQos.isFromPersistenceStore()) {
            if (!this.authenticate.isAcceptWrongSenderAddress(sessionInfo)) {
               log.warning(sessionInfo.getId() + " sends message '" + msgUnit.getKeyOid() + "' with invalid sender name '" + qosData.getSender() + "', we fix this");
               qosData.setSender(sessionInfo.getSessionName());
            }
            else {
               log.info(sessionInfo.getId() + " sends message '" + msgUnit.getKeyOid() + "' with invalid sender name '" + qosData.getSender() + "', we accept it");
            }
         }
      }

      /*
      msgUnitRaw = new MsgUnitRaw(
               (msgUnit.getKey().size() > 0) ? sessionSecCtx.importMessage(msgUnit.getKey()) : msgUnit.getKey(),
               (msgUnit.getContent().length > 0) ? sessionSecCtx.importMessage(msgUnit.getContent()) : msgUnit.getContent(),
               (msgUnit.getQos().size() > 0) ? sessionSecCtx.importMessage(msgUnit.getQos()) : msgUnit.getQos());
      */

      // check if this user is permitted to do this action with this message
      I_Subject subjSecCtx = sessionSecCtx.getSubject();
      SessionHolder sessionHolder = new SessionHolder(sessionInfo, addressServer);
      DataHolder dataHolder = new DataHolder(action, msgUnit);
      if (!sessionSecCtx.isAuthorized(sessionHolder, dataHolder)) {
         String key = (msgUnit.getKeyData()==null) ? msgUnit.getKey() : msgUnit.getKeyData().getUrl();
         throw new XmlBlasterException(glob, ErrorCode.USER_SECURITY_AUTHORIZATION_NOTAUTHORIZED, ME,
                       "Subject '" + subjSecCtx.getName() + "' is not permitted to perform action '" + action +
                       "' on key '" + key + "'");
      }

      return msgUnit;
   }

   /**
    * ping xmlBlaster if everything is OK and if xmlBlaster is willing to accept requests.
    * @return "<qos><state id='OK'/></qos>" if we are ready, otherwise the current run level string
    * @see org.xmlBlaster.engine.AvailabilityChecker#getStatus(String)
    */
   public final String ping(AddressServer addressServer, String qos) {
      String ret = "<qos><state id='" + this.availabilityChecker.getStatus(qos) + "'/></qos>";
      if (log.isLoggable(Level.FINER)) log.finer("Entering ping("+qos+"), returning " + ret + " ...");
      return ret;
   }

   /**
    * @todo Call me
    */
   public final void shutdown() {
      this.availabilityChecker.shutdown();
   }
}