gnurequest.java

gcc的组建

     * checks and any other functionality, if required.
     */
  }

  /**
   * Submit the request, suspending the current thread until the answer is
   * received.
   *
   * This implementation requires to set the IOR property ({@link #setIOR(IOR)}
   * before calling this method.
   *
   * @throws BAD_INV_ORDER, minor code 0, if the IOR has not been previously
   * set.
   *
   * @throws SystemException if this exception has been thrown on remote side.
   * The exact exception type and the minor code are the same as they have been
   * for the exception, thrown on remoted side.
   */
  public synchronized void invoke() throws BAD_INV_ORDER
  {
    waitWhileBusy();
    complete = false;
    running = true;

    if (ior == null)
      throw new BAD_INV_ORDER("Set IOR property first");

    try
      {
        Forwardings:
        while (true)
          {
            try
              {
                p_invoke();
                break Forwardings;
              }
            catch (ForwardRequest e)
              {
                try
                  {
                    ObjectImpl impl = (ObjectImpl) e.forward;
                    SimpleDelegate delegate =
                      (SimpleDelegate) impl._get_delegate();
                    ior = delegate.getIor();
                  }
                catch (Exception ex)
                  {
                    BAD_PARAM bad =
                      new BAD_PARAM("Unsupported forwarding target");
                    bad.initCause(ex);
                    throw bad;
                  }
              }
          }
      }
    finally
      {
        running = false;
        complete = true;
      }
  }

  /** {@inheritDoc} */
  public String operation()
  {
    return m_operation;
  }

  /**
   * Get the orb, related to the invocation target.
   */
  public ORB orb()
  {
    return orb;
  }

  /** {@inheritDoc} */
  public boolean poll_response()
  {
    return complete && !running;
  }

  /** {@inheritDoc} */
  public NamedValue result()
  {
    return m_result;
  }

  /**
   * {@inheritDoc}
   *
   */
  public Any return_value()
  {
    return m_result.value();
  }

  /** {@inheritDoc} */
  public synchronized void send_deferred()
  {
    waitWhileBusy();
    new Thread()
      {
        public void run()
        {
          invoke();
        }
      }.start();
  }

  /**
   * Send a request and forget about it, not waiting for a response. This can be
   * done also for methods that normally are expected to return some values.
   *
   * TODO It is generally recommended to reuse the threads. Reuse?
   */
  public void send_oneway()
  {
    final gnuRequest cloned = Clone();
    cloned.oneWay = true;

    new Thread()
      {
        public void run()
        {
          cloned.invoke();
        }
      }.start();
  }

  /**
   * Set the argument list. This field is initialised as empty non null instance
   * by default, so the method is only used in cases when the direct replacement
   * is desired.
   *
   * @param a_args the argument list.
   */
  public void set_args(NVList a_args)
  {
    if (a_args instanceof gnuNVList)
      m_args = (gnuNVList) a_args;
    else
      {
        try
          {
            // In case if this is another implementation of the NVList.
            m_args.list.clear();
            for (int i = 0; i < a_args.count(); i++)
              {
                m_args.add(a_args.item(i));
              }
          }
        catch (Bounds ex)
          {
            Unexpected.error(ex);
          }
      }
  }

  /**
   * Set the context list that is later returned by the method
   * {@link #contexts()}.
   *
   * @param a_context_list a new context list.
   */
  public void set_context_list(ContextList a_context_list)
  {
    m_context_list = a_context_list;
  }

  /**
   * Set the exception container. This field is initialised as empty non null
   * instance by default, so the method is only used in cases when the direct
   * replacement is desired.
   *
   * @param a_environment the new exception container.
   */
  public void set_environment(Environment a_environment)
  {
    m_environment = a_environment;
  }

  /**
   * Set the list of exceptions. This field is initialised as empty non null
   * instance by default, so the method is only used in cases when the direct
   * replacement is desired.
   *
   * @param a_exceptions a list of exceptions.
   */
  public void set_exceptions(ExceptionList a_exceptions)
  {
    m_exceptions = a_exceptions;
  }

  /**
   * Set the operation name.
   *
   * @param a_operation the operation name.
   */
  public void set_operation(String a_operation)
  {
    m_operation = a_operation;
  }

  /**
   * Set the named value, returned as result. This field is initialised as empty
   * non null instance by default, so the method is only used in cases when the
   * direct replacement is desired.
   *
   * @param a_result the result keeper.
   */
  public void set_result(NamedValue a_result)
  {
    m_result = a_result;
  }

  /**
   * Set the type of the named value, returned as a result. Instantiates a new
   * instance of the result value.
   */
  public void set_return_type(TypeCode returns)
  {
    if (m_result == null || !returns.equal(m_result.value().type()))
      {
        m_result = new gnuNamedValue();
        m_result.value().type(returns);
      }
  }

  /**
   * Set the invocation target.
   *
   * @param a_target the CORBA object for that the method will be invoked.
   */
  public void set_target(org.omg.CORBA.Object a_target)
  {
    m_target = a_target;
  }

  /**
   * Do the actual invocation. This implementation requires to set the IOR
   * property ({@link #setIOR(IOR)} before calling this method.
   * 
   * @throws BAD_INV_ORDER, minor code 0, if the IOR has not been previously set
   * or if the direct argument addition is mixed with the direct argument
   * writing into the output stream.
   * 
   * @return the server response in binary form.
   */
  public synchronized RawReply submit()
    throws ForwardRequest
  {
    gnu.CORBA.GIOP.MessageHeader header = new gnu.CORBA.GIOP.MessageHeader();

    header.setBigEndian(Big_endian);

    // The byte order will be Big Endian by default.
    header.message_type = gnu.CORBA.GIOP.MessageHeader.REQUEST;
    header.version = useVersion(ior.Internet.version);

    RequestHeader rh = header.create_request_header();
    rh.operation = m_operation;
    rh.object_key = ior.key;

    // Update interceptor.
    m_rqh = rh;

    if (m_interceptor != null)
      m_interceptor.send_request(m_info);

    // Prepare the submission.
    BufferedCdrOutput request_part = new BufferedCdrOutput();

    request_part.setOffset(header.getHeaderSize());
    request_part.setVersion(header.version);
    request_part.setCodeSet(CodeSetServiceContext.negotiate(ior.Internet.CodeSets));
    request_part.setOrb(orb);
    request_part.setBigEndian(header.isBigEndian());

    // This also sets the stream encoding to the encoding, specified
    // in the header.
    rh.write(request_part);

    if (m_args != null && m_args.count() > 0)
      {
        write_parameters(header, request_part);

        if (m_parameter_buffer != null)
          throw new BAD_INV_ORDER("Please either add parameters or "
            + "write them into stream, but not both " + "at once.");
      }

    if (m_parameter_buffer != null)
      {
        write_parameter_buffer(header, request_part);
      }

    // Now the message size is available.
    header.message_size = request_part.buffer.size();

    Socket socket = null;

    java.lang.Object key = ior.Internet.host + ":" + ior.Internet.port;

    synchronized (SocketRepository.class)
      {
        socket = SocketRepository.get_socket(key);
      }

    try
      {
        long pause = PAUSE_INITIAL;

        if (socket == null)
          {
            // The BindException may be thrown under very heavy parallel
            // load. For some time, just wait, exceptiong the socket to free.
            Open: for (int i = 0; i < PAUSE_STEPS; i++)
              {
                try
                  {
                    if (orb instanceof OrbFunctional)
                      socket = ((OrbFunctional) orb).socketFactory.createClientSocket(
                        ior.Internet.host, ior.Internet.port);
                    else
                      socket = new Socket(ior.Internet.host, ior.Internet.port);
                    break Open;
                  }
                catch (BindException ex)
                  {
                    try
                      {
                        // Expecting to free a socket via finaliser.
                        System.gc();
                        Thread.sleep(pause);
                        pause = pause * 2;
                        if (pause > PAUSE_MAX)
                          pause = PAUSE_MAX;
                      }
                    catch (InterruptedException iex)
                      {
                      }
                  }
              }
          }

        if (socket == null)
          throw new NO_RESOURCES(ior.Internet.host + ":" + ior.Internet.port
            + " in use");
        socket.setKeepAlive(true);

        OutputStream socketOutput = socket.getOutputStream();

        // Write the message header.
        header.write(socketOutput);

        // Write the request header and parameters (if present).
        request_part.buffer.writeTo(socketOutput);

        socketOutput.flush();
        if (!socket.isClosed())
          {
            MessageHeader response_header = new MessageHeader();
            InputStream socketInput = socket.getInputStream();
            response_header.read(socketInput);

            byte[] r;
            if (orb instanceof OrbFunctional)
              {
                OrbFunctional fo = (OrbFunctional) orb;
                r = response_header.readMessage(socketInput, socket,
                  fo.TOUT_WHILE_READING, fo.TOUT_AFTER_RECEIVING);
              }
            else
              r = response_header.readMessage(socketInput, null, 0, 0);

            return new RawReply(orb, response_header, r);
          }
        else
          return EMPTY;
      }
    catch (IOException io_ex)
      {
        COMM_FAILURE m = new COMM_FAILURE("Unable to open a socket at "
          + ior.Internet.host + ":" + ior.Internet.port, 0xC9,
          CompletionStatus.COMPLETED_NO);
        m.initCause(io_ex);
        throw m;
      }
    finally
      {
        try
          {
            if (socket != null && !socket.isClosed())
              {
                socket.setSoTimeout(OrbFunctional.TANDEM_REQUESTS);
                SocketRepository.put_socket(key, socket);
              }
          }
        catch (IOException scx)
          {
            InternalError ierr = new InternalError();
            ierr.initCause(scx);
            throw ierr;
          }
      }
  }

  /** {@inheritDoc} */
  public org.omg.CORBA.Object target()
  {
    return m_target;
  }

  /**
   * Get the used version. Normally, it is better to respond using the same
   * version as it is specified in IOR, but not above the maximal supported
   * version.
   */
  public Version useVersion(Version desired)
  {
    if (desired.until_inclusive(MAX_SUPPORTED.major, MAX_SUPPORTED.minor))
      return desired;
    else
      return MAX_SUPPORTED;
  }

  /**
   * Wait while the response to request, submitted using
   * {@link #send_deferred()} or {@link #invoke()} (from other thread) is
   * returned.
   *
   * FIXME It is possible to rewrite this using Object.wait() and
   * Object.notify(), but be sure to prepare the test as well.
   */
  public synchronized void waitWhileBusy()
  {
    // Waiting constants.
    long wait = 10;
    long increment = 2;
    long max = 5000;

    while (running)
      {
        try
          {
            Thread.sleep(wait);
            if (wait < max)
              wait = wait * increment;
          }
        catch (InterruptedException ex)
          {
          }
      }
  }

  /**
   * Do actual invocation. This method recursively calls itself if the
   * redirection is detected.
   */
  private void p_invoke()
    throws SystemException, ForwardRequest
  {
    RawReply response = submit();

    if (m_rph == null)
      m_rph = response.header.create_reply_header();

    BufferredCdrInput input = response.getStream();
    input.setOrb(orb);

    m_rph.read(input);