From 61523667e8589275138a66ad23fda254c66c7dfe Mon Sep 17 00:00:00 2001 From: Rajith Muditha Attapattu Date: Fri, 21 Nov 2008 17:57:16 +0000 Subject: Appologies for the sudden checkin without notice, close to the release cycle. Reverting the changes back. Will attach a patch and commit after the release. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@719657 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/client/AMQSession.java | 55 ++++++------------ .../apache/qpid/thread/DefaultThreadFactory.java | 18 ------ .../apache/qpid/thread/RealtimeThreadFactory.java | 47 ---------------- .../java/org/apache/qpid/thread/ThreadFactory.java | 7 --- .../java/org/apache/qpid/thread/Threading.java | 26 --------- .../qpid/transport/network/io/IoReceiver.java | 26 +++------ .../apache/qpid/transport/network/io/IoSender.java | 27 +++------ .../org/apache/qpid/testkit/perf/LatencyTest.java | 38 ++++--------- .../org/apache/qpid/testkit/perf/PerfConsumer.java | 23 +------- .../org/apache/qpid/testkit/perf/PerfProducer.java | 22 +------- .../qpid/testkit/soak/MultiThreadedConsumer.java | 17 +----- .../qpid/testkit/soak/MultiThreadedProducer.java | 14 +---- .../apache/qpid/testkit/soak/ResourceLeakTest.java | 20 +------ .../apache/qpid/testkit/soak/SimpleConsumer.java | 23 +------- .../apache/qpid/testkit/soak/SimpleProducer.java | 22 +------- .../main/java/org/apache/qpid/tools/QpidBench.java | 65 ++++------------------ 16 files changed, 71 insertions(+), 379 deletions(-) delete mode 100644 java/common/src/main/java/org/apache/qpid/thread/DefaultThreadFactory.java delete mode 100644 java/common/src/main/java/org/apache/qpid/thread/RealtimeThreadFactory.java delete mode 100644 java/common/src/main/java/org/apache/qpid/thread/ThreadFactory.java delete mode 100644 java/common/src/main/java/org/apache/qpid/thread/Threading.java (limited to 'java') diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index af0ed3faa3..b5d12d9520 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -67,26 +67,16 @@ import org.apache.qpid.client.failover.FailoverException; import org.apache.qpid.client.failover.FailoverNoopSupport; import org.apache.qpid.client.failover.FailoverProtectedOperation; import org.apache.qpid.client.failover.FailoverRetrySupport; -import org.apache.qpid.client.message.AMQMessageDelegateFactory; -import org.apache.qpid.client.message.AbstractJMSMessage; -import org.apache.qpid.client.message.CloseConsumerMessage; -import org.apache.qpid.client.message.JMSBytesMessage; -import org.apache.qpid.client.message.JMSMapMessage; -import org.apache.qpid.client.message.JMSObjectMessage; -import org.apache.qpid.client.message.JMSStreamMessage; -import org.apache.qpid.client.message.JMSTextMessage; -import org.apache.qpid.client.message.MessageFactoryRegistry; -import org.apache.qpid.client.message.UnprocessedMessage; +import org.apache.qpid.client.message.*; import org.apache.qpid.client.protocol.AMQProtocolHandler; -import org.apache.qpid.client.state.AMQState; -import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.util.FlowControllingBlockingQueue; +import org.apache.qpid.client.state.AMQStateManager; +import org.apache.qpid.client.state.AMQState; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.FieldTable; import org.apache.qpid.framing.FieldTableFactory; import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.jms.Session; -import org.apache.qpid.thread.Threading; import org.apache.qpid.url.AMQBindingURL; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -281,8 +271,6 @@ public abstract class AMQSession receiver, int bufferSize, long timeout) @@ -57,18 +55,10 @@ final class IoReceiver implements Runnable this.bufferSize = bufferSize; this.socket = transport.getSocket(); this.timeout = timeout; - - try - { - receiverThread = Threading.getThreadFactory().createThread(this); - } - catch(Exception e) - { - throw new Error("Error creating IOReceiver thread",e); - } - receiverThread.setDaemon(true); - receiverThread.setName(String.format("IoReceiver - %s", socket.getRemoteSocketAddress())); - receiverThread.start(); + + setDaemon(true); + setName(String.format("IoReceiver - %s", socket.getRemoteSocketAddress())); + start(); } void close(boolean block) @@ -85,10 +75,10 @@ final class IoReceiver implements Runnable { socket.shutdownInput(); } - if (block && Thread.currentThread() != receiverThread) + if (block && Thread.currentThread() != this) { - receiverThread.join(timeout); - if (receiverThread.isAlive()) + join(timeout); + if (isAlive()) { throw new TransportException("join timed out"); } diff --git a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java index 29f0c766fc..36ea14856a 100644 --- a/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java +++ b/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java @@ -24,7 +24,6 @@ import java.net.Socket; import java.nio.ByteBuffer; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.qpid.thread.Threading; import org.apache.qpid.transport.Sender; import org.apache.qpid.transport.SenderException; import org.apache.qpid.transport.TransportException; @@ -33,7 +32,7 @@ import org.apache.qpid.transport.util.Logger; import static org.apache.qpid.transport.util.Functions.*; -public final class IoSender implements Runnable, Sender +public final class IoSender extends Thread implements Sender { private static final Logger log = Logger.get(IoSender.class); @@ -55,8 +54,7 @@ public final class IoSender implements Runnable, Sender private final Object notFull = new Object(); private final Object notEmpty = new Object(); private final AtomicBoolean closed = new AtomicBoolean(false); - private final Thread senderThread; - + private volatile Throwable exception = null; @@ -76,18 +74,9 @@ public final class IoSender implements Runnable, Sender throw new TransportException("Error getting output stream for socket", e); } - try - { - senderThread = Threading.getThreadFactory().createThread(this); - } - catch(Exception e) - { - throw new Error("Error creating IOSender thread",e); - } - - senderThread.setDaemon(true); - senderThread.setName(String.format("IoSender - %s", socket.getRemoteSocketAddress())); - senderThread.start(); + setDaemon(true); + setName(String.format("IoSender - %s", socket.getRemoteSocketAddress())); + start(); } private static final int pof2(int n) @@ -199,10 +188,10 @@ public final class IoSender implements Runnable, Sender try { - if (Thread.currentThread() != senderThread) + if (Thread.currentThread() != this) { - senderThread.join(timeout); - if (senderThread.isAlive()) + join(timeout); + if (isAlive()) { throw new SenderException("join timed out"); } diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java b/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java index 4a4f3d124b..35a2374fbc 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java @@ -37,7 +37,6 @@ import javax.jms.MessageProducer; import javax.jms.TextMessage; import org.apache.qpid.testkit.MessageFactory; -import org.apache.qpid.thread.Threading; /** * Latency test sends an x number of messages in warmup mode and wait for a confirmation @@ -315,36 +314,19 @@ public class LatencyTest extends PerfBase implements MessageListener public static void main(String[] args) { - final LatencyTest latencyTest = new LatencyTest(); - Runnable r = new Runnable() + LatencyTest latencyTest = new LatencyTest(); + latencyTest.test(); + latencyTest.printToConsole(); + if (System.getProperty("file") != null) { - public void run() + try { - latencyTest.test(); - latencyTest.printToConsole(); - if (System.getProperty("file") != null) - { - try - { - latencyTest.writeToFile(); - } - catch(Exception e) - { - e.printStackTrace(); - } - } + latencyTest.writeToFile(); + } + catch(Exception e) + { + e.printStackTrace(); } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating latency test thread",e); } - t.start(); } } \ No newline at end of file diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java b/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java index 9781a7e839..cd12c7010d 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java @@ -27,8 +27,6 @@ import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.TextMessage; -import org.apache.qpid.thread.Threading; - /** * PerfConsumer will receive x no of messages in warmup mode. * Once it receives the Start message it will then signal the PerfProducer. @@ -244,24 +242,7 @@ public class PerfConsumer extends PerfBase implements MessageListener public static void main(String[] args) { - final PerfConsumer cons = new PerfConsumer(); - Runnable r = new Runnable() - { - public void run() - { - cons.test(); - } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating consumer thread",e); - } - t.start(); + PerfConsumer cons = new PerfConsumer(); + cons.test(); } } \ No newline at end of file diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java b/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java index e9421d7f22..757b1bfcda 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java @@ -27,7 +27,6 @@ import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import org.apache.qpid.testkit.MessageFactory; -import org.apache.qpid.thread.Threading; /** * PerfProducer sends an x no of messages in warmup mode and wait for a confirmation @@ -202,24 +201,7 @@ public class PerfProducer extends PerfBase public static void main(String[] args) { - final PerfProducer prod = new PerfProducer(); - Runnable r = new Runnable() - { - public void run() - { - prod.test(); - } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating producer thread",e); - } - t.start(); + PerfProducer prod = new PerfProducer(); + prod.test(); } } \ No newline at end of file diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java index d5514873e6..a91d9e7e85 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java @@ -29,8 +29,6 @@ import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; -import org.apache.qpid.thread.Threading; - /** * Test Description * ================ @@ -69,7 +67,7 @@ public class MultiThreadedConsumer extends BaseTest { final Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE); - Runnable r = new Runnable() + Thread t = new Thread(new Runnable() { public void run() { @@ -133,18 +131,7 @@ public class MultiThreadedConsumer extends BaseTest } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating consumer thread",e); - } - + }); t.setName("session-" + i); t.start(); } // for loop diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java index 1cf4ee28ca..279e5ea0bf 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java @@ -32,7 +32,6 @@ import javax.jms.TextMessage; import org.apache.qpid.client.AMQQueue; import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.thread.Threading; /** * Test Description @@ -80,7 +79,7 @@ public class MultiThreadedProducer extends SimpleProducer for (int i = 0; i < session_count; i++) { final Session session = con.createSession(transacted, Session.AUTO_ACKNOWLEDGE); - Runnable r = new Runnable() + Thread t = new Thread(new Runnable() { private Random gen = new Random(); @@ -143,16 +142,7 @@ public class MultiThreadedProducer extends SimpleProducer } - }; - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating producer thread",e); - } + }); t.setName("session-" + i); t.start(); diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java index 1ae2c35970..c33f9ffbf2 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java @@ -30,7 +30,6 @@ import javax.jms.Session; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.AMQQueue; import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.thread.Threading; /** * Test Description @@ -132,23 +131,8 @@ public class ResourceLeakTest extends BaseTest public static void main(String[] args) { - final ResourceLeakTest test = new ResourceLeakTest(); - Runnable r = new Runnable(){ - public void run() - { - test.test(); - } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating test thread",e); - } + ResourceLeakTest test = new ResourceLeakTest(); + test.test(); } } diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java index cd6d9013f8..b3eb97dafe 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java @@ -29,8 +29,6 @@ import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; -import org.apache.qpid.thread.Threading; - /** * Test Description * ================ @@ -128,24 +126,9 @@ public class SimpleConsumer extends BaseTest public static void main(String[] args) { - final SimpleConsumer test = new SimpleConsumer(); - Runnable r = new Runnable(){ - public void run() - { - test.setUp(); - test.test(); - } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating consumer thread",e); - } + SimpleConsumer test = new SimpleConsumer(); + test.setUp(); + test.test(); } } diff --git a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java index 805ce7ac29..1080092536 100644 --- a/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java +++ b/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java @@ -33,7 +33,6 @@ import javax.jms.TextMessage; import org.apache.qpid.client.AMQQueue; import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.thread.Threading; /** * Test Description @@ -139,24 +138,9 @@ public class SimpleProducer extends BaseTest public static void main(String[] args) { - final SimpleProducer test = new SimpleProducer(); - Runnable r = new Runnable(){ - public void run() - { - test.setUp(); - test.test(); - } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating producer thread",e); - } + SimpleProducer test = new SimpleProducer(); + test.setUp(); + test.test(); } } diff --git a/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java b/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java index 4bba7b113d..7411e81bd6 100644 --- a/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java +++ b/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java @@ -20,45 +20,23 @@ */ package org.apache.qpid.tools; -import static org.apache.qpid.tools.QpidBench.Mode.BOTH; -import static org.apache.qpid.tools.QpidBench.Mode.CONSUME; -import static org.apache.qpid.tools.QpidBench.Mode.PUBLISH; - -import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; - -import javax.jms.DeliveryMode; -import javax.jms.Destination; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.MessageListener; -import javax.jms.MessageProducer; -import javax.jms.TextMessage; +import java.util.UUID; +import javax.jms.*; import org.apache.qpid.client.AMQConnection; -import org.apache.qpid.thread.Threading; -import org.apache.qpid.transport.DeliveryProperties; -import org.apache.qpid.transport.ExchangeBind; -import org.apache.qpid.transport.Header; -import org.apache.qpid.transport.MessageAcceptMode; -import org.apache.qpid.transport.MessageAcquireMode; -import org.apache.qpid.transport.MessageCreditUnit; -import org.apache.qpid.transport.MessageDeliveryMode; -import org.apache.qpid.transport.MessageFlowMode; -import org.apache.qpid.transport.MessageProperties; -import org.apache.qpid.transport.MessageSubscribe; -import org.apache.qpid.transport.MessageTransfer; -import org.apache.qpid.transport.QueueDeclare; -import org.apache.qpid.transport.SessionException; -import org.apache.qpid.transport.SessionListener; +import org.apache.qpid.transport.*; +import org.apache.qpid.transport.network.io.IoTransport; import org.apache.qpid.util.UUIDGen; import org.apache.qpid.util.UUIDs; +import static org.apache.qpid.tools.QpidBench.Mode.*; + /** * QpidBench * @@ -434,7 +412,7 @@ public class QpidBench { case CONSUME: case BOTH: - Runnable r = new Runnable() + new Thread() { public void run() { @@ -454,18 +432,7 @@ public class QpidBench throw new RuntimeException(e); } } - }; - - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating consumer thread",e); - } - t.start(); + }.start(); break; } @@ -473,7 +440,7 @@ public class QpidBench { case PUBLISH: case BOTH: - Runnable r = new Runnable() + new Thread() { public void run() { @@ -493,17 +460,7 @@ public class QpidBench throw new RuntimeException(e); } } - }; - Thread t; - try - { - t = Threading.getThreadFactory().createThread(r); - } - catch(Exception e) - { - throw new Error("Error creating publisher thread",e); - } - t.start(); + }.start(); break; } } -- cgit v1.2.1