diff options
Diffstat (limited to 'qpid/java/client/src/old_test/java/org/apache/qpid/codec')
3 files changed, 513 insertions, 0 deletions
diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.java b/qpid/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.java new file mode 100644 index 0000000000..1db7e200bd --- /dev/null +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/codec/BasicDeliverTest.java @@ -0,0 +1,277 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.codec; + +import org.apache.qpid.framing.*; +import org.apache.mina.common.*; +import org.apache.mina.common.support.BaseIoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; + +import java.net.SocketAddress; + +/** + */ +public class BasicDeliverTest +{ + public static void main(String[] argv) throws Exception + { + BasicDeliverTest test = new BasicDeliverTest(); + + //warm up: + test.encode(512, 100000); + + //real tests: + test.encode(16, 10000, 15); + test.encode(32, 10000, 15); + test.encode(64, 10000, 15); + test.encode(128, 10000, 15); + test.encode(256, 10000, 15); + test.encode(512, 10000, 15); + test.encode(1024, 10000, 15); + test.encode(2048, 10000, 15); + + test.decode(16, 10000, 15); + test.decode(32, 10000, 15); + test.decode(64, 10000, 15); + test.decode(128, 10000, 15); + test.decode(256, 10000, 15); + test.decode(512, 10000, 15); + test.decode(1024, 10000, 15); + test.decode(2048, 10000, 15); + } + + void decode(int size, int count, int iterations) throws Exception + { + long min = Long.MAX_VALUE; + long max = 0; + long total = 0; + for (int i = 0; i < iterations; i++) + { + long time = decode(size, count); + total += time; + if (time < min) + { + min = time; + } + if (time > max) + { + max = time; + } + } + System.out.println("Decoded " + count + " messages of " + size + + " bytes: avg=" + (total / iterations) + ", min=" + min + ", max=" + max); + } + + + long decode(int size, int count) throws Exception + { + AMQDataBlock block = getDataBlock(size); + ByteBuffer data = ByteBuffer.allocate((int) block.getSize()); // XXX: Is cast a problem? + block.writePayload(data); + data.flip(); + AMQDecoder decoder = new AMQDecoder(false); + long start = System.currentTimeMillis(); + for (int i = 0; i < count; i++) + { + decoder.decode(session, data, decoderOutput); + data.rewind(); + } + return System.currentTimeMillis() - start; + } + + void encode(int size, int count, int iterations) throws Exception + { + long min = Long.MAX_VALUE; + long max = 0; + long total = 0; + for (int i = 0; i < iterations; i++) + { + long time = encode(size, count); + total += time; + if (time < min) + { + min = time; + } + if (time > max) + { + max = time; + } + } + System.out.println("Encoded " + count + " messages of " + size + + " bytes: avg=" + (total / iterations) + ", min=" + min + ", max=" + max); + } + + long encode(int size, int count) throws Exception + { + IoSession session = null; + AMQDataBlock block = getDataBlock(size); + AMQEncoder encoder = new AMQEncoder(); + long start = System.currentTimeMillis(); + for (int i = 0; i < count; i++) + { + encoder.encode(session, block, encoderOutput); + } + return System.currentTimeMillis() - start; + } + + private final ProtocolEncoderOutput encoderOutput = new ProtocolEncoderOutput() + { + + public void write(ByteBuffer byteBuffer) + { + } + + public void mergeAll() + { + } + + public WriteFuture flush() + { + return null; + } + }; + + private final ProtocolDecoderOutput decoderOutput = new ProtocolDecoderOutput() + { + public void write(Object object) + { + } + + public void flush() + { + } + }; + + private final IoSession session = new BaseIoSession() + { + + protected void updateTrafficMask() + { + //To change body of implemented methods use File | Settings | File Templates. + } + + public IoService getService() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public IoServiceConfig getServiceConfig() + { + return null; + } + + public IoHandler getHandler() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public IoSessionConfig getConfig() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public IoFilterChain getFilterChain() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public TransportType getTransportType() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public SocketAddress getRemoteAddress() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public SocketAddress getLocalAddress() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public SocketAddress getServiceAddress() + { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public int getScheduledWriteRequests() + { + return 0; //To change body of implemented methods use File | Settings | File Templates. + } + + public int getScheduledWriteBytes() + { + return 0; //To change body of implemented methods use File | Settings | File Templates. + } + }; + + private static final char[] DATA = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ".toCharArray(); + + static CompositeAMQDataBlock getDataBlock(int size) + { + //create a frame representing message delivery + AMQFrame[] frames = new AMQFrame[3]; + frames[0] = wrapBody(createBasicDeliverBody()); + frames[1] = wrapBody(createContentHeaderBody()); + frames[2] = wrapBody(createContentBody(size)); + + return new CompositeAMQDataBlock(frames); + } + + static AMQFrame wrapBody(AMQBody body) + { + AMQFrame frame = new AMQFrame(1, body); + return frame; + } + + static ContentBody createContentBody(int size) + { + ContentBody body = new ContentBody(); + body.payload = ByteBuffer.allocate(size); + for (int i = 0; i < size; i++) + { + body.payload.put((byte) DATA[i % DATA.length]); + } + return body; + } + + static ContentHeaderBody createContentHeaderBody() + { + ContentHeaderBody body = new ContentHeaderBody(); + body.properties = new BasicContentHeaderProperties(); + body.weight = 1; + body.classId = 6; + return body; + } + + static BasicDeliverBody createBasicDeliverBody() + { + BasicDeliverBody body = new BasicDeliverBody((byte) 8, (byte) 0, + BasicDeliverBody.getClazz((byte) 8, (byte) 0), + BasicDeliverBody.getMethod((byte) 8, (byte) 0), + new AMQShortString("myConsumerTag"), 1, + new AMQShortString("myExchange"), false, + new AMQShortString("myRoutingKey")); + return body; + } +} diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/codec/Client.java b/qpid/java/client/src/old_test/java/org/apache/qpid/codec/Client.java new file mode 100644 index 0000000000..3886021277 --- /dev/null +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/codec/Client.java @@ -0,0 +1,133 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.codec; + +import org.apache.mina.transport.socket.nio.SocketConnector; +import org.apache.mina.common.ConnectFuture; +import org.apache.mina.common.IoHandlerAdapter; +import org.apache.mina.common.IoSession; +import org.apache.mina.filter.codec.ProtocolCodecFilter; +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.BasicDeliverBody; +import org.apache.qpid.framing.ContentBody; + +import java.net.InetSocketAddress; + +public class Client extends IoHandlerAdapter +{ + //private static final int[] DEFAULT_SIZES = new int[]{1024, 512, 256, 128, 56}; + //private static final int[] DEFAULT_SIZES = new int[]{256, 256, 256, 256, 256, 512, 512, 512, 512, 512}; + private static final int[] DEFAULT_SIZES = new int[]{256, 512, 256, 512, 256, 512, 256, 512, 256, 512}; + //private static final int[] DEFAULT_SIZES = new int[]{1024, 1024, 1024, 1024, 1024}; + + private final IoSession _session; + private final long _start; + private final int _size; + private final int _count; + private int _received; + private boolean _closed; + + Client(String host, int port, int size, int count) throws Exception + { + _count = count; + _size = size; + AMQDataBlock block = BasicDeliverTest.getDataBlock(size); + + InetSocketAddress address = new InetSocketAddress(host, port); + ConnectFuture future = new SocketConnector().connect(address, this); + future.join(); + _session = future.getSession(); + + _start = System.currentTimeMillis(); + for(int i = 0; i < count; i++) + { + _session.write(block); + } + } + + void close() + { + long time = System.currentTimeMillis() - _start; + System.out.println("Received " + _received + " messages of " + _size + + " bytes in " + time + "ms."); + _session.close(); + synchronized(this) + { + _closed = true; + notify(); + } + } + + void waitForClose() throws InterruptedException + { + synchronized(this) + { + while(!_closed) + { + wait(); + } + } + } + + public void sessionCreated(IoSession session) throws Exception + { + session.getFilterChain().addLast("protocolFilter", new ProtocolCodecFilter(new AMQCodecFactory(false))); + } + + public void messageReceived(IoSession session, Object object) throws Exception + { + if(isContent(object) && ++_received == _count) close(); + } + + public void exceptionCaught(IoSession session, Throwable throwable) throws Exception + { + throwable.printStackTrace(); + close(); + } + + private static boolean isDeliver(Object o) + { + return o instanceof AMQFrame && ((AMQFrame) o).getBodyFrame() instanceof BasicDeliverBody; + } + + private static boolean isContent(Object o) + { + return o instanceof AMQFrame && ((AMQFrame) o).getBodyFrame() instanceof ContentBody; + } + + public static void main(String[] argv) throws Exception + { + String host = argv.length > 0 ? argv[0] : "localhost"; + int port = argv.length > 1 ? Integer.parseInt(argv[1]) : 8888; + int count = argv.length > 2 ? Integer.parseInt(argv[2]) : 10000; + int[] sizes = argv.length > 3 ? new int[]{Integer.parseInt(argv[3])} : DEFAULT_SIZES; + + System.out.println("Connecting to " + host + ":" + port); + + for(int i = 0; i < sizes.length; i++) + { + new Client(host, port, sizes[i], count).waitForClose(); + Thread.sleep(1000); + } + } + +} diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/codec/Server.java b/qpid/java/client/src/old_test/java/org/apache/qpid/codec/Server.java new file mode 100644 index 0000000000..fa4295e0b2 --- /dev/null +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/codec/Server.java @@ -0,0 +1,103 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.codec; + +import org.apache.mina.common.IoHandlerAdapter; +import org.apache.mina.common.IoSession; +import org.apache.mina.transport.socket.nio.SocketAcceptor; +import org.apache.mina.util.SessionUtil; +import org.apache.mina.filter.codec.ProtocolCodecFilter; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.CompositeAMQDataBlock; + +import java.net.InetSocketAddress; + +public class Server extends IoHandlerAdapter +{ + Server(int port) throws Exception + { + new SocketAcceptor().bind(new InetSocketAddress(port), this); + System.out.println("Listening on " + port); + } + + public void sessionCreated(IoSession session) throws Exception + { + SessionUtil.initialize(session); + session.getFilterChain().addLast("protocolFilter", new ProtocolCodecFilter(new AMQCodecFactory(false))); + } + + public void messageReceived(IoSession session, Object object) throws Exception + { + getAccumulator(session).received(session, (AMQFrame) object); + } + + public void sessionOpened(IoSession session) throws Exception + { + System.out.println("sessionOpened()"); + } + + public void sessionClosed(IoSession session) throws Exception + { + System.out.println("sessionClosed()"); + } + + public void exceptionCaught(IoSession session, Throwable t) throws Exception + { + System.out.println("exceptionCaught()"); + t.printStackTrace(); + session.close(); + } + + private Accumulator getAccumulator(IoSession session) + { + Accumulator a = (Accumulator) session.getAttribute(ACCUMULATOR); + if(a == null) + { + a = new Accumulator(); + session.setAttribute(ACCUMULATOR, a); + } + return a; + } + + private static final String ACCUMULATOR = Accumulator.class.getName(); + + private static class Accumulator + { + private final AMQFrame[] frames = new AMQFrame[3]; + private int i; + + void received(IoSession session, AMQFrame frame) + { + frames[i++] = frame; + if(i >= frames.length) + { + i = 0; + session.write(new CompositeAMQDataBlock(frames)); + } + } + } + + public static void main(String[] argv) throws Exception + { + int port = argv.length > 0 ? Integer.parseInt(argv[0]) : 8888; + new Server(port); + } +} |