socketconnection.java

java开源的企业总线.xmlBlaster

      }
   }

   /**
    * Publish a message.
    * The normal publish is handled here like a publishArr
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/interface.publish.html">The interface.publish requirement</a>
    */
   public final String publish(MsgUnitRaw msgUnit) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering publish(): id=" + sessionId);

      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.PUBLISH, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addMessage(msgUnit);
         Object response = getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.WAIT_ON_RESPONSE, SocketUrl.SOCKET_TCP);
         String[] arr = (String[])response; // return the QoS
         return arr[0]; // 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.PUBLISH.toString(), e1);
      }
   }

   /**
    * Publish multiple messages in one sweep.
    * <p />
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/interface.publish.html">The interface.publish requirement</a>
    */
   public final String[] publishArr(MsgUnitRaw[] msgUnitArr) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering publishArr: id=" + sessionId);

      if (msgUnitArr == null) {
         if (log.isLoggable(Level.FINE)) log.fine("The argument of method publishArr() are invalid");
         throw new XmlBlasterException(glob, ErrorCode.USER_ILLEGALARGUMENT, ME + ".InvalidArguments",
                                       "The argument of method publishArr() are invalid");
      }
      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.PUBLISH, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addMessage(msgUnitArr);
         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, "publishArr", e1);
      }
   }

   /**
    * Publish multiple messages in one sweep.
    * <p />
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/interface.publish.html">The interface.publish requirement</a>
    */
   public final void publishOneway(MsgUnitRaw[] msgUnitArr) throws XmlBlasterException {
      if (log.isLoggable(Level.FINER)) log.finer("Entering publishOneway: id=" + sessionId);

      if (msgUnitArr == null) {
         if (log.isLoggable(Level.FINE)) log.fine("The argument of method publishOneway() are invalid");
         return;
      }

      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.PUBLISH_ONEWAY, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addMessage(msgUnitArr);
         getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.ONEWAY, this.useUdpForOneway);
      }
      catch (Throwable e) {
         if (log.isLoggable(Level.FINE)) log.fine("Sending of oneway message failed: " + e.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, MethodName.PUBLISH_ONEWAY.toString(), e);
      }
   }

   /*
   public final String[] sendUpdate(MsgUnitRaw[] msgUnitArr)
      throws XmlBlasterException
      see HandleClient.java
   */


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

      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.ERASE, 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 TODO
      }
      catch (IOException e1) {
         if (log.isLoggable(Level.FINE)) log.fine("IO exception: " + e1.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, MethodName.ERASE.toString(), e1);
      }
   }

   /**
    * Synchronous access a message.
    * <p />
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/interface.get.html">The interface.get requirement</a>
    */
   public final MsgUnitRaw[] get(String xmlKey_literal,
                                  String qos_literal) throws XmlBlasterException
   {
      if (log.isLoggable(Level.FINER)) log.finer("Entering get() xmlKey=\n" + xmlKey_literal + ") ...");
      try {
         MsgInfo parser = new MsgInfo(glob, MsgInfo.INVOKE_BYTE, MethodName.GET, sessionId);
         parser.setPluginConfig(this.pluginInfo);
         parser.addKeyAndQos(xmlKey_literal, qos_literal);
         Object response = getCbReceiver().requestAndBlockForReply(parser, SocketExecutor.WAIT_ON_RESPONSE, SocketUrl.SOCKET_TCP);
         return (MsgUnitRaw[])response;
      }
      catch (IOException e1) {
         if (log.isLoggable(Level.FINE)) log.fine("IO exception: " + e1.toString());
         throw new XmlBlasterException(glob, ErrorCode.COMMUNICATION_NOCONNECTION, ME, MethodName.GET.toString(), e1);
      }
   }

   /**
    * Check server.
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/src/java/org/xmlBlaster/protocol/corba/xmlBlaster.idl" target="others">CORBA xmlBlaster.idl</a>
    */
   public String ping(String qos) throws XmlBlasterException {
      SocketCallbackImpl receiver = getCbReceiver();
      if (receiver == null) {
         return Constants.RET_OK; // fake a return for ping on startup
         /*
         // 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(ME, "Creating default callback server type=" + getType());
         I_CallbackServer server = glob.getCbServerPluginManager().getPlugin(getType(), getVersion());
         server.initialize(this.glob, getLoginName(), null);
         // NOTE: This happens only if the client has no callback configured, we create a faked one here (as the SOCKET plugin needs it)
         */
      }
      return receiver.ping(qos);
   }

   /**
    * Dump of the state, remove in future.
    */
   public String toXml() throws XmlBlasterException
   {
      return toXml("");
   }

   /**
    * Dump of the state, remove in future.
    */
   public String toXml(String extraOffset) throws XmlBlasterException
   {
      if (this.sock == null) return "<noConnection />";
      else return "<connected/>";
   }

   /**
    * Register a listener for to receive information about the progress of incoming data.
    * Only one listener is supported, the last call overwrites older calls.
    * @param listener Your listener, pass 0 to unregister.
    * @return The previously registered listener or 0
    */
   public I_ProgressListener registerProgressListener(I_ProgressListener listener) {
      SocketCallbackImpl cbRec = this.cbReceiver;
      if (cbRec != null)
         return this.cbReceiver.registerProgressListener(listener);
      else {
         if (log.isLoggable(Level.FINE)) log.fine("The callback receiver is null, will be registered when the callback receiver is registered.");
         I_ProgressListener ret = this.tmpProgressListener;
         this.tmpProgressListener = listener;
         return ret;
      }
   }


   /**
    * Command line usage.
    * <p />
    *  <li>-dispatch/connection/plugin/socket/port
    *                      Specify a port number where xmlBlaster SOCKET server listens
    *                      Default is port "+DEFAULT_SERVER_PORT+", the port 0 switches this feature off</li>
    *  <li>-dispatch/connection/plugin/socket/hostname
    *                      Specify a hostname where the xmlBlaster web server runs.
    *                      Default is the localhost</li>
    *  <li>-dispatch/connection/plugin/socket/localPort
    *                      You can specify our client side port as well (usually you shouldn't)
    *                      Default is that the port is chosen by the operating system</li>
    *  <li>-dispatch/connection/plugin/socket/localHostname
    *                      Specify the hostname who we are. Makes sense for multi homed computers
    *                      Defaults to our hostname</li>
    *  <li>-dispatch/connection/plugin/socket/responseTimeout  How long to wait for a method invocation to return
    *                      Defaults to 'forever', the value to pass is milli seconds</li>
    *  <li>-dispatch/connection/plugin/socket/multiThreaded Use seperate threads per update() on client side [true]</li>
    *  <li>-dump[socket]   true switches on detailed SOCKET debugging [false]</li>
    * <p />
    * These variables may be set in xmlBlaster.properties as well.
    * Don't use the "-" prefix there.
    */
   public static String usage()
   {
      String text = "\n";
      text += "SocketConnection 'SOCKET' options:\n";
      text += "   -dispatch/connection/plugin/socket/port\n";
      text += "                       Specify a port number where xmlBlaster SOCKET server listens.\n";
      text += "                       Default is port "+SocketUrl.DEFAULT_SERVER_PORT+", the port 0 switches this feature off.\n";
      text += "   -dispatch/connection/plugin/socket/hostname\n";
      text += "                       Specify a hostname where the xmlBlaster web server runs.\n";
      text += "                       Default is the localhost.\n";
      text += "   -dispatch/connection/plugin/socket/localPort\n";
      text += "                       You can specify our client side port as well (usually you shouldn't)\n";
      text += "                       Default is that the port is chosen by the operating system.\n";
      text += "   -dispatch/connection/plugin/socket/localHostname\n";
      text += "                       Specify the hostname who we are. Makes sense for multi homed computers.\n";
      text += "                       Defaults to our hostname.\n";
      text += "   -dispatch/connection/plugin/socket/responseTimeout\n";
      text += "                       How long to wait for a method invocation to return.\n";
//    text += "                       The default is " +getDefaultResponseTimeout() + ".\n";
      text += "                       Defaults to 'forever', the value to pass is milli seconds.\n";
      text += "   -dispatch/connection/plugin/socket/multiThreaded\n";
      text += "                       Use seperate threads per update() on client side [true].\n";
      text += "   -dispatch/connection/plugin/socket/SSL\n";
      text += "                       True enables SSL support on server socket [false].\n";
      text += "   -dispatch/connection/plugin/socket/trustStore\n";
      text += "                       The path of your trusted keystore file. Use the java utility keytool.\n";
      text += "   -dispatch/connection/plugin/socket/trustStorePassword\n";
      text += "                       The password of your trusted keystore file.\n";
      text += "   -dispatch/connection/plugin/socket/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 += "   -dispatch/connection/plugin/socket/compress/minSize\n";
      text += "                       Compress message bigger than given bytes, see above.\n";
      text += "   -dump[socket]       true switches on detailed SOCKET debugging [false].\n";
      text += "\n";
      return text;
   }
}