socketconnection.java

java开源的企业总线.xmlBlaster

         if (!(e instanceof IOException) && !(e instanceof java.net.ConnectException)) e.printStackTrace();
         String str = "Socket client connection to '" + this.socketUrl.getUrl() + "' failed, try options '-dispatch/connection/plugin/socket/hostname <ip> -dispatch/connection/plugin/socket/port <port>' and check if the xmlBlaster server has loaded the socket driver in xmlBlaster.properties";
         throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, str, e);
      }

      if (log.isLoggable(Level.FINE)) log.fine("Created '" + getProtocol() + "' protocol plugin and connect to xmlBlaster server");
   }


   /**
    * Reset the driver on problems
    */
   public void resetConnection()
   {
      if (log.isLoggable(Level.FINE)) log.fine("SocketClient is re-initialized, no connection available");
      try {
         shutdown();
      }
      catch (XmlBlasterException ex) {
         log.severe("disconnect. Could not shutdown properly. " + ex.getMessage());
      }
   }


   /**
    * A string with the local address and port (the client side).
    * @return For example "localhost:66557"
    */
   public SocketUrl getLocalSocketUrl() {
      if (this.localSocketUrl == null) {
         // Happens if on client startup an xmlBlaster server is not available
         if (log.isLoggable(Level.FINE)) log.fine("Can't determine client address, no socket connection available");
         return null;
      }
      return this.localSocketUrl;
   }

   /**
    * @see I_XmlBlasterConnection#setConnectReturnQos(ConnectReturnQos)
    */
   public void setConnectReturnQos(ConnectReturnQos connectReturnQos) {
      this.sessionId = connectReturnQos.getSecretSessionId();
      this.loginName = connectReturnQos.getSessionName().getLoginName();
      this.ME = "SocketConnection-"+loginName;
   }

   /**
    * Login to the server.
    * <p />
    * @param connectQos The encrypted connect QoS
    * @exception XmlBlasterException if login fails
    */
   public String connect(String connectQos) throws XmlBlasterException {
      if (connectQos == null)
         throw new XmlBlasterException(glob, ErrorCode.USER_ILLEGALARGUMENT, ME+".connect()", "Please specify a valid QoS");
      if (log.isLoggable(Level.FINER)) log.finer("Entering connect");
      if (isConnected() && isLoggedIn()) {
         log.warning("You are already logged in, we try again: " + toXml());
         Thread.dumpStack();
         //log.warn(ME, "You are already logged in, no relogin possible.");
         //return "";
      }

      connectLowlevel(this.clientAddress);

      if (getCbReceiver() == null) {
         // SocketCallbackImpl.java must be instantiated first
         //throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME,
         //      "Sorry, SOCKET callback handler is not available but is necessary if client connection is of type 'SOCKET', please do not mix 'SOCKET' with other protocols in the same client connection.");
         log.info("Creating default callback server type=" + getType());
         I_CallbackServer server = glob.getCbServerPluginManager().getPlugin(getType(), getVersion());
         // NOTE: This address should come from the client !!!
         org.xmlBlaster.util.qos.address.CallbackAddress cba = new org.xmlBlaster.util.qos.address.CallbackAddress(glob);
         // TODO: extract the real loginName from connectQos
         server.initialize(this.glob, getLoginName(), cba, this.cbClient);
         // NOTE: This happens only if the client has no callback configured, we create a faked one here (as the SOCKET plugin needs it)
      }

      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.CONNECT, sessionId); // sessionId is usually null on login, on reconnect != null
         parser.setPluginConfig(this.pluginInfo);
         parser.addMessage(connectQos);
         return (String)getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.WAIT_ON_RESPONSE, SocketUrl.SOCKET_TCP);
      }
      catch (XmlBlasterException e) {
         throw e;
      }
      catch (Throwable e) {
         if (!(e instanceof IOException) && !(e instanceof java.net.ConnectException)) e.printStackTrace();
         if (log.isLoggable(Level.FINE)) log.fine(e.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, "login failed", e);
      }
   }

   /**
    * Returns the protocol type.
    * @return "SOCKET"
    */
   public final String getProtocol() {
      return (this.pluginInfo == null) ? "SOCKET" : this.pluginInfo.getType();
   }

    /**
    * Does a logout and removes the callback server.
    * <p />
    * @param sessionId The client sessionId
    */
   public boolean disconnect(String qos) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering logout/disconnect: id=" + sessionId);

      if (!isLoggedIn()) {
         log.warning("You are not logged in, no logout possible.");
         return false;
      }

      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.DISCONNECT, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addMessage((qos==null)?"":qos);
         // We close first the callback thread, this could be a bit early ?
         getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.WAIT_ON_RESPONSE/*ONEWAY*/, SocketUrl.SOCKET_TCP);
         getCbReceiver().running = false; // To avoid error messages as xmlBlaster closes the connection during disconnect()
         return true;
      }
      catch (XmlBlasterException e) {
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, "disconnect", e);
      }
      catch (IOException e1) {
         if (log.isLoggable(Level.FINE)) log.fine("IO exception: " + e1.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, "disconnect", e1);
      }
      finally {
       //  shutdown(); // the callback server
       //  sessionId = null;
      }
   }

   /**
    * Shut down the callback server.
    * Is called by logout()
    */
   public void shutdown() throws XmlBlasterException
   {
      if (log.isLoggable(Level.FINER)) log.finer("Entering shutdown of callback server");
      if (this.cbReceiver != null) {
         this.cbClient = this.cbReceiver.getCbClient(); // remember for reconnects
         this.cbReceiver.shutdownSocket();
      }
      Socket sk = this.sock;
      if (sk != null) {
         try { sk.getInputStream().close();  } catch (IOException e) { log.fine("InputStream.close(): " + e.toString()); }
         try { sk.getOutputStream().close(); } catch (IOException e) { log.fine("OutputStream.close(): " + e.toString()); }
         try { sk.close(); this.sock=null;   } catch (IOException e) { log.warning("socket.close(): " + e.toString()); }
      }
   }

   /**
    * @return true if you are logged in
    */
   public final boolean isLoggedIn()
   {
      return this.sessionId != null;
   }

   /**
    * @return true if the socket connection is established
    */
   public final boolean isConnected()
   {
      return this.sock != null; // && cbReceiver != null
   }

   /**
    * Access handle of callback server.
    * <p />
    * Opens the socket connection if not logged in.
    */
   //public I_CallbackServer getCallbackServer() throws XmlBlasterException
   //{
   //   return getCbReceiver();
   //}

   /**
    * Called by SocketCallbackImpl on creation
    */
   final void registerCbReceiver(SocketCallbackImpl cbReceiver) {
      this.cbReceiver = cbReceiver;
      if (this.tmpProgressListener != null) {
         if (log.isLoggable(Level.FINE)) log.fine("The progressListener will be registered now since it could not register it before");
         this.cbReceiver.registerProgressListener(this.tmpProgressListener);
         this.tmpProgressListener = null;
      }
      if (this.cbReceiver != null) {
         this.cbClient = this.cbReceiver.getCbClient(); // remember for reconnects
      }
   }

   /**
    * Access handle of callback server.
    * <p />
    * Returns the valid SocketCallbackImpl, opens the socket connection if not logged in.
    */
   private final SocketCallbackImpl getCbReceiver() throws XmlBlasterException {
      return this.cbReceiver;
   }

   /**
    * Enforced by I_XmlBlasterConnection interface (failsafe mode).
    * Subscribe to messages.
    * <p />
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/interface.subscribe.html">The interface.subscribe requirement</a>
    */
   public final String subscribe(String xmlKey_literal, String qos_literal) throws XmlBlasterException
   {
      if (log.isLoggable(Level.FINER)) log.finer("Entering subscribe(id=" + sessionId + ")");
      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.SUBSCRIBE, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addKeyAndQos(xmlKey_literal, qos_literal);
         Object response = getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.WAIT_ON_RESPONSE, SocketUrl.SOCKET_TCP);
         return (String)response; // return the QoS
      }
      catch (IOException e1) {
         if (log.isLoggable(Level.FINE)) log.fine("IO exception: " + e1.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, MethodName.SUBSCRIBE.toString(), e1);
      }
   }

   /**
    * Unsubscribe from messages.
    * <p />
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/interface.unSubscribe.html">The interface.unSubscribe requirement</a>
    */
   public final String[] unSubscribe(String xmlKey_literal,
                                 String qos_literal) throws XmlBlasterException
   {
      if (log.isLoggable(Level.FINER)) log.finer("Entering unSubscribe(): id=" + sessionId);
      if (log.isLoggable(Level.FINEST)) log.finest("Entering unSubscribe(): id=" + sessionId + " key='" + xmlKey_literal + "' qos='" + qos_literal + "'");

      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.UNSUBSCRIBE, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addKeyAndQos(xmlKey_literal, qos_literal);
         Object response = getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.WAIT_ON_RESPONSE, SocketUrl.SOCKET_TCP);
         return (String[])response;
      }
      catch (IOException e1) {
         if (log.isLoggable(Level.FINE)) log.fine("IO exception: " + e1.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, MethodName.UNSUBSCRIBE.toString(), e1);