clientdispatchconnection.java

java开源的企业总线.xmlBlaster

      String[] rawReturnValArr = this.driver.unSubscribe(key, qos); // Invoke remote server

      connectionsHandler.getDispatchStatistic().incrNumUnSubscribe(1);
      
      if (unSubscribeEntry.wantReturnObj()) {
         UnSubscribeReturnQos[] retQosArr = new UnSubscribeReturnQos[rawReturnValArr.length];
         for (int ii=0; ii<rawReturnValArr.length; ii++) {
            if (securityInterceptor != null) { // decrypt return value ...
               CryptDataHolder dataHolder = new CryptDataHolder(MethodName.UNSUBSCRIBE, new MsgUnitRaw(null, (byte[])null, rawReturnValArr[ii]));
               dataHolder.setReturnValue(true);
               String xmlQos = securityInterceptor.importMessage(dataHolder).getQos();
               retQosArr[ii] = new UnSubscribeReturnQos(glob, xmlQos);
            }
         }

         try {
            unSubscribeEntry.setReturnObj(retQosArr);
         }
         catch (Throwable e) {
            log.warning("Can't parse returned unSubscribe value setting to default: " + e.toString());
            for (int ii=0; ii<rawReturnValArr.length; ii++) {
               retQosArr[ii] = new UnSubscribeReturnQos(glob, "<qos/>");
            }
            unSubscribeEntry.setReturnObj(retQosArr);
         }
      }
   }

   /**
    * Encrypt and send a synchronous get request, decrypt the returned data
    */
   private void get(MsgQueueEntry entry) throws XmlBlasterException {
      MsgQueueGetEntry getEntry = (MsgQueueGetEntry)entry;

      String key = getEntry.getGetKey().toXml();
      String qos = getEntry.getGetQos().toXml();
      if (this.securityInterceptor != null) {  // We export/encrypt the message (call the interceptor)
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.GET, new MsgUnitRaw(key, (byte[])null, qos));
         MsgUnitRaw msgUnitRaw = securityInterceptor.exportMessage(dataHolder);
         key = msgUnitRaw.getKey();
         qos = msgUnitRaw.getQos();
         if (log.isLoggable(Level.FINE)) log.fine("Exported/encrypted get request.");
      }
      else {
         log.warning("No session security context, get request is not encrypted");
      }

      MsgUnitRaw[] rawReturnValArr = this.driver.get(key, qos); // Invoke remote server

      connectionsHandler.getDispatchStatistic().incrNumGet(1);
      
      MsgUnit[] msgUnitArr = new MsgUnit[rawReturnValArr.length];
      if (getEntry.wantReturnObj()) {
         for (int ii=0; ii<rawReturnValArr.length; ii++) {
            if (this.securityInterceptor != null) { // decrypt return value ...
               CryptDataHolder dataHolder = new CryptDataHolder(MethodName.GET, rawReturnValArr[ii]);
               dataHolder.setReturnValue(true);
               rawReturnValArr[ii] = securityInterceptor.importMessage(dataHolder);
            }
            // NOTE: We use PUBLISH here instead of GET_RETURN to have the whole MsgUnit stored
            msgUnitArr[ii] = new MsgUnit(glob, rawReturnValArr[ii], MethodName.PUBLISH);
         }

         getEntry.setReturnObj(msgUnitArr);
      }
   }

   /**
    * Encrypt and send a erase request, decrypt the returned data
    */
   private void erase(MsgQueueEntry entry) throws XmlBlasterException {
      MsgQueueEraseEntry eraseEntry = (MsgQueueEraseEntry)entry;

      String key = eraseEntry.getEraseKey().toXml();
      String qos = eraseEntry.getEraseQos().toXml();
      if (securityInterceptor != null) {  // We export/encrypt the message (call the interceptor)
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.ERASE, new MsgUnitRaw(key, (byte[])null, qos));
         MsgUnitRaw msgUnitRaw = securityInterceptor.exportMessage(dataHolder);
         key = msgUnitRaw.getKey();
         qos = msgUnitRaw.getQos();
         if (log.isLoggable(Level.FINE)) log.fine("Exported/encrypted erase request.");
      }
      else {
         log.warning("No session security context, erase request is not encrypted");
      }

      String[] rawReturnValArr = this.driver.erase(key, qos); // Invoke remote server

      connectionsHandler.getDispatchStatistic().incrNumErase(1);
      
      if (eraseEntry.wantReturnObj()) {
         EraseReturnQos[] retQosArr = new EraseReturnQos[rawReturnValArr.length];
         for (int ii=0; ii<rawReturnValArr.length; ii++) {
            if (securityInterceptor != null) { // decrypt return value ...
               CryptDataHolder dataHolder = new CryptDataHolder(MethodName.ERASE, new MsgUnitRaw(null, (byte[])null, rawReturnValArr[ii]));
               dataHolder.setReturnValue(true);
               String xmlQos = securityInterceptor.importMessage(dataHolder).getQos();
               retQosArr[ii] = new EraseReturnQos(glob, xmlQos);
            }
         }

         try {
            eraseEntry.setReturnObj(retQosArr);
         }
         catch (Throwable e) {
            log.warning("Can't parse returned erase value setting to default: " + e.toString());
            for (int ii=0; ii<rawReturnValArr.length; ii++) {
               retQosArr[ii] = new EraseReturnQos(glob, "<qos/>");
            }
            eraseEntry.setReturnObj(retQosArr);
         }
      }
   }

   /**
    * Encrypt and send a connect request, decrypt the returned data
    */
   private void connect(MsgQueueEntry entry) throws XmlBlasterException {
      MsgQueueConnectEntry connectEntry = (MsgQueueConnectEntry)entry;
      if (securityInterceptor != null) {  // We export/encrypt the message (call the interceptor)
         log.fine("TODO: Crypting msg with exportMessage() is not supported for connect() as the server currently can't handle encrypted ConnectQos (for SOCKET see HandleClient.java:234)");
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.CONNECT, new MsgUnitRaw(null, (byte[])null, connectEntry.getConnectQosData().toXml()));
         this.encryptedConnectQos = securityInterceptor.exportMessage(dataHolder).getQos();
         if (log.isLoggable(Level.FINE)) log.fine("Exported/encrypted connect request.");
      }
      else {
         log.warning("No session security context, connect request is not encrypted");
         this.encryptedConnectQos = connectEntry.getConnectQosData().toXml();
      }

      // TODO: pass connectEntry.getConnectQosData().getSender().getLoginName(); as this is used by SOCKET:requestId
      String rawReturnVal = this.driver.connect(this.encryptedConnectQos); // Invoke remote server

      connectionsHandler.getDispatchStatistic().incrNumConnect(1);
      
      if (securityInterceptor != null) { // decrypt return value ...
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.CONNECT, new MsgUnitRaw(null, (byte[])null, rawReturnVal));
         dataHolder.setReturnValue(true);
         rawReturnVal = securityInterceptor.importMessage(dataHolder).getQos();
      }

      try {
         this.connectReturnQos = new ConnectReturnQos(glob, rawReturnVal);
      }
      catch (XmlBlasterException e) {
         log.severe("Can't parse returned connect QoS value '" + rawReturnVal + "': " + e.getMessage());
         throw e;
      }

      if (!connectEntry.getConnectQosData().getSessionName().isSession()) {
         // We need to remember the server side assigned public session id for reconnect polling
         // If do we should probably take a clone:
         //ConnectQos connectQos = new ConnectQos(this.glob, this.connectReturnQos.getData());
         ConnectQosData connectQos = connectEntry.getConnectQosData();
         connectQos.setSessionName(this.connectReturnQos.getSessionName());
         connectQos.getSessionQos().setSecretSessionId(this.connectReturnQos.getSecretSessionId());
         if (securityInterceptor != null) {  // We export/encrypt the message (call the interceptor)
            CryptDataHolder dataHolder = new CryptDataHolder(MethodName.CONNECT, new MsgUnitRaw(null, (byte[])null, connectQos.toXml()));
            this.encryptedConnectQos = securityInterceptor.exportMessage(dataHolder).getQos();
            if (log.isLoggable(Level.FINE)) log.fine("Exported/encrypted connect request.");
         }
         else {
            log.warning("No session security context, connect request is not encrypted");
            this.encryptedConnectQos = connectQos.toXml();
         }
      }

      if (connectEntry.wantReturnObj()) {
         connectEntry.setReturnObj(this.connectReturnQos);
      }
      this.driver.setConnectReturnQos(this.connectReturnQos);
   }

   /**
    * Encrypt and send a disconnect request, decrypt the returned data
    */
   private void disconnect(MsgQueueEntry entry) throws XmlBlasterException {
      MsgQueueDisconnectEntry disconnectEntry = (MsgQueueDisconnectEntry)entry;
      String qos = disconnectEntry.getDisconnectQos().toXml();
      if (securityInterceptor != null) {  // We export/encrypt the message (call the interceptor)
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.DISCONNECT, new MsgUnitRaw(null, (byte[])null, qos));
         qos = securityInterceptor.exportMessage(dataHolder).getQos();
         if (log.isLoggable(Level.FINE)) log.fine("Exported/encrypted disconnect request.");
      }
      else {
         log.warning("No session security context, disconnect request is not encrypted");
      }

      //returns void
      this.driver.disconnect(qos); // Invoke remote server
   }

   /**
    * @see org.xmlBlaster.util.dispatch.DispatchConnection#doPing(String)
    */
   public final String doPing(String data) throws XmlBlasterException {
      String ret = driver.ping(data);
      return (ret==null) ? "" : ret;
   }

   /**
    * Nothing to do here
    */
   public final void resetConnection() {
      if (log.isLoggable(Level.FINE)) log.fine("resetConnection(): Initializing driver for polling");
      this.connectReturnQos = null;
      this.driver.resetConnection();
   }

   /**
    * On reconnect polling try to establish the connection. 
    */
   protected final void reconnect() throws XmlBlasterException {
      if (this.driver == null) return;
      if (log.isLoggable(Level.FINER)) log.finer("Entering reconnect(" + this.driver.getProtocol() + ")");

      if (this.connectReturnQos != null) {
         // needed to avoid failure
         this.connectionsHandler.getDispatchStatistic().clearCurrentReads();
         this.connectionsHandler.getDispatchStatistic().clearCurrentWrites();
         super.ping("", false);
         return;
      }

      if (this.encryptedConnectQos == null) {
         // We never had connected on application layer, so try low level layer only
         this.driver.connectLowlevel((Address)super.address);
         return;
      }

      // low level connect (e.g. on TCP/IP layer) and remote invoke method connect()
      String rawReturnVal = this.driver.connect(this.encryptedConnectQos); // Invoke remote server

      connectionsHandler.getDispatchStatistic().incrNumConnect(1);
      
      if (securityInterceptor != null) { // decrypt return value ...
         CryptDataHolder dataHolder = new CryptDataHolder(MethodName.CONNECT, new MsgUnitRaw(null, (byte[])null, rawReturnVal));
         dataHolder.setReturnValue(true);
         rawReturnVal = securityInterceptor.importMessage(dataHolder).getQos();
      }

      this.connectReturnQos = null;
      try {
         this.connectReturnQos = new ConnectReturnQos(glob, rawReturnVal);
         if (this.connectEntry != null) {
            if (this.connectEntry.wantReturnObj()) {
               this.connectEntry.setReturnObj(this.connectReturnQos);
            }
            connectionsHandler.getDispatchManager().postSendNotification(this.connectEntry);
         }
      }
      catch (XmlBlasterException e) {
         log.severe("reconnect(): Can't parse returned connect QoS value '" + rawReturnVal + "': " + e.getMessage());
         throw e;
      }
      this.driver.setConnectReturnQos(this.connectReturnQos);
   }

   /**
    * Stop all callback drivers of this client.
    */
   public final void shutdown() throws XmlBlasterException {
      super.shutdown();
      if (driver != null) {
         driver.shutdown();
      }
   }

   /**
    * Dump state of this object into a XML ASCII string.
    * <br>
    * @param extraOffset indenting of tags for nice output
    * @return internal state as an XML ASCII string
    */
   public final String toXml(String extraOffset) {
      StringBuffer sb = new StringBuffer(256);
      if (extraOffset == null) extraOffset = "";
      String offset = Constants.OFFSET + extraOffset;

      sb.append(offset + "<ClientDispatchConnection>");
      super.address.toXml(" " + offset);
      if (driver == null)
         sb.append(offset).append(" <noProtocolDriver />");
      else
         sb.append(offset).append(" <address type='" + driver.getProtocol() + "' state='" + getState() + "'/>");
      sb.append(offset).append("</ClientDispatchConnection>");

      return sb.toString();
   }
   
   public I_ProgressListener registerProgressListener(I_ProgressListener listener) {
      if (this.driver == null) return null;
      return this.driver.registerProgressListener(listener);
   }

   protected boolean forcePingFailure() {
      return true;
   }
   
}