import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
public void write(DataOutput out, StompFrame frame) throws IOException {
    write(out, new AsciiBuffer(frame.getCommand()));
    out.writeByte(NEWLINE_BYTE);

    for (Map.Entry<String, String> entry : frame.getProperties().entrySet()) {
        write(out, encodeHeader(entry.getKey()));
        out.writeByte(COLON_BYTE);
        write(out, encodeHeader(entry.getValue()));
        out.writeByte(NEWLINE_BYTE);
    }

    // denotes end of headers with a new line
    out.writeByte(NEWLINE_BYTE);
    write(out, frame.getContent());
    out.writeByte(NULL_BYTE);
    out.writeByte(NEWLINE_BYTE);
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的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.AsciiBuffer; //导入依赖的package包/类
public AsciiBuffer receiveAscii() {
    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
    try {

        int c = in.read();
        while (c >= 0) {
            if (c == 0) {
                return buffer.toBuffer().ascii();
            }
            buffer.write(c);
            c = in.read();
        }
    }catch (IOException e) {
        throw new IllegalStateException("Error receiving!", e);
    }

    throw new IllegalStateException("Socket closed, couldn't complete receiving", new EOFException());
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
@Override
public StompFrame parse(ByteBuffer data) throws IOException {

    while (data.hasRemaining()) {
        byte nextByte = data.get();

        // As of STOMP v1.2 lines can end with CRLF or just LF.
        // Any extra LF before start of the frame command are keep alive values.
        if (nextByte == '\r') {
            continue;
        }

        if (nextByte != '\n') {
            try {
                scratch.put(nextByte);
                continue;
            } catch (BufferOverflowException e) {
                throw new IOException("The maximum command length was exceeded");
            }
        }

        // Once we hit the true end of line and have read Command data we can
        // move onto the next stage, header parsing.
        if (scratch.position() != 0) {
            scratch.flip();
            AsciiBuffer command = new Buffer(scratch).trim().ascii();
            LOG.trace("New incoming STOMP frame, command := {}", command);
            StompFrame frame = new StompFrame(command.toString());
            currentParser = initiateHeaderRead(frame);
            return currentParser.parse(data);
        }
    }

    return null;
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
private void checkVirtualHostConfigChanges(LoggingTracker tracker) {
    // this is the current config (updated config in cases where config has changed)
    Map<AsciiBuffer,VirtualHostDTO> hostConfigById = resolveHostConfigById();

    CollectionSupport.DiffResult<AsciiBuffer> virtualHostConfigDiff = CollectionSupport.diff(virtualHosts.keySet(), hostConfigById.keySet());

    // apply updates to the virtual hosts for configs that have changed
    updateVirtualHosts(tracker, hostConfigById, virtualHostConfigDiff);
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
private Map<AsciiBuffer, VirtualHostDTO> resolveHostConfigById() {
    HashMap<AsciiBuffer, VirtualHostDTO> rc = new HashMap<AsciiBuffer, VirtualHostDTO>();
    for (VirtualHostDTO dto : config.virtual_hosts) {
        rc.put(AsciiBuffer.ascii(dto.id), dto);
    }
    return rc;
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
public static void main(String[] args) throws Exception {

		String user = env("ACTIVEMQ_USER", "admin");
		String password = env("ACTIVEMQ_PASSWORD", "password");
		String host = env("ACTIVEMQ_HOST", "localhost");
		int port = Integer.parseInt(env("ACTIVEMQ_PORT", "1883"));
		final String destination = arg(args, 0, "/topic/event");

		int messages = 10000;
		int size = 256;

		String DATA = "abcdefghijklmnopqrstuvwxyz";
		String body = "";
		for (int i = 0; i < size; i++) {
			body += DATA.charAt(i % DATA.length());
		}
		Buffer msg = new AsciiBuffer(body);

		MQTT mqtt = new MQTT();
		mqtt.setHost(host, port);
		mqtt.setUserName(user);
		mqtt.setPassword(password);

		FutureConnection connection = mqtt.futureConnection();
		connection.connect().await();

		final LinkedList<Future<Void>> queue = new LinkedList<Future<Void>>();
		UTF8Buffer topic = new UTF8Buffer(destination);
		for (int i = 1; i <= messages; i++) {

			// Send the publish without waiting for it to complete. This allows
			// us
			// to send multiple message without blocking..
			queue.add(connection.publish(topic, msg, QoS.AT_LEAST_ONCE, false));

			// Eventually we start waiting for old publish futures to complete
			// so that we don't create a large in memory buffer of outgoing
			// message.s
			if (queue.size() >= 1000) {
				queue.removeFirst().await();
			}

			if (i % 1000 == 0) {
				System.out.println(String.format("Sent %d messages.", i));
			}
		}

		queue.add(connection.publish(topic, new AsciiBuffer("SHUTDOWN"), QoS.AT_LEAST_ONCE, false));
	   
		while (!queue.isEmpty()) {
			queue.removeFirst().await();
		}

		connection.disconnect().await();

		System.exit(0);
	}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
AsciiBuffer getNextId() {
    return connection.nextId();
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
ExportStreamManager(OutputStream target, int version) throws IOException {
    this.target = target;
    this.version = version;
    stream = new TarOutputStream(new GZIPOutputStream(target));
    store("ver", new AsciiBuffer(""+version));
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
@Override
protected AsciiBuffer createBuffer(byte[] data) {
    return new AsciiBuffer(data);
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
public static void main(String []args) throws Exception {

        String user = env("ACTIVEMQ_USER", "admin");
        String password = env("ACTIVEMQ_PASSWORD", "password");
        String host = env("ACTIVEMQ_HOST", "localhost");
        int port = Integer.parseInt(env("ACTIVEMQ_PORT", "1883"));
        final String destination = arg(args, 0, "/topic/event");

        int messages = 10000;
        int size = 256;

        String DATA = "abcdefghijklmnopqrstuvwxyz";
        String body = "";
        for( int i=0; i < size; i ++) {
            body += DATA.charAt(i%DATA.length());
        }
       Buffer msg = new AsciiBuffer(body);

        MQTT mqtt = new MQTT();
        mqtt.setHost(host, port);
        mqtt.setUserName(user);
        mqtt.setPassword(password);

        FutureConnection connection = mqtt.futureConnection();
        connection.connect().await();

        final LinkedList<Future<Void>> queue = new LinkedList<Future<Void>>();
        UTF8Buffer topic = new UTF8Buffer(destination);
        for( int i=1; i <= messages; i ++) {

            // Send the publish without waiting for it to complete. This allows us
            // to send multiple message without blocking..
            queue.add(connection.publish(topic, msg, QoS.AT_LEAST_ONCE, false));

            // Eventually we start waiting for old publish futures to complete
            // so that we don't create a large in memory buffer of outgoing message.s
            if( queue.size() >= 1000 ) {
                queue.removeFirst().await();
            }

            if( i % 1000 == 0 ) {
                System.out.println(String.format("Sent %d messages.", i));
            }
        }

        queue.add(connection.publish(topic, new AsciiBuffer("SHUTDOWN"), QoS.AT_LEAST_ONCE, false));
        while( !queue.isEmpty() ) {
            queue.removeFirst().await();
        }

        connection.disconnect().await();

        System.exit(0);
    }
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
public static AsciiBuffer toAsciiBuffer(String value) {
    return new AsciiBuffer(value);
}
 

import org.fusesource.hawtbuf.AsciiBuffer; //导入依赖的package包/类
public static String fromAsciiBuffer(AsciiBuffer buffer) {
    return buffer.toString();
}