httpnetwork.java

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

   * @param host	the server host.
   * @param port	the server port.
   * @param proxy	the proxy host.
   * @param proxyport	the proxy port.
   * @return		a socket connected to a ServerSocket at the specified
   *			network address and port.
   *
   * @exception IOException	if the connection can't be established
   */
  Socket createTunnelSocket(InetAddress host, int port,
                            InetAddress proxy, int proxyport) throws IOException {
    return createSocket(proxy, proxyport);
  }

  final class NetServerOut extends Daemon {
    Socket socket = null;
    InputStream is = null;
    OutputStream os = null;

    NetServerOut(String name, Logger logmon) {
      super(name + ".NetServerOut");
      // Overload logmon definition in Daemon
      this.logmon = logmon;
    }

    protected void open() throws IOException {
      // Open the connection.
      socket = null;
      boolean phase1 = true;
      while (true) {
        if (proxy == null) {
          try {
            socket = createSocket(server);
            break;
          } catch (IOException exc) {
            logmon.log(BasicLevel.WARN,
                       this.getName() + ", connection refused", exc);
            if (! phase1) throw exc;
            phase1 = false;
            server.resetAddr();
          }
        } else {
          try {
            socket = createTunnelSocket(server.getAddr(), server.getPort(),
                                        proxy, proxyport);
            break;
          } catch (IOException exc) {
            logmon.log(BasicLevel.WARN,
                       this.getName() + ", connection refused", exc);
            if (! phase1) throw exc;
            phase1 = false;
            proxy = InetAddress.getByName(proxyhost);
          }
        }
      }
      setSocketOption(socket);

      os = socket.getOutputStream();
      is = socket.getInputStream();
    }

    protected void close() {
      if (socket != null) {
        try {
          os.close();
        } catch (Exception exc) {}
        try {
          is.close();
        } catch (Exception exc) {}
        try {
          socket.close();
        } catch (Exception exc) {}
      }
    }

    protected void shutdown() {
      thread.interrupt();
      close();
    }

    public void run() {
      Message msgout = null;
      int ack = -1;

      byte[] buf = new byte[120];

      try {
	while (running) {
          canStop = true;
	  try {
	    try {
              if (logmon.isLoggable(BasicLevel.DEBUG))
                logmon.log(BasicLevel.DEBUG,
                           this.getName() + ", waiting message");

              msgout = qout.get(activationPeriod);
	    } catch (InterruptedException exc) {
              if (logmon.isLoggable(BasicLevel.DEBUG))
                logmon.log(BasicLevel.DEBUG,
                           this.getName() + ", interrupted");
	    }
            open();

            do {
              if (logmon.isLoggable(BasicLevel.DEBUG))
                logmon.log(BasicLevel.DEBUG,
                           this.getName() + ", sendRequest: " + msgout + ", ack=" + ack);

              if ((msgout != null) &&(msgout.not.expiration != -1))
                logmon.log(BasicLevel.FATAL,
                           getName() + ": AF YYY " + msgout.not);

              long currentTimeMillis = System.currentTimeMillis();
              do {
                if ((msgout != null) &&
                    (msgout.not.expiration > 0) &&
                    (msgout.not.expiration < currentTimeMillis)) {
                  if (logmon.isLoggable(BasicLevel.DEBUG))
                    logmon.log(BasicLevel.DEBUG,
                               getName() + ": AF removes expired notification XXX " +
                               msgout.from + ", " + msgout.not);
                  //  Suppress the processed notification from message queue,
                  // and deletes it. It can be done outside of a transaction
                  // and commited later (on next handle).
                  qout.removeMessage(msgout);
                  msgout.delete();
                  msgout.free();

                  msgout = qout.get(0L);
                  continue;
                }
                break;
              } while (true);

              sendRequest(msgout, os, ack, currentTimeMillis);
              getReply(is, buf);

              canStop = false;
              ack = handle(msgout);
              canStop = true;
              // Get next message to send if any
              msgout = qout.get(0);
            } while (running && ((msgout != null) || (ack != -1)));
          } catch (Exception exc) {
            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG,
                         this.getName() + ", connection closed", exc);
          } finally {
            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG,
                         this.getName() + ", connection ends");
            try {
              os.close();
            } catch (Exception exc) {}
            os = null;
            try {
              is.close();
            } catch (Exception exc) {}
            is = null;
            try {
              socket.close();
            } catch (Exception exc) {}
            socket = null;
          }
        }
      } finally {
        logmon.log(BasicLevel.WARN, ", exited");
	finish();
      }
    }
  }

  final class NetServerIn extends Daemon {
    ServerSocket listen = null;
    
    Socket socket = null;
    InputStream is = null;
    OutputStream os = null;

    NetServerIn(String name, ServerSocket listen, Logger logmon) throws IOException {
      super(name + ".NetServerIn");
      this.listen = listen;
      // Overload logmon definition in Daemon
      this.logmon = logmon;
    }

    protected void open(Socket socket) throws IOException {
      setSocketOption(socket);

      os = socket.getOutputStream();
      is = socket.getInputStream();

      if (logmon.isLoggable(BasicLevel.DEBUG))
        logmon.log(BasicLevel.DEBUG, this.getName() + ", connected");
    }

    protected void close() {
      if (socket != null) {
        try {
          os.close();
        } catch (Exception exc) {}
        try {
          is.close();
        } catch (Exception exc) {}
        try {
          socket.close();
        } catch (Exception exc) {}
      }
      try {
	listen.close();
      } catch (Exception exc) {}
    }

    protected void shutdown() {
      close();
    }

    public void run() {
      Message msgout= null;
      int ack = -1;

      byte[] buf = new byte[120];

      try {
	while (running) {
          canStop = true;

          // Get the connection
          try {
            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG,
                         this.getName() + ", waiting connection");
            socket = listen.accept();
            open(socket);
            
            short from = getRequest(is, buf);
            long currentTimeMillis = System.currentTimeMillis();
            do {
              canStop = false;
              ack = handle(msgout);
              canStop = true;

              do {
                msgout = qout.getMessageTo(from);

                if ((msgout != null) &&(msgout.not.expiration != -1))
                  logmon.log(BasicLevel.FATAL,
                             getName() + ": AF YYY " + msgout.not);

                if ((msgout != null) &&
                    (msgout.not.expiration > 0) &&
                    (msgout.not.expiration < currentTimeMillis)) {
                  if (logmon.isLoggable(BasicLevel.DEBUG))
                    logmon.log(BasicLevel.DEBUG,
                               getName() + ": AF removes expired notification " +
                               msgout.from + ", " + msgout.not);
                  //  Suppress the processed notification from message queue,
                  // and deletes it. It can be done outside of a transaction
                  // and commited later (on next handle).
                  qout.removeMessage(msgout);
                  msgout.delete();
                  msgout.free();

                  continue;
                }
                break;
              } while (true);
 
              if (logmon.isLoggable(BasicLevel.DEBUG))
                logmon.log(BasicLevel.DEBUG,
                           this.getName() + ", sendReply: " + msgout);

              sendReply(msgout, os, ack, currentTimeMillis);

              logmon.log(BasicLevel.DEBUG,
                         getName() + ": AF WWW " + msgout);

              getRequest(is, buf);
            } while (running);
          } catch (Exception exc) {
            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG, ", connection closed", exc);
          } finally {
            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG, ", connection ends");
            try {
              os.close();
            } catch (Exception exc) {}
            os = null;
            try {
              is.close();
            } catch (Exception exc) {}
            is = null;
            try {
              socket.close();
            } catch (Exception exc) {}
            socket = null;
          }
        }
      } finally {
        logmon.log(BasicLevel.WARN, ", exited");
        finish();
      }
    }
  }

  /**
   * Class used to read messages through a stream.
   */
  final class MessageInputStream extends ByteArrayInputStream {
    MessageInputStream() {
      super(new byte[256]);
    }

    private void readFully(InputStream is, int length) throws IOException {
      count = 0;
      if (length > buf.length) buf = new byte[length];
      
      int nb = -1;
      do {
        nb = is.read(buf, count, length-count);
        if (logmon.isLoggable(BasicLevel.DEBUG))
          logmon.log(BasicLevel.DEBUG, getName() + ", reads:" + nb);
        if (nb < 0) throw new EOFException();
        count += nb;
      } while (count != length);
      pos  = 0;
    }

    int msgLen;
    int msgBoot;
    int msgAck;

    Message msg = null;

    int readFrom(InputStream is) throws Exception {
      readFully(is, 12);
      // Reads message length
      msgLen = ((buf[0] & 0xFF) << 24) + ((buf[1] & 0xFF) << 16) +
        ((buf[2] & 0xFF) <<  8) + ((buf[3] & 0xFF) <<  0);
      // Reads boot timestamp of source server
      msgBoot = ((buf[4] & 0xFF) << 24) + ((buf[5] & 0xFF) << 16) +
        ((buf[6] & 0xFF) <<  8) + ((buf[7] & 0xFF) <<  0);
      msgAck = ((buf[8] & 0xFF) << 24) + ((buf[9] & 0xFF) << 16) +
        ((buf[10] & 0xFF) <<  8) + ((buf[11] & 0xFF) <<  0);

      if (msgLen > (Message.LENGTH +11)) {
        msg = Message.alloc();
        readFully(is, Message.LENGTH);

        int idx = msg.readFromBuf(buf, 0);
        // Reads notification attributes
        boolean persistent = ((buf[idx] & Message.PERSISTENT) == 0)?false:true;
        boolean detachable = ((buf[idx] & Message.DETACHABLE) == 0)?false:true;

        readFully(is, msgLen - (Message.LENGTH +12));
        // Reads notification object
        ObjectInputStream ois = new ObjectInputStream(this);
        msg.not = (Notification) ois.readObject();
        if (msg.not.expiration > 0) {
          msg.not.expiration += System.currentTimeMillis();
        }
        msg.not.persistent = persistent;
        msg.not.detachable = detachable;
        msg.not.detached = false;

        return msgLen;
      }
      msg = null;
      return 12;
    }

    int getLength() {
      return msgLen;
    }

    int getBootTS() {
      return msgBoot;
    }

    int getAckStamp() {
      return msgAck;
    }

    Message getMessage() {
      return msg;
    }
  }

  /**
   * Class used to send messages through a stream.
   */
  final class MessageOutputStream extends ByteArrayOutputStream {
    private ObjectOutputStream oos = null;

    MessageOutputStream() throws IOException {
      super(256);
      oos = new ObjectOutputStream(this);
      count = 0;
        buf[Message.LENGTH +12] = (byte)((ObjectStreamConstants.STREAM_MAGIC >>> 8) & 0xFF);
        buf[Message.LENGTH +13] = (byte)((ObjectStreamConstants.STREAM_MAGIC >>> 0) & 0xFF);
        buf[Message.LENGTH +14] = (byte)((ObjectStreamConstants.STREAM_VERSION >>> 8) & 0xFF);
        buf[Message.LENGTH +15] = (byte)((ObjectStreamConstants.STREAM_VERSION >>> 0) & 0xFF);
    }

    void writeMessage(Message msg, int ack,
                      long currentTimeMillis) throws IOException {
      // Writes boot timestamp of source server
      buf[4] = (byte) (getBootTS() >>>  24);
      buf[5] = (byte) (getBootTS() >>>  16);
      buf[6] = (byte) (getBootTS() >>>  8);
      buf[7] = (byte) (getBootTS() >>>  0);

      // Writes stamp of last received message
      buf[8] = (byte) (ack >>>  24);
      buf[9] = (byte) (ack >>>  16);
      buf[10] = (byte) (ack >>>  8);
      buf[11] = (byte) (ack >>>  0);

      count = 12;
      if (msg != null) {
        int idx = msg.writeToBuf(buf, 12);
        // Writes notification attributes
        buf[idx++] = (byte) ((msg.not.persistent?Message.PERSISTENT:0) |
                             (msg.not.detachable?Message.DETACHABLE:0));
        // Be careful, the stream header is hard-written in buf
        count = (Message.LENGTH + 12 +4);

        try {
          if (msg.not.expiration > 0) {
            msg.not.expiration -= currentTimeMillis;
          }
          oos.writeObject(msg.not);
          oos.reset();
          oos.flush();
        } finally {
          if ((msg.not != null) && (msg.not.expiration > 0)) {
            msg.not.expiration += currentTimeMillis;
          }
        }
      }
      // Writes boot timestamp of source server
      buf[0] = (byte) (count >>>  24);
      buf[1] = (byte) (count >>>  16);
      buf[2] = (byte) (count >>>  8);
      buf[3] = (byte) (count >>>  0);
    }
  }
}