Author: david.lloyd(a)jboss.com Date: 2008-12-16 18:06:36 -0500 (Tue, 16 Dec 2008) New Revision: 4755 Modified: remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java Log: Fix executor usage Modified: remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java =================================================================== --- remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java 2008-11-30 04:19:39 UTC (rev 4754) +++ remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java 2008-12-16 23:06:36 UTC (rev 4755) @@ -3,16 +3,17 @@ import java.io.IOException; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; -import java.util.Collection; -import java.util.Iterator; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.Executors; import org.jboss.remoting.core.EndpointImpl; import org.jboss.remoting.spi.RequestHandler; import org.jboss.remoting.spi.RequestHandlerSource; import org.jboss.remoting.spi.Handle; import org.jboss.xnio.IoUtils; +import org.jboss.xnio.CloseableExecutor; +import org.jboss.xnio.log.Logger; /** * The standalone interface into Remoting. This class contains static methods that are useful to standalone programs @@ -20,6 +21,8 @@ */ public final class Remoting { + private static final Logger log = Logger.getLogger("org.jboss.remoting"); + /** * Create an endpoint. The endpoint will create its own thread pool with a maximum of 10 threads. * @@ -38,22 +41,42 @@ * @return the endpoint */ public static Endpoint createEndpoint(final String name, final int maxThreads) { - final ThreadPoolExecutor executor = new ThreadPoolExecutor(0, maxThreads, Long.MAX_VALUE, TimeUnit.NANOSECONDS, new AlwaysBlockingQueue<Runnable>(new SynchronousQueue<Runnable>()), new ThreadPoolExecutor.AbortPolicy()); + final CloseableExecutor executor = createExecutor(maxThreads); final EndpointImpl endpoint = new EndpointImpl(executor, name); endpoint.addCloseHandler(new CloseHandler<Endpoint>() { public void handleClose(final Endpoint closed) { - executor.shutdown(); - try { - executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + IoUtils.safeClose(executor); } }); return endpoint; } + private static final ThreadFactory OUR_THREAD_FACTORY = new ThreadFactory() { + private final ThreadFactory defaultThreadFactory = Executors.defaultThreadFactory(); + + public Thread newThread(final Runnable r) { + final Thread thread = defaultThreadFactory.newThread(r); + thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + public void uncaughtException(final Thread t, final Throwable e) { + log.error(e, "Uncaught exception in thread %s", t); + } + }); + return thread; + } + }; + /** + * Create a simple thread pool that is compatible with Remoting. The thread pool will have a maximum of {@code maxThreads} + * threads. + * + * @param maxThreads the maximum thread count + * @return a closeable executor + */ + public static CloseableExecutor createExecutor(final int maxThreads) { + return IoUtils.closeableExecutor(new ThreadPoolExecutor(1, maxThreads, 30L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(50), OUR_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy()), 30L, TimeUnit.SECONDS); + } + + /** * Create an endpoint using the given {@code Executor} to execute tasks. * * @param executor the executor to use @@ -105,130 +128,5 @@ } } - private static class AlwaysBlockingQueue<E> implements BlockingQueue<E> { - private final BlockingQueue<E> delegate; - - public AlwaysBlockingQueue(final BlockingQueue<E> delegate) { - this.delegate = delegate; - } - - public boolean offer(final E o) { - try { - delegate.put(o); - return true; - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return false; - } - } - - public boolean offer(final E o, final long timeout, final TimeUnit unit) throws InterruptedException { - return delegate.offer(o, timeout, unit); - } - - public E poll(final long timeout, final TimeUnit unit) throws InterruptedException { - return delegate.poll(timeout, unit); - } - - public E take() throws InterruptedException { - return delegate.take(); - } - - public void put(final E o) throws InterruptedException { - delegate.put(o); - } - - public int remainingCapacity() { - return delegate.remainingCapacity(); - } - - public boolean add(final E o) { - return delegate.add(o); - } - - public int drainTo(final Collection<? super E> c) { - return delegate.drainTo(c); - } - - public int drainTo(final Collection<? super E> c, final int maxElements) { - return delegate.drainTo(c, maxElements); - } - - public E poll() { - return delegate.poll(); - } - - public E remove() { - return delegate.remove(); - } - - public E peek() { - return delegate.peek(); - } - - public E element() { - return delegate.element(); - } - - public int size() { - return delegate.size(); - } - - public boolean isEmpty() { - return delegate.isEmpty(); - } - - public boolean contains(final Object o) { - return delegate.contains(o); - } - - public Iterator<E> iterator() { - return delegate.iterator(); - } - - public Object[] toArray() { - return delegate.toArray(); - } - - public <T> T[] toArray(final T[] a) { - //noinspection SuspiciousToArrayCall - return delegate.toArray(a); - } - - public boolean remove(final Object o) { - return delegate.remove(o); - } - - public boolean containsAll(final Collection<?> c) { - return delegate.containsAll(c); - } - - public boolean addAll(final Collection<? extends E> c) { - return delegate.addAll(c); - } - - public boolean removeAll(final Collection<?> c) { - return delegate.removeAll(c); - } - - public boolean retainAll(final Collection<?> c) { - return delegate.retainAll(c); - } - - public void clear() { - delegate.clear(); - } - - public boolean equals(final Object o) { - return delegate.equals(o); - } - - public int hashCode() { - return delegate.hashCode(); - } - } - - // privates - private Remoting() { /* empty */ } }