socketbuffer.java

pastry的java实现的2.0b版

   * @param size DESCRIBE THE PARAMETER
   */
  protected void initialize(int size) {
//    System.out.println("SB.initialize("+size+")");
    ebaos = new ExposedByteArrayOutputStream(size);
    o = new ExposedDataOutputStream(ebaos, size);
  }

  /**
   * Will grow the buffer as needed
   *
   * @param msg
   * @param reset DESCRIBE THE PARAMETER
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public void serialize(PRawMessage msg, boolean reset) throws IOException {
    // asdf consider backing with a DataOutputStream to properly handle String
//    boolean done = false;
//    while (!done) {
//      try {
    boolean includeSize = reset;
    if (reset) {
      o.reset();
      ebaos.reset();
    }

    // mark stream location so we can add the size at the beginning
    int o_offset = o.bytesWritten();

    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    // + Payload Length +
    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    // version, header, Reserved, space for the length, which will be filled
    // in later
    if (includeSize) {
      o.write(ZERO, 0, 4);
    }

    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    // + Appl Address (0 is the router)                                +
    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    // + Next Header + Priority + Type (Application specifc) + // zero is
    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+


    address = msg.getDestination();
    o.writeInt(address);

    sender = msg.getSender();
    boolean hasSender = (sender != null);
    o.writeBoolean(hasSender);

    priority = msg.getPriority();
    o.writeByte(priority);

    type = msg.getType();
    o.writeShort(type);

    if (isRouteMessage()) {
      RouteMessage rm = (RouteMessage) msg;
      sendOpts = rm.getOptions();
      rmSubAddress = rm.getAuxAddress();
      rmSubType = rm.getInternalType();
    }

    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    // + NodeHandle sender +
    // + +
    // ... flexable size
    // + +
    // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    if (hasSender) {
      msg.getSender().serialize(o);
    }

    // so we know where to advance the str a few lines down
    int amtToSkip = o.bytesWritten();

    msg.serialize(o);
    o.flush();

    // go back and fill in the size
    int totalSize = ebaos.size();
    buffer = ByteBuffer.wrap(ebaos.buf());
    buffer.clear();
    if (includeSize) {
      buffer.position(o_offset);
      int messageSize = totalSize - 4;
      buffer.putInt(messageSize);
      // messageSize
    }
    // set up the buffer for writing
    buffer.clear();
    buffer.limit(totalSize);

    // to be able to deserialize if we need to reroute
    str = new SocketDataInputStream(
      new ByteArrayInputStream(buffer.array()));

    // don't forget to advance the stream past the header info
    str.skipBytes(amtToSkip);

//        done = true;
//      } catch (Exception e) {
//        if (logger.level <= Logger.WARNING)
//          logger
//              .logException("Growing buffer to " + (buffer.capacity() * 2), e);
//        initialize(buffer.capacity() * 2);
//      } // try
//    } // while
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param md DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public Message deserialize(MessageDeserializer md) throws IOException {
    if (md == null) {
      md = defaultDeserializer;
    }

    if (isRouteMessage()) {
      return RouteMessage.build(str, nhf, spn);
    }

//    try {
    Message m = (Message) md.deserialize(str, type, priority, sender);
//    } catch (IOException ioe) {
//      if (logger.level <= Logger.SEVERE) logger.log("Error deserializing address:"+address+" type:"+type);
//    }
//    System.out.println("SB.deserialize("+m+")");
    return m;
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @return DESCRIBE THE RETURN VALUE
   */
  public String toString() {
//    if (internalMessage != null) return "SocketBuffer ["+internalMessage+"]";
    if (isRouteMessage()) {
      return "SocketBuffer[RouteMessage[" + rmSubType + "@" + rmSubAddress + "]]";
    }
    return "SocketBuffer a:" + address + " t:" + type;
  }


  /**
   * DESCRIBE THE CLASS
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  static class ExposedByteArrayOutputStream extends ByteArrayOutputStream {
    /**
     * Constructor for ExposedByteArrayOutputStream.
     *
     * @param size DESCRIBE THE PARAMETER
     */
    public ExposedByteArrayOutputStream(int size) {
      super(size);
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @return DESCRIBE THE RETURN VALUE
     */
    public byte[] buf() {
      return buf;
    }
  }

  /**
   * DESCRIBE THE CLASS
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  static class ExposedDataOutputStream extends DataOutputStream implements
    OutputBuffer {
    int capacity;

    /**
     * Constructor for ExposedDataOutputStream.
     *
     * @param sub DESCRIBE THE PARAMETER
     * @param capacity DESCRIBE THE PARAMETER
     */
    public ExposedDataOutputStream(OutputStream sub, int capacity) {
      super(sub);
      this.capacity = capacity;
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @return DESCRIBE THE RETURN VALUE
     */
    public int bytesRemaining() {
      return capacity - size();
    }

    /**
     * DESCRIBE THE METHOD
     */
    public void reset() {
      written = 0;
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @return DESCRIBE THE RETURN VALUE
     */
    public int bytesWritten() {
      return written;
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @param v DESCRIBE THE PARAMETER
     * @exception IOException DESCRIBE THE EXCEPTION
     */
    public void writeByte(byte v) throws IOException {
      this.write(v);
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @param v DESCRIBE THE PARAMETER
     * @exception IOException DESCRIBE THE EXCEPTION
     */
    public void writeChar(char v) throws IOException {
      writeChar((int) v);
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @param v DESCRIBE THE PARAMETER
     * @exception IOException DESCRIBE THE EXCEPTION
     */
    public void writeShort(short v) throws IOException {
      writeShort((int) v);
    }
  }

  /**
   * DESCRIBE THE CLASS
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  static class SocketDataInputStream extends DataInputStream implements
    InputBuffer {
    ByteArrayInputStream bais;

    /**
     * Constructor for SocketDataInputStream.
     *
     * @param arg0 DESCRIBE THE PARAMETER
     */
    public SocketDataInputStream(ByteArrayInputStream arg0) {
      super(arg0);
      bais = arg0;
    }

    /**
     * DESCRIBE THE METHOD
     *
     * @return DESCRIBE THE RETURN VALUE
     * @exception IOException DESCRIBE THE EXCEPTION
     */
    public short peakShort() throws IOException {
      bais.mark(2);
      short temp = readShort();
      bais.reset();
      return temp;
    }

    /**
     * I'm not sure this will always work. May need to contain the BAIS, and
     * check it directly.
     *
     * @return DESCRIBE THE RETURN VALUE
     */
    public int bytesRemaining() {
      try {
        return this.available();
      } catch (IOException ioe) {
        ioe.printStackTrace();
      }
      return -1;
    }

  }

}