dispatchconnection.java

java开源的企业总线.xmlBlaster

   }

   private boolean isAuthenticationException(XmlBlasterException ex) {
	   ErrorCode code = ex.getErrorCode();
	   if (code == null)
		   return false;
	   return code.isOfType(ErrorCode.USER_SECURITY_AUTHENTICATION);
   }
   
   /**
    * @param toReconnected If true if the connection is OK (it is a transition to reconnected)
    * @param byDispatchConnectionsHandler true if invoked by DispatchConnectionsHandler,
    *        false if invoked by our timer/ping
    * @param The problem, is expected to be not null for toReconnected==false
    * @exception XmlBlasterException If delivery failed
    */
   protected final void handleTransition(boolean byDispatchConnectionsHandler,
                                       Throwable throwable) throws XmlBlasterException {

      boolean toReconnected = (throwable == null) ? true : false;

      XmlBlasterException ex = (throwable == null) ? null : ((throwable instanceof XmlBlasterException) ? (XmlBlasterException)throwable : null);
      ConnectionStateEnum oldState = this.state;
      this.serverAcceptsRequests = (ex == null) ? true : !ex.isErrorCode(ErrorCode.COMMUNICATION_NOCONNECTION_SERVERDENY);
      this.physicalConnectionOk = (ex == null ||
                                   glob.isServerSide() && !ex.isServerSide() ||
                                   !glob.isServerSide() && ex.isServerSide()) ? true : false;
      if (ex != null) {
    	  if (isAuthenticationException(ex)) {
    	      connectionsHandler.toDead(this, ex);
    	      if (byDispatchConnectionsHandler) {
    	         throw ex;
    	      }
    	      else {
    	         // ping timer thread, no sense to throw an exception:
    	      }
    	  }
         if (retry(ex)) {
            toReconnected = false;
         }
         else {
            toReconnected = true;
         }
      }

      if (ex != null) {
         if (ErrorCode.COMMUNICATION_USER_HOLDBACK.equals(ex.getErrorCode())) {
            this.connectionsHandler.getDispatchManager().setDispatcherActive(false);
            log.warning("We have set dispatchActive=false, please activate dispatcher manually when the client side problem is resolved: " + ex.getMessage());
         }
      }

      if (log.isLoggable(Level.FINE)) log.fine(ME + "Connection transition " + oldState.toString() + " -> toReconnected=" + toReconnected + " byDispatchConnectionsHandler=" + byDispatchConnectionsHandler + ": " + ((ex == null) ? "" : ex.toXml()));

      synchronized (this.connectionsHandler.dispatchManager) {
         if (isDead()) {   // ignore, not possible
            if (log.isLoggable(Level.FINE)) log.fine(ME + "Connection transition " + oldState.toString() + " -> " + this.state.toString() +
                      " for " + ME + ": We ignore it: " + ((throwable == null) ? "No throwable" : throwable.toString()));
            if (connectionsHandler.getDispatchManager().isShutdown()) {
               return; // Can happen if DispatchWorker is currently delivering and the client has disconnected in the same time (thus DEAD)
            }
            if (throwable == null) {
               throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALSTATE, ME, "Connection transition " + oldState.toString() + " -> " + this.state.toString());
            }
            //Thread.currentThread().dumpStack();
            throw XmlBlasterException.convert(glob, ME, "Connection transition " + oldState.toString() + " -> " + this.state.toString(), throwable);
         }

         /*
         if (oldState == ConnectionStateEnum.UNDEF) { //startup
            if (toReconnected) {
               this.state = ConnectionStateEnum.ALIVE;
               connectionsHandler.toAlive(this);
            }
            else {
               this.state = ConnectionStateEnum.POLLING;
               connectionsHandler.toPolling(this);
            }
         }
         */

         if (toReconnected && isAlive()) { //everything is ok
            if (this.address.getPingInterval() > 0L) { // respan ping timer (even for native plugins we do a dummy ping)
               // timerKey==null (byDispatchConnectionsHandler==false) -> call from a ping timeout: respan
               //
               // timerKey!=null (byDispatchConnectionsHandler==true)  -> call between pings from successful update() invocation:
               // We don't need to ping directly after a successful invocation
               // so we respan the timer.
               // Probably this slows down on many updates and seldom pings,
               // should we remove the following two lines?
               spanPingTimer(this.address.getPingInterval(), true);
            }
            return;
         }

         if (timerKey != null) {
            this.glob.getPingTimer().removeTimeoutListener(timerKey);
            timerKey = null;
         }

         if (toReconnected /*&& this.serverAcceptsRequests*/ && (isPolling() || isUndef())) {
            this.state = ConnectionStateEnum.ALIVE;
            retryCounter = 0; // success
            log.info("Connection '" + getAddress().getType() + "' transition " + oldState.toString() + " -> " + this.state.toString() + ": Success, " + ME + " connected.");
            if (this.address.getPingInterval() > 0L) // respan ping timer
               timerKey = this.glob.getPingTimer().addTimeoutListener(this, this.address.getPingInterval(), null);
            connectionsHandler.toAlive(this);
            return;
         }

         if (this.address.getRetries() == -1 || retryCounter < this.address.getRetries()) {
            // poll for connection ...
            this.state = ConnectionStateEnum.POLLING;
            retryCounter++;
            if (this.address.getDelay() > 0L) { // respan reconnect poller
               if (log.isLoggable(Level.FINE)) log.fine("Polling for server with delay=" + this.address.getDelay() + " oldState=" + oldState + " retryCounter=" + retryCounter);
               if (!this.physicalConnectionOk) resetConnection();
               // next line would have spanned timeout twice and outside the synchronize (2007-00-05 Michele)
               long realSpanTime = address.getDelay();
               //  timerKey = this.glob.getPingTimer().addTimeoutListener(this, this.address.getDelay(), "poll");
               if (oldState == ConnectionStateEnum.ALIVE || oldState == ConnectionStateEnum.UNDEF) {
                  String str = (throwable != null) ? ": " + throwable.toString() : "";
                  //if (throwable != null) throwable.printStackTrace();
                  log.warning(ME + "Connection transition " + oldState.toString() + " -> " + this.state.toString() + ": " +
                               this.address.getLogId() +
                               " is unaccessible, we poll for it every " + this.address.getDelay() + " msec" + str);
                  if (log.isLoggable(Level.FINE)) log.fine(ME + "Connection transition " + oldState.toString() + " -> " + this.state.toString() + " for " + ME +
                               ": retryCounter=" + retryCounter + ", delay=" + this.address.getDelay() + ", maxRetries=" + this.address.getRetries() + str);
                  connectionsHandler.toPolling(this);
                  // spanPingTimer(400, false); // do one instant try
                  realSpanTime = 400;
               }
               if (log.isLoggable(Level.FINE))
                  log.fine(ME + " Respanning timeout for polling to '" + realSpanTime + "' ms");
               spanPingTimer(realSpanTime, false); 
               if (byDispatchConnectionsHandler)
                  throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION_POLLING, ME, "We are in polling mode, can't handle request. oldState=" + oldState);
               return;
            }
            else {
               if (oldState == ConnectionStateEnum.ALIVE) {
                  resetConnection();
                  String str = (throwable != null) ? ": " + throwable.toString() : "";
                  log.warning("Connection transition " + oldState.toString() + " -> " + this.state.toString() + ": " + this.address.toString() + " is unaccessible" + str);
                  if (log.isLoggable(Level.FINE)) log.fine(ME + "Connection transition " + oldState.toString() + " -> " + this.state.toString() + " for " + ME +
                                 ": retryCounter=" + retryCounter + ", delay=" + this.address.getDelay() + ", maxRetries=" + this.address.getRetries() + str);
               }
            }
         }

         // error giving up ...
         this.state = ConnectionStateEnum.DEAD;
      } // synchronized because of timerKey and status transition

      // error giving up ...
      if (throwable == null) {
         ex = new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION_DEAD, ME,
              "Connection transition " + oldState.toString() + " -> " + this.state.toString() + ": " +
              this.address.toString() + " is unaccessible.", throwable);
      }
      else if (throwable instanceof XmlBlasterException) {
         ex = (XmlBlasterException)throwable;
      }
      else {
         ex = new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "Sending of messge failed", throwable);
      }
      log.warning(ME + "Connection transition " + oldState.toString() + " -> " + this.state.toString() + ": retryCounter=" + retryCounter +
                   ", maxRetries=" + this.address.getRetries());

      connectionsHandler.toDead(this, ex);

      if (byDispatchConnectionsHandler) {
         throw ex;
      }
      else {
         // ping timer thread, no sense to throw an exception:
      }
   }

   /**
    * Stop all remote connections.
    */
   public void shutdown() throws XmlBlasterException {
      this.state = ConnectionStateEnum.DEAD;
      if (log.isLoggable(Level.FINER)) log.finer(ME + "Entering shutdown ...");
      if (this.timerKey != null) {
         this.glob.getPingTimer().removeTimeoutListener(this.timerKey);
         this.timerKey = null;
      }
      retryCounter = 0;
   }

   public final ConnectionStateEnum getState() {
      return this.state;
   }

   public final boolean isUndef() {
      return this.state == ConnectionStateEnum.UNDEF;
   }

   public final boolean isAlive() {
      return this.state == ConnectionStateEnum.ALIVE;
   }

   public final boolean isPolling() {
      return this.state == ConnectionStateEnum.POLLING;
   }

   public final boolean isDead() {
      return this.state == ConnectionStateEnum.DEAD;
   }

   /**
    * 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
    */
   abstract public String toXml(String extraOffset);

   /**
    *
    * @return true if the implementation has to throw an exception in case the ping is invoked when
    * there is an ongoing write or read operation on the same connection.
    * Normally it shall give false on the callback and true on the client side. The client side shall
    * give true because otherwise if such a situation arizes it will never go in polling and no
    * reconnect will be triggered.
    */
   abstract protected boolean forcePingFailure();
}