soapconnection.java

一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。

                        new QName("urn:ProxyService", "GetAdminTopicReply"),
                                  GetAdminTopicReply.class, beanSer, beanSer);

    receiveCall = new Call();
    receiveCall.setSOAPMappingRegistry(mappingReg);
    receiveCall.setTargetObjectURI("urn:ProxyService");
    receiveCall.setMethodName("getReply");
    receiveCall.setEncodingStyleURI(Constants.NS_URI_SOAP_ENC);
  }

  /**
   * Sending a JMS request through the SOAP protocol.
   *
   * @exception IllegalStateException  If the SOAP service fails.
   */
  public synchronized void send(AbstractJmsRequest request)
    throws Exception {
    Hashtable h = request.soapCode();

    // Setting the call's parameters:
    Vector params = new Vector();
    params.addElement(new Parameter("name", String.class, name, null));
    params.add(new Parameter("cnxId", Integer.class,
                             new Integer(cnxId), null));
    params.add(new Parameter("map", Hashtable.class, h, null));
    sendCall.setParams(params);

    // Sending the request, checking the reply:
    try {
      Response resp = sendCall.invoke(serviceUrl,"");

      // Check the response.
      if (resp.generatedFault ()) {
        throw new IllegalStateException("The SOAP service failed to process"
                                        + " the call: "
                                        + resp.getFault().getFaultString());
      }
    } catch (SOAPException exc) {
      throw new IllegalStateException("The SOAP call failed: "
                                      + exc.getMessage());
    }
  }

  /** Closes the <code>SoapConnection</code>. */
  public void close() {}

  /**
   * Actually tries to set a first SOAP connection with the server.
   *
   * @param params  Factory parameters.
   * @param name  The user's name.
   * @param password  The user's password.
   *
   * @exception JMSSecurityException  If the user identification is incorrrect.
   * @exception IllegalStateException  If the SOAP service fails.
   */
  private void connect(FactoryParameters factParams, String name,
                       String password)
               throws JMSException
  {
    // Setting the timer values:
    long startTime = System.currentTimeMillis();
    long endTime = startTime + factParams.connectingTimer * 1000;
    long currentTime;
    long nextSleep = 2000;
    boolean tryAgain;
    int attemptsC = 0;
    Response resp;
    String error;

    try {
      serviceUrl = new URL("http://"
                           + factParams.getHost()
                           + ":"
                           + factParams.getPort()
                           + "/soap/servlet/rpcrouter");
    }
    catch (MalformedURLException exc) {}

    // Building the Call object for checking the user's identification:
    Call checkCall = new Call();
    checkCall.setTargetObjectURI("urn:ProxyService");
    checkCall.setMethodName("setConnection");
    checkCall.setEncodingStyleURI(Constants.NS_URI_SOAP_ENC);

    Vector params = new Vector();
    params.addElement(new Parameter("name", String.class, name, null));
    params.addElement(new Parameter("password", String.class, password, null));
    params.addElement(new Parameter("timeout",
                                    Integer.class,
                                    new Integer(factParams.cnxPendingTimer),
                                    null));

    checkCall.setParams(params);

    while (true) {
      tryAgain = false;
      attemptsC++;
      error = null;

      try {
        resp = checkCall.invoke(serviceUrl,"");

        // SOAP sends a fault back: the service is possibly not started or
        // not running.
        if (resp.generatedFault ()) {
          error = resp.getFault().getFaultString();
          tryAgain = true;
        }
        // RPC call worked:
        else {
          Integer result = (Integer) resp.getReturnValue().getValue();

          // The returned value is either the key of the connection, or -1
          // if the user is invalid:
          if (result.intValue() == -1) {
            throw new JMSSecurityException("Can't open the connection with"
                                           + " the server on host "
                                           + factParams.getHost()
                                           + " and port "
                                           + factParams.getPort()
                                           + ": invalid user identification.");
          }
          cnxId = result.intValue();
          break;
        }
      }
      // SOAP call failed: the server may not be started.
      catch (SOAPException exc) {
        tryAgain = true;
        error = exc.getMessage();
      }
      // Trying again to connect:
      if (tryAgain) {
        currentTime = System.currentTimeMillis();
        // Keep on trying as long as timer is ok:
        if (currentTime < endTime) {

          if (currentTime + nextSleep > endTime)
            nextSleep = endTime - currentTime;

          // Sleeping for a while:
          try {
            Thread.sleep(nextSleep);
          }
          catch (InterruptedException intExc) {}

          // Trying again!
          nextSleep = nextSleep * 2;
          continue;
        }
        // If timer is over, throwing an IllegalStateException:
        else {
          long attemptsT = (System.currentTimeMillis() - startTime) / 1000;
          throw new IllegalStateException("Could not open the connection"
                                          + " with server on host "
                                          + factParams.getHost()
                                          + " and port "
                                          + factParams.getPort()
                                          + " after " + attemptsC
                                          + " attempts during "
                                          + attemptsT + " secs: "
                                          + error);
        }
      }
    }
  }

  public AbstractJmsReply receive() throws Exception {
    Vector params = new Vector();
    params.addElement(new Parameter("name", String.class, name, null));
    params.addElement(new Parameter("cnxId", int.class, new Integer(cnxId), null));    
    receiveCall.setParams(params);
    
    Response resp = null;
    AbstractJmsReply reply = null;
    
    try {
      resp = receiveCall.invoke(serviceUrl, "");
    } catch (SOAPException exc) {
      throw new IOException("The SOAP call failed: " + exc.getMessage());
    }

    if (resp.generatedFault()) {
      throw new IOException("The SOAP service failed to process the call: "
                            + resp.getFault().getFaultString());
    }   
    
    try {
      Hashtable h = (Hashtable) resp.getReturnValue().getValue();
            
      String className = (String) h.get("className");
      Class clazz = Class.forName(className);
      Class [] classParam = { new Hashtable().getClass() };
      Method m = clazz.getMethod("soapDecode",classParam);
      reply = (AbstractJmsReply) m.invoke(null,new Object[]{h});
    } catch (Exception exc) {
      throw new IOException(exc.getMessage());
    }

    return reply;
  }
}