import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public synchronized Object unmarshal(Buffer sequence) throws IOException {
    bytesIn.restart(sequence);
    // DataByteArrayInputStreamStream dis = new DataByteArrayInputStreamStream(new
    // ByteArrayInputStream(sequence));

    if (!sizePrefixDisabled) {
        int size = bytesIn.readInt();
        if (sequence.getLength() - 4 != size) {
            // throw new IOException("Packet size does not match marshaled
            // size");
        }

        if (size > maxFrameSize) {
            throw new IOException(
                    "Frame size of " + (size / (1024 * 1024)) +
                    " MB larger than max allowed " +
                            (maxFrameSize / (1024 * 1024)) + " MB");
        }
    }

    Object command = doUnmarshal(bytesIn);
    // if( !cacheEnabled && ((DataStructure)command).isMarshallAware() ) {
    // ((MarshallAware) command).setCachedMarshalledForm(this, sequence);
    // }
    return command;
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public int tightMarshalNestedObject1(DataStructure o, BooleanStream bs) throws IOException {
    bs.writeBoolean(o != null);
    if (o == null) {
        return 0;
    }

    if (o.isMarshallAware()) {
        // MarshallAware ma = (MarshallAware)o;
        Buffer sequence = null;
        // sequence=ma.getCachedMarshalledForm(this);
        bs.writeBoolean(sequence != null);
        if (sequence != null) {
            return 1 + sequence.getLength();
        }
    }

    byte type = o.getDataStructureType();
    DataStreamMarshaller dsm = (DataStreamMarshaller) dataMarshallers[type & 0xFF];
    if (dsm == null) {
        throw new IOException("Unknown data type: " + type);
    }
    return 1 + dsm.tightMarshal1(this, o, bs);
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public void looseMarshalNestedObject(DataStructure o, DataByteArrayOutputStream dataOut) throws IOException {
    dataOut.writeBoolean(o != null);
    if (o != null) {
        if( o instanceof Message) {
            if( !isTightEncodingEnabled() && !isCacheEnabled() ) {
                CachedEncodingTrait encoding = ((Message) o).getCachedEncoding();
                if( encoding !=null && !encoding.tight() && encoding.version()==getVersion()) {
                    Buffer buffer = encoding.buffer();
                    dataOut.write(buffer.data, buffer.offset + 4, buffer.length() - 4);
                    return;
                }
            }
        }
        byte type = o.getDataStructureType();
        dataOut.writeByte(type);
        DataStreamMarshaller dsm = (DataStreamMarshaller) dataMarshallers[type & 0xFF];
        if (dsm == null) {
            throw new IOException("Unknown data type: " + type);
        }
        dsm.looseMarshal(this, o, dataOut);
    }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
private void storeContent() {
    try {
        if (dataOut != null) {
            dataOut.close();
            Buffer bs = bytesOut.toBuffer();
            if (compressed) {
                int pos = bs.offset;
                bs.offset = 0;
                BufferEditor e = BufferEditor.big(bs);
                e.writeInt(length);
                bs.offset = pos;
            }
            setContent(bs);
            bytesOut = null;
            dataOut = null;
        }
    } catch (IOException ioe) {
        throw new RuntimeException(ioe.getMessage(), ioe); // TODO verify
                                                            // RuntimeException
    }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
private Listener connectionListener() {
	return new Listener() {

		@Override
		public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
			// nothing to do
		}

		@Override
		public void onFailure(Throwable value) {
			// nothing to do
		}

		@Override
		public void onDisconnected() {
			fireConnectionLost();
		}

		@Override
		public void onConnected() {
			fireReconnected();
		}
	};
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
@Test
public void testFlush() throws Exception {
	final Buffer value = Buffer.ascii("TESTDATA");
	codec.write(value);
	final int bytesThatWillBeWritten = value.length();
	expect(writableByteChannel.write((ByteBuffer) anyObject())).andAnswer(createWriteAnswer(bytesThatWillBeWritten));
	replay(writableByteChannel);

	final BufferState state = codec.flush();

	assertEquals(BufferState.EMPTY, state);
	assertEquals(false, codec.full());
	assertEquals(true, codec.empty());
	assertEquals(bytesThatWillBeWritten, codec.getWriteCounter());

	assertEquals(BufferState.WAS_EMPTY, codec.flush());
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public Exchange get(final CamelContext camelContext, final String key) {
    Exchange answer = null;

    try {
        byte[] lDbKey = keyBuilder(repositoryName, key);
        LOG.trace("Getting key index {}", key);
        byte[] rc = levelDBFile.getDb().get(lDbKey);

        if (rc != null) {
            answer = codec.unmarshallExchange(camelContext, new Buffer(rc));
        }
    } catch (IOException e) {
        throw new RuntimeException("Error getting key " + key + " from repository " + repositoryName, e);
    }

    LOG.debug("Getting key  [{}] -> {}", key, answer);
    return answer;
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public Exchange recover(CamelContext camelContext, final String exchangeId) {
    Exchange answer = null;

    try {
        byte[] completedLDBKey = keyBuilder(getRepositoryNameCompleted(), exchangeId);

        byte[] rc = levelDBFile.getDb().get(completedLDBKey);

        if (rc != null) {
            answer = codec.unmarshallExchange(camelContext, new Buffer(rc));
        }
    } catch (IOException e) {
        throw new RuntimeException("Error recovering exchangeId " + exchangeId + " from repository " + repositoryName, e);
    }

    LOG.debug("Recovering exchangeId [{}] -> {}", exchangeId, answer);
    return answer;
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public Buffer marshallExchange(CamelContext camelContext, Exchange exchange, boolean allowSerializedHeaders) throws IOException {
    DataByteArrayOutputStream baos = new DataByteArrayOutputStream();
    // use DefaultExchangeHolder to marshal to a serialized object
    DefaultExchangeHolder pe = DefaultExchangeHolder.marshal(exchange, false, allowSerializedHeaders);
    // add the aggregated size and timeout property as the only properties we want to retain
    DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_SIZE, exchange.getProperty(Exchange.AGGREGATED_SIZE, Integer.class));
    DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_TIMEOUT, exchange.getProperty(Exchange.AGGREGATED_TIMEOUT, Long.class));
    // add the aggregated completed by property to retain
    DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_COMPLETED_BY, exchange.getProperty(Exchange.AGGREGATED_COMPLETED_BY, String.class));
    // add the aggregated correlation key property to retain
    DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_CORRELATION_KEY, exchange.getProperty(Exchange.AGGREGATED_CORRELATION_KEY, String.class));
    // and a guard property if using the flexible toolbox aggregator
    DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_COLLECTION_GUARD, exchange.getProperty(Exchange.AGGREGATED_COLLECTION_GUARD, String.class));
    // persist the from endpoint as well
    if (exchange.getFromEndpoint() != null) {
        DefaultExchangeHolder.addProperty(pe, "CamelAggregatedFromEndpoint", exchange.getFromEndpoint().getEndpointUri());
    }
    exchangeCodec.encode(pe, baos);
    return baos.toBuffer();
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public SortedIndex<Buffer, Buffer> getRepositoryIndex(Transaction tx, String name, boolean create) {
    SortedIndex<Buffer, Buffer> answer = null;

    SortedIndex<String, Integer> indexes = ROOT_INDEXES_FACTORY.open(tx);
    Integer location = indexes.get(name);

    if (create && location == null) {
        // create it..
        SortedIndex<Buffer, Buffer> created = INDEX_FACTORY.create(tx);
        int page = created.getIndexLocation();

        // add it to indexes so we can find it the next time
        indexes.put(name, page);

        LOG.debug("Created new repository index with name {} at location {}", name, page);

        answer = created;
    } else if (location != null) {
        LOG.trace("Repository index with name {} at location {}", name, location);
        answer = INDEX_FACTORY.open(tx, location);
    }

    LOG.trace("Repository index with name {} -> {}", name, answer);
    return answer;
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public void confirm(final CamelContext camelContext, final String exchangeId) {
    LOG.debug("Confirming exchangeId [{}]", exchangeId);
    try {
        final Buffer confirmKeyBuffer = codec.marshallKey(exchangeId);
        hawtDBFile.execute(new Work<Buffer>() {
            public Buffer execute(Transaction tx) {
                SortedIndex<Buffer, Buffer> indexCompleted = hawtDBFile.getRepositoryIndex(tx, getRepositoryNameCompleted(), true);
                Buffer buffer = indexCompleted.remove(confirmKeyBuffer);
                LOG.trace("Removed confirm index {} -> {}", confirmKeyBuffer, buffer);
                return buffer;
            }

            @Override
            public String toString() {
                return "Confirming exchangeId [" + exchangeId + "]";
            }
        });

    } catch (IOException e) {
        throw new RuntimeException("Error confirming exchangeId " + exchangeId + " from repository " + repositoryName, e);
    }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
private int size(final String repositoryName) {
    int answer = hawtDBFile.execute(new Work<Integer>() {
        public Integer execute(Transaction tx) {
            SortedIndex<Buffer, Buffer> index = hawtDBFile.getRepositoryIndex(tx, repositoryName, false);
            return index != null ? index.size() : 0;
        }

        @Override
        public String toString() {
            return "Size[" + repositoryName + "]";
        }
    });

    LOG.debug("Size of repository [{}] -> {}", repositoryName, answer);
    return answer;
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
@Override
protected void processCommand(ByteBuffer plain) throws Exception {

    byte[] fill = new byte[plain.remaining()];
    plain.get(fill);

    ByteBuffer payload = ByteBuffer.wrap(fill);

    if (magic.position() != 8) {

        while (payload.hasRemaining() && magic.position() < 8) {
            magic.put(payload.get());
        }

        if (!magic.hasRemaining()) {
            magic.flip();
            doConsume(new AmqpHeader(new Buffer(magic)));
            magic.position(8);
        }
    }

    if (payload.hasRemaining()) {
        doConsume(AmqpSupport.toBuffer(payload));
    }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
public Object unmarshal(DataInput dataIn) throws IOException {
    if( !magicRead ) {
        Buffer magic = new Buffer(8);
        magic.readFrom(dataIn);
        magicRead = true;
        return new AmqpHeader(magic);
    } else {
        int size = dataIn.readInt();
        if( size > maxFrameSize) {
            throw new AmqpProtocolException("Frame size exceeded max frame length.");
        }
        Buffer frame = new Buffer(size);
        frame.bigEndianEditor().writeInt(size);
        frame.readFrom(dataIn);
        frame.clear();
        return frame;
    }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
protected EncodedMessage readIncomingMessage(Delivery incoming) {
    Buffer buffer;
    int count;

    while ((count = endpoint.recv(incomingBuffer, 0, incomingBuffer.length)) > 0) {
        streamBuffer.write(incomingBuffer, 0, count);
    }

    buffer = streamBuffer.toBuffer();

    try {
        return new EncodedMessage(incoming.getMessageFormat(), buffer.data, buffer.offset, buffer.length);
    } finally {
        streamBuffer.reset();
    }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
@Override
    public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
        String message = body.utf8().toString();
        this.onMsg.send(message);
        ack.run();
        Message msg = new Message();
        msg.topic = topic.toString();
        msg.message = message;
        try {
            Gson gson = new Gson();
            String jsonMsg = gson.toJson(msg);
            this.onTopicAndMsg.send(jsonMsg);
        } catch (Exception e) {
            Log.error("{} unable to serialize topic and message to JSON, onTopicAndMsg port will not be used");
//            e.printStackTrace();
        }
    }
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
private void initXPathEvaluator() {
    XPathExpression.XPATH_EVALUATOR_FACTORY = new XPathExpression.XPathEvaluatorFactory() {

        @Override
        public XPathExpression.XPathEvaluator create(String xpath) {
            return new XalanXPathEvaluator(xpath) {
                @Override
                public boolean evaluate(Filterable m) throws FilterException {
                    Buffer body = m.getBodyAs(Buffer.class);
                    if (body != null) {
                        return evaluate(new InputSource(new BufferInputStream(body)));
                    } else {
                        return super.evaluate(m);
                    }
                }
            };
        }
    };
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
/** *********************************************************************************************************
 *
 *  Inbound messages
 *
 ** *********************************************************************************************************/
@Override
public void onMessage(String message) {
    if (this.firstMessage) {
        // if the first message the client sends us is a text message (as denoted by this callback method
        // being called), then we will respond with text messages
        binaryTransfers = false;
        firstMessage = false;
    }

    // convert the string message to bytes message.. our codecs just work with bytes
    Buffer buffer = new AsciiBuffer(message);

    // delegate to the callback the deals with bytes
    onMessage(buffer.getData(), buffer.getOffset(), buffer.getLength());

}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
/**
 * Converts the body of the given message into the specified type. If the type is not specified in the
 * {@link InboundMapping}, it returns the body as it is. However, if this payload is a Hawtbuf Buffer an automatic
 * conversion to Vert.x buffer is done.
 *
 * @param inbound the inbound mapping configuration
 * @param msg     the message from camel
 * @return the body to be sent on the event bus, the type of the returned object depends on the mapping
 * configuration.
 */
static Object convert(InboundMapping inbound, Message msg) {
  if (inbound.getBodyType() != null) {
    return msg.getBody(inbound.getBodyType());
  } else {
    Object body = msg.getBody();
    if (body instanceof org.fusesource.hawtbuf.Buffer) {
      // Map to Vert.x buffers.
      return io.vertx.core.buffer.Buffer.buffer(((Buffer) body).toByteArray());
    }
    return body;
  }
}
 

import org.fusesource.hawtbuf.Buffer; //导入依赖的package包/类
@SuppressWarnings("deprecation")
protected UTF8Buffer tightUnmarshalString(DataByteArrayInputStream dataIn, BooleanStream bs) throws IOException {
    if (bs.readBoolean()) {
        boolean ascii = bs.readBoolean(); // ignored for now.
        int size = dataIn.readShort();
        if( size== 0 ) {
            return new UTF8Buffer("");
        } else {
            Buffer buffer = dataIn.readBuffer(size);
            return buffer.utf8();
        }
    } else {
        return null;
    }
}