socketdriver.java

java开源的企业总线.xmlBlaster

   }

   /**
    * Access the handle to the xmlBlaster authenication core
    */
   I_Authenticate getAuthenticate() {
      return this.authenticate;
   }

   /**
    * Access the handle to the xmlBlaster core
    */
   I_XmlBlaster getXmlBlaster() {
      return this.xmlBlasterImpl;
   }

   AddressServer getAddressServer() {
      return this.addressServer;
   }

   /**
    * Start xmlBlaster SOCKET access.
    * <p />
    * Enforced by interface I_Driver.<br />
    * This method returns as soon as the listener socket is alive and ready or on error.
    * @param glob Global handle to access logging, property and commandline args
    * @param authenticate Handle to access authentication server
    * @param xmlBlasterImpl Handle to access xmlBlaster core
    */
   private synchronized void init(Global glob, AddressServer addressServer, I_Authenticate authenticate, I_XmlBlaster xmlBlasterImpl)
      throws XmlBlasterException
   {
      this.glob = glob;
      this.ME = "SocketDriver" + this.glob.getLogPrefixDashed() + "-" + getType();

      if (log.isLoggable(Level.FINER)) log.finer("Entering init()");
      this.addressServer = addressServer;
      this.authenticate = authenticate;
      this.xmlBlasterImpl = xmlBlasterImpl;

      this.socketUrl = new SocketUrl(glob, this.addressServer);

      if (Constants.COMPRESS_ZLIB_STREAM.equals(this.addressServer.getCompressType())) {
         log.info("Full stream compression enabled with '" + Constants.COMPRESS_ZLIB_STREAM + "' for " + getType());
      }
      else if (Constants.COMPRESS_ZLIB.equals(this.addressServer.getCompressType())) {
         log.info("Message compression enabled with  '" + Constants.COMPRESS_ZLIB + "', minimum size for compression is " + this.addressServer.getMinSize() + " bytes for " + getType());
      }

      if (this.socketUrl.getPort() < 1) {
         log.info("Option protocol/socket/port set to " + this.socketUrl.getPort() + ", socket server not started");
         return;
      }
   }

   /**
    * Activate xmlBlaster access through this protocol.
    */
   public synchronized void activate() throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering activate");

      if (startUdpListener()) {
         listenerUDP = new Thread(new UDPListener());
         listenerUDP.setName("XmlBlaster."+getType()+".udpListener");
         listenerUDP.start();
         while (!listenerReadyUDP) {
            try { Thread.sleep(10); } catch( InterruptedException i) {}
         }
      }

      start(); // Start the listen thread
      while (!listenerReady) {
         try { Thread.sleep(10); } catch( InterruptedException i) {}
      }
   }

   public boolean isActive() {
      return running == true;
   }

   /**
    * Deactivate xmlBlaster access (standby), no clients can connect.
    */
   public synchronized void deActivate() throws RuntimeException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering deActivate");
      running = false; runningUDP = false;

      boolean closeHack = true;
      if (listen != null && closeHack) {
         // On some JDKs, listen.close() is not immediate (has a delay for about 1 sec.)
         // force closing by invoking server with this temporary client:
         try {
            java.net.Socket socket = new Socket(listen.getInetAddress(), this.socketUrl.getPort());
            socket.close();
         } catch (java.io.IOException e) {
            log.warning("Tcp shutdown problem: " + e.toString());
         }
      }
      /*
      if (socketUDP != null && closeHack) {
         // On some JDKs, listen.close() is not immediate (has a delay for about 1 sec.)
         // force closing by invoking server with this temporary client:
         try {
            java.net.DatagramSocket socket = new DatagramSocket(this.socketUrl.getPort(), socketUDP.getLocalAddress());
            socket.close();
         } catch (java.io.IOException e) {
            log.warn(ME, "Udp shutdown problem: " + e.toString());
         }
      }
      */
      if (listen != null) {
         try {
            listen.close();
         } catch (java.io.IOException e) {
            log.warning("TCP socket shutdown problem: " + e.toString());
         }
         listen = null;
         //log.info(ME, "TCP socket driver stopped, all resources released.");
      }
      if (socketUDP != null) {
         socketUDP.close();
         socketUDP = null;
         //log.info(ME, "UDP socket driver stopped, all resources released.");
      }

      // shutdown all clients connected
      while (true) {
         HandleClient h = null;
         synchronized (handleClientSet) {
            Iterator it = handleClientSet.iterator();
            if (it.hasNext()) {
               h = (HandleClient)it.next();
               it.remove();
            }
            else
               break;
         }
         if (h == null)
            break;
         h.shutdown();
      }
      synchronized (handleClientSet) {
        handleClientSet.clear();
      }
      synchronized (handleClientMap) {
        handleClientMap.clear();
      }
   }

   final void removeClient(HandleClient h) {
      synchronized (handleClientSet) {
         boolean removed = handleClientSet.remove(h);
         if (!removed) { // May be called twice: from SessionInfo.shutdown() and from HandleClient->exiting run()
            if (log.isLoggable(Level.FINE)) log.fine("Didn't find a client object to remove: " + h.toString());
         }
      }
      synchronized(handleClientMap) {
         Object removed = handleClientMap.remove(h.getSecretSessionId());
         if (removed == null) {
            if (log.isLoggable(Level.FINE)) log.fine("Didn't find a client handle to remove: " + h.toString());
         }
      }
   }

   final Global getGlobal()
   {
      return this.glob;
   }

   /**
    * Is SSL support switched on?
    */
   public final boolean isSSL()
   {
      if (sslMarker != 0) return sslMarker==-1 ? false : true;

      boolean ssl = this.addressServer.getEnv("SSL", false).getValue();
      sslMarker = ssl ? 1 : -1;
      if (log.isLoggable(Level.FINE)) log.fine(addressServer.getEnvLookupKey("SSL") + "=" + ssl);
      return ssl;
   }

   /**
    * Starts the server socket and waits for clients to connect.
    */
   public void run()
   {
      try {
         int backlog = this.addressServer.getEnv("backlog", 50).getValue(); // queue for max 50 incoming connection request
         if (log.isLoggable(Level.FINE)) log.fine(addressServer.getEnvLookupKey("backlog") + "=" + backlog);


         if (isSSL()) {
             listen = this.socketUrl.createServerSocketSSL(backlog, this.addressServer);
         }
         else {
             listen = new ServerSocket(this.socketUrl.getPort(), backlog, this.socketUrl.getInetAddress());
         }

         log.info("Started successfully " + getType() + " driver on '" + this.socketUrl.getUrl() + "'");
         listenerReady = true;
         while (running) {
            Socket accept = listen.accept();
            if (log.isLoggable(Level.INFO))
            	log.info(ME + ": New incoming request on " + this.socketUrl.getUrl() + " from " + accept.getInetAddress() + ":" + accept.getPort());
            if (!running) {
               log.info("Closing server '" + this.socketUrl.getUrl() + "'");
               break;
            }
            HandleClient hh = new HandleClient(glob, this, accept, socketUDP);
            synchronized (handleClientSet) {
               handleClientSet.add(hh);
            }
         }
      }
      catch (java.net.UnknownHostException e) {
         log.severe("Socket server problem, IP address '" + this.socketUrl.getHostname() + "' is invalid: " + e.toString());
      }
      catch (java.net.BindException e) {
         log.severe("Socket server problem '" + this.socketUrl.getUrl() + "', the port " + this.socketUrl.getPort() + " is not available: " + e.toString());
      }
      catch (java.net.SocketException e) {
         log.info("Socket '" + this.socketUrl.getUrl() + "' closed successfully: " + e.toString());
      }
      catch (IOException e) {
         log.severe("Socket server problem on '" + this.socketUrl.getUrl() + "': " + e.toString());
      }
      catch (Throwable e) {
         log.severe("Socket server problem on '" + this.socketUrl.getUrl() + "': " + e.toString());
         e.printStackTrace();
      }
      finally {
         listenerReady = false;
         if (listen != null) {
            try { listen.close(); } catch (java.io.IOException e) { log.warning("listen.close()" + e.toString()); }
            listen = null;
         }
      }
   }


   /**
    * Close the listener port, the driver shuts down.
    */
   public void shutdown() throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering shutdown");

      try {
         deActivate();
      } catch (Exception e) {
         log.severe(e.toString());
      }

      this.glob.unregisterMBean(this.mbeanHandle);

      this.isShutdown = true;

      log.info("Socket driver '" + getType() + "' stopped, all resources released.");
   }

   public boolean isShutdown() {
      return this.isShutdown;
   }

   /**
    * @return A link for JMX usage
    */
   public java.lang.String getUsageUrl() {
      return Global.getJavadocUrl(this.getClass().getName(), null);
   }

   /* dummy to have a copy/paste functionality in jconsole */
   public void setUsageUrl(java.lang.String url) {
   }

   /**
    * Command line usage.
    * <p />
    * <ul>
    *  <li><i>-plugin/socket/port</i>        The SOCKET web server port [7607]</li>
    *  <li><i>-plugin/socket/hostname</i>    Specify a hostname where the SOCKET web server runs
    *                                          Default is the localhost.</li>
    *  <li><i>-plugin/socket/backlog</i>     Queue size for incoming connection request [50]</li>
    *  <li><i>-dump[socket]</i>       true switches on detailed SOCKET debugging [false]</li>
    * </ul>
    * <p />
    * Enforced by interface I_Driver.
    */
   public String usage()
   {
      String text = "\n";
      text += "SocketDriver options:\n";
      text += "   -"+getEnvPrefix()+"port\n";
      text += "                       The SOCKET server port [7607].\n";
      text += "   -"+getEnvPrefix()+"hostname\n";
      text += "                       Specify a hostname where the SOCKET server runs.\n";
      text += "                       Default is the localhost.\n";
      text += "   -"+getEnvPrefix()+"startUdpListener\n";
      text += "                       Start a UDP datagram listener socket [false].\n";
      text += "   -"+getEnvPrefix()+"useUdpForOneway\n";
      text += "                       Use UDP instead of TCP for updateOneway() calls [false].\n";
      text += "   -"+getEnvPrefix()+"SoTimeout\n";
      text += "                       How long may a socket read block in msec [0] (0 is forever).\n";
      text += "   -"+getEnvPrefix()+"responseTimeout\n";
      text += "                       Max wait for the method return value/exception in msec.\n";
//      text += "                       The default is " +getDefaultResponseTimeout() + ".\n";
      text += "                       Defaults to 'forever', the value to pass is milli seconds.\n";
      text += "   -"+getEnvPrefix()+"backlog\n";
      text += "                       Queue size for incoming connection request [50].\n";
      text += "   -"+getEnvPrefix()+"threadPrio\n";
      text += "                       The priority 1=min - 10=max of the listener thread [5].\n";
      text += "   -"+getEnvPrefix()+"SSL\n";
      text += "                       True enables SSL support on server socket [false].\n";
      text += "   -"+getEnvPrefix()+"keyStore\n";
      text += "                       The path of your keystore file. Use the java utility keytool.\n";
      text += "   -"+getEnvPrefix()+"keyStorePassword\n";
      text += "                       The password of your keystore file.\n";
      text += "   -"+getEnvPrefix()+"compress/type\n";
      text += "                       Valid values are: '', '"+Constants.COMPRESS_ZLIB_STREAM+"', '"+Constants.COMPRESS_ZLIB+"' [].\n";
      text += "                       '' disables compression, '"+Constants.COMPRESS_ZLIB_STREAM+"' compresses whole stream.\n";
      text += "                       '"+Constants.COMPRESS_ZLIB+"' only compresses flushed chunks bigger than 'compress/minSize' bytes.\n";
      text += "   -"+getEnvPrefix()+"compress/minSize\n";
      text += "                       Compress message bigger than given bytes, see above.\n";
      text += "   -dump[socket]       true switches on detailed "+getType()+" debugging [false].\n";
      text += "   " + Global.getJmxUsageLinkInfo(this.getClass().getName(), null);
      text += "\n";
      return text;
   }
}