[jboss-remoting-commits] JBoss Remoting SVN: r4855 - in remoting3/trunk: api/src/main/java/org/jboss/remoting and 2 other directories.

Author: david.lloyd(a)jboss.com Date: 2009-02-26 13:30:20 -0500 (Thu, 26 Feb 2009) New Revision: 4855 Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/AbstractContextImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientContextImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientExternalizer.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceExternalizer.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ConcurrentReferenceHashMap.java remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureClientSource.java remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureReplyImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandler.java remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandlerSource.java remoting3/trunk/api/src/main/java/org/jboss/remoting/OrderedExecutor.java remoting3/trunk/api/src/main/java/org/jboss/remoting/QueueExecutor.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestListenerExecutor.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceContextImpl.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceRegistration.java remoting3/trunk/api/src/test/java/org/jboss/remoting/EndpointTestCase.java Removed: remoting3/trunk/core/ Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/Remoting.java remoting3/trunk/build.xml remoting3/trunk/testing-support/src/main/resources/testing.policy Log: Merge core and api into a single module (since they cannot be separated anyway) Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/AbstractContextImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/AbstractContextImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/AbstractContextImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/AbstractContextImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,56 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +import org.jboss.remoting.spi.AbstractHandleableCloseable; + +/** + * + */ +abstract class AbstractContextImpl<T> extends AbstractHandleableCloseable<T> { + + private final ConcurrentMap<Object, Object> attributes = new ConcurrentHashMap<Object, Object>(); + + AbstractContextImpl(final Executor executor) { + super(executor); + } + + public ConcurrentMap<Object, Object> getAttributes() { + return attributes; + } + + protected Executor getExecutor() { + return super.getExecutor(); + } + + protected boolean isOpen() { + return super.isOpen(); + } + + public String toString() { + return "generic context instance <" + Integer.toHexString(hashCode()) + ">"; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientContextImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientContextImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientContextImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientContextImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,56 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.util.concurrent.Executor; +import org.jboss.remoting.ClientContext; +import org.jboss.remoting.ServiceContext; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class ClientContextImpl extends AbstractContextImpl<ClientContext> implements ClientContext { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.client-context"); + + private final ServiceContextImpl serviceContext; + + ClientContextImpl(final Executor executor) { + super(executor); + serviceContext = null; + } + + ClientContextImpl(final ServiceContextImpl serviceContext) { + super(serviceContext.getExecutor()); + this.serviceContext = serviceContext; + } + + public ServiceContext getServiceContext() { + return serviceContext; + } + + public String toString() { + return "client context instance <" + Integer.toHexString(hashCode()) + ">"; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientExternalizer.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientExternalizer.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientExternalizer.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientExternalizer.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,69 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import org.jboss.marshalling.Creator; +import org.jboss.marshalling.Externalizer; +import org.jboss.remoting.spi.RequestHandler; + +/** + * + */ +final class ClientExternalizer implements Externalizer { + + private static final long serialVersionUID = 814228455390899997L; + + private final EndpointImpl endpoint; + + ClientExternalizer(final EndpointImpl endpoint) { + this.endpoint = endpoint; + } + + private static <I, O> void doWriteExternal(final ClientImpl<I, O> client, final ObjectOutput output) throws IOException { + output.writeObject(client.getRequestClass()); + output.writeObject(client.getReplyClass()); + output.writeObject(client.getRequestHandlerHandle().getResource()); + } + + public void writeExternal(final Object o, final ObjectOutput output) throws IOException { + doWriteExternal((ClientImpl<?, ?>) o, output); + } + + private <I, O> ClientImpl<I, O> doCreateExternal(Class<I> requestClass, Class<O> replyClass, RequestHandler handler) throws IOException { + return ClientImpl.create(handler.getHandle(), endpoint.getExecutor(), requestClass, replyClass); + } + + public Object createExternal(final Class<?> aClass, final ObjectInput input, final Creator creator) throws IOException, ClassNotFoundException { + final Class<?> requestClass = (Class<?>) input.readObject(); + final Class<?> replyClass = (Class<?>) input.readObject(); + final RequestHandler handler = (RequestHandler) input.readObject(); + return doCreateExternal(requestClass, replyClass, handler); + } + + public void readExternal(final Object o, final ObjectInput input) throws IOException, ClassNotFoundException { + // no op + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,140 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.util.concurrent.Executor; +import org.jboss.remoting.Client; +import org.jboss.remoting.CloseHandler; +import org.jboss.remoting.IndeterminateOutcomeException; +import org.jboss.remoting.RemoteRequestException; +import org.jboss.remoting.spi.Handle; +import org.jboss.remoting.spi.RemoteRequestContext; +import org.jboss.remoting.spi.ReplyHandler; +import org.jboss.remoting.spi.RequestHandler; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class ClientImpl<I, O> extends AbstractContextImpl<Client<I, O>> implements Client<I, O> { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.client"); + + private final Handle<RequestHandler> handle; + private final Class<I> requestClass; + private final Class<O> replyClass; + + private ClientImpl(final Handle<RequestHandler> handle, final Executor executor, final Class<I> requestClass, final Class<O> replyClass) { + super(executor); + this.handle = handle; + this.requestClass = requestClass; + this.replyClass = replyClass; + } + + static <I, O> ClientImpl<I, O> create(final Handle<RequestHandler> handle, final Executor executor, final Class<I> requestClass, final Class<O> replyClass) { + final ClientImpl<I, O> ci = new ClientImpl<I, O>(handle, executor, requestClass, replyClass); + handle.addCloseHandler(new CloseHandler<Handle<RequestHandler>>() { + public void handleClose(final Handle<RequestHandler> closed) { + IoUtils.safeClose(ci); + } + }); + return ci; + } + + protected void closeAction() throws IOException { + handle.close(); + } + + public O invoke(final I request) throws IOException { + if (! isOpen()) { + throw new IOException("Client is not open"); + } + log.trace("Client.invoke() sending request \"%s\"", request); + final I actualRequest = castRequest(request); + final QueueExecutor executor = new QueueExecutor(); + final FutureReplyImpl<O> futureReply = new FutureReplyImpl<O>(executor, replyClass); + final ReplyHandler replyHandler = futureReply.getReplyHandler(); + final RemoteRequestContext requestContext = handle.getResource().receiveRequest(actualRequest, replyHandler); + futureReply.setRemoteRequestContext(requestContext); + futureReply.addNotifier(IoUtils.<O>attachmentClosingNotifier(), executor); + executor.runQueue(); + try { + final O reply = futureReply.getInterruptibly(); + log.trace("Client.invoke() received reply \"%s\"", reply); + return reply; + } catch (InterruptedException e) { + try { + futureReply.cancel(); + throw new IndeterminateOutcomeException("The current thread was interrupted before the result could be read"); + } finally { + Thread.currentThread().interrupt(); + } + } + } + + public IoFuture<? extends O> send(final I request) throws IOException { + if (! isOpen()) { + throw new IOException("Client is not open"); + } + log.trace("Client.send() sending request \"%s\"", request); + final I actualRequest = castRequest(request); + final FutureReplyImpl<O> futureReply = new FutureReplyImpl<O>(executor, replyClass); + final ReplyHandler replyHandler = futureReply.getReplyHandler(); + final RemoteRequestContext requestContext = handle.getResource().receiveRequest(actualRequest, replyHandler); + futureReply.setRemoteRequestContext(requestContext); + return futureReply; + } + + /** + * Since type is erased, it's possible that the wrong type was passed. + * @param request + * @return + * @throws RemoteRequestException + */ + private I castRequest(final Object request) throws RemoteRequestException { + try { + return requestClass.cast(request); + } catch (ClassCastException e) { + throw new RemoteRequestException("Invalid request type sent (got <" + request.getClass().getName() + ">, expected <? extends " + requestClass.getName() + ">"); + } + } + + public String toString() { + return "client instance <" + Integer.toHexString(hashCode()) + ">"; + } + + Handle<RequestHandler> getRequestHandlerHandle() { + return handle; + } + + Class<I> getRequestClass() { + return requestClass; + } + + Class<O> getReplyClass() { + return replyClass; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceExternalizer.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceExternalizer.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceExternalizer.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceExternalizer.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,69 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import org.jboss.marshalling.Creator; +import org.jboss.marshalling.Externalizer; +import org.jboss.remoting.spi.RequestHandlerSource; + +/** + * + */ +final class ClientSourceExternalizer implements Externalizer { + + private static final long serialVersionUID = 814228455390899997L; + + private final EndpointImpl endpoint; + + ClientSourceExternalizer(final EndpointImpl endpoint) { + this.endpoint = endpoint; + } + + private static <I, O> void doWriteExternal(final ClientSourceImpl<I, O> clientSource, final ObjectOutput output) throws IOException { + output.writeObject(clientSource.getRequestClass()); + output.writeObject(clientSource.getReplyClass()); + output.writeObject(clientSource.getRequestHandlerSourceHandle().getResource()); + } + + public void writeExternal(final Object o, final ObjectOutput output) throws IOException { + doWriteExternal((ClientSourceImpl<?, ?>) o, output); + } + + private <I, O> ClientSourceImpl<I, O> doCreateExternal(Class<I> requestClass, Class<O> replyClass, RequestHandlerSource handlerSource) throws IOException { + return ClientSourceImpl.create(handlerSource.getHandle(), endpoint, requestClass, replyClass); + } + + public Object createExternal(final Class<?> aClass, final ObjectInput input, final Creator creator) throws IOException, ClassNotFoundException { + final Class<?> requestClass = (Class<?>) input.readObject(); + final Class<?> replyClass = (Class<?>) input.readObject(); + final RequestHandlerSource handlerSource = (RequestHandlerSource) input.readObject(); + return doCreateExternal(requestClass, replyClass, handlerSource); + } + + public void readExternal(final Object o, final ObjectInput input) throws IOException, ClassNotFoundException { + // no op + } +} \ No newline at end of file Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,98 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import org.jboss.remoting.Client; +import org.jboss.remoting.ClientSource; +import org.jboss.remoting.CloseHandler; +import org.jboss.remoting.Endpoint; +import org.jboss.remoting.spi.AbstractHandleableCloseable; +import org.jboss.remoting.spi.Handle; +import org.jboss.remoting.spi.RequestHandler; +import org.jboss.remoting.spi.RequestHandlerSource; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class ClientSourceImpl<I, O> extends AbstractHandleableCloseable<ClientSource<I, O>> implements ClientSource<I, O> { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.client-source"); + + private final Handle<RequestHandlerSource> handle; + private final Endpoint endpoint; + private final Class<I> requestClass; + private final Class<O> replyClass; + + private ClientSourceImpl(final Handle<RequestHandlerSource> handle, final EndpointImpl endpoint, final Class<I> requestClass, final Class<O> replyClass) { + super(endpoint.getExecutor()); + this.handle = handle; + this.endpoint = endpoint; + this.requestClass = requestClass; + this.replyClass = replyClass; + } + + static <I, O> ClientSourceImpl<I, O> create(final Handle<RequestHandlerSource> handle, final EndpointImpl endpoint, final Class<I> requestClass, final Class<O> replyClass) { + final ClientSourceImpl<I, O> csi = new ClientSourceImpl<I, O>(handle, endpoint, requestClass, replyClass); + handle.addCloseHandler(new CloseHandler<Handle<RequestHandlerSource>>() { + public void handleClose(final Handle<RequestHandlerSource> closed) { + IoUtils.safeClose(csi); + } + }); + return csi; + } + + protected void closeAction() throws IOException { + handle.close(); + } + + public Client<I, O> createClient() throws IOException { + if (! isOpen()) { + throw new IOException("Client source is not open"); + } + final Handle<RequestHandler> clientHandle = handle.getResource().createRequestHandler(); + try { + return endpoint.createClient(clientHandle.getResource(), requestClass, replyClass); + } finally { + IoUtils.safeClose(clientHandle); + } + } + + public String toString() { + return "client source instance <" + Integer.toString(hashCode()) + ">"; + } + + Handle<RequestHandlerSource> getRequestHandlerSourceHandle() { + return handle; + } + + Class<I> getRequestClass() { + return requestClass; + } + + Class<O> getReplyClass() { + return replyClass; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ConcurrentReferenceHashMap.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ConcurrentReferenceHashMap.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ConcurrentReferenceHashMap.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ConcurrentReferenceHashMap.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,1709 @@ +/* + * Written by Doug Lea with assistance from members of JCP JSR-166 + * Expert Group and released to the public domain, as explained at + * http://creativecommons.org/licenses/publicdomain + */ + +package org.jboss.remoting; +import java.io.IOException; +import java.io.Serializable; +import java.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.SoftReference; +import java.lang.ref.WeakReference; +import java.util.AbstractCollection; +import java.util.AbstractMap; +import java.util.AbstractSet; +import java.util.Collection; +import java.util.ConcurrentModificationException; +import java.util.EnumSet; +import java.util.Enumeration; +import java.util.HashMap; +import java.util.Hashtable; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.concurrent.locks.ReentrantLock; + +/** + * An advanced hash table supporting configurable garbage collection semantics + * of keys and values, optional referential-equality, full concurrency of + * retrievals, and adjustable expected concurrency for updates. + * + * This table is designed around specific advanced use-cases. If there is any + * doubt whether this table is for you, you most likely should be using + * {@link java.util.concurrent.ConcurrentHashMap} instead. + * + * This table supports strong, weak, and soft keys and values. By default keys + * are weak, and values are strong. Such a configuration offers similar behavior + * to {@link java.util.WeakHashMap}, entries of this table are periodically + * removed once their corresponding keys are no longer referenced outside of + * this table. In other words, this table will not prevent a key from being + * discarded by the garbage collector. Once a key has been discarded by the + * collector, the corresponding entry is no longer visible to this table; + * however, the entry may occupy space until a future table operation decides to + * reclaim it. For this reason, summary functions such as <tt>size</tt> and + * <tt>isEmpty</tt> might return a value greater than the observed number of + * entries. In order to support a high level of concurrency, stale entries are + * only reclaimed during blocking (usually mutating) operations. + * + * Enabling soft keys allows entries in this table to remain until their space + * is absolutely needed by the garbage collector. This is unlike weak keys which + * can be reclaimed as soon as they are no longer referenced by a normal strong + * reference. The primary use case for soft keys is a cache, which ideally + * occupies memory that is not in use for as long as possible. + * + * By default, values are held using a normal strong reference. This provides + * the commonly desired guarantee that a value will always have at least the + * same life-span as it's key. For this reason, care should be taken to ensure + * that a value never refers, either directly or indirectly, to its key, thereby + * preventing reclamation. If this is unavoidable, then it is recommended to use + * the same reference type in use for the key. However, it should be noted that + * non-strong values may disappear before their corresponding key. + * + * While this table does allow the use of both strong keys and values, it is + * recommended to use {@link java.util.concurrent.ConcurrentHashMap} for such a + * configuration, since it is optimized for that case. + * + * Just like {@link java.util.concurrent.ConcurrentHashMap}, this class obeys + * the same functional specification as {@link java.util.Hashtable}, and + * includes versions of methods corresponding to each method of + * <tt>Hashtable</tt>. However, even though all operations are thread-safe, + * retrieval operations do <em>not</em> entail locking, and there is + * <em>not</em> any support for locking the entire table in a way that + * prevents all access. This class is fully interoperable with + * <tt>Hashtable</tt> in programs that rely on its thread safety but not on + * its synchronization details. + * + * <p> + * Retrieval operations (including <tt>get</tt>) generally do not block, so + * may overlap with update operations (including <tt>put</tt> and + * <tt>remove</tt>). Retrievals reflect the results of the most recently + * <em>completed</em> update operations holding upon their onset. For + * aggregate operations such as <tt>putAll</tt> and <tt>clear</tt>, + * concurrent retrievals may reflect insertion or removal of only some entries. + * Similarly, Iterators and Enumerations return elements reflecting the state of + * the hash table at some point at or since the creation of the + * iterator/enumeration. They do <em>not</em> throw + * {@link ConcurrentModificationException}. However, iterators are designed to + * be used by only one thread at a time. + * + * <p> + * The allowed concurrency among update operations is guided by the optional + * <tt>concurrencyLevel</tt> constructor argument (default <tt>16</tt>), + * which is used as a hint for internal sizing. The table is internally + * partitioned to try to permit the indicated number of concurrent updates + * without contention. Because placement in hash tables is essentially random, + * the actual concurrency will vary. Ideally, you should choose a value to + * accommodate as many threads as will ever concurrently modify the table. Using + * a significantly higher value than you need can waste space and time, and a + * significantly lower value can lead to thread contention. But overestimates + * and underestimates within an order of magnitude do not usually have much + * noticeable impact. A value of one is appropriate when it is known that only + * one thread will modify and all others will only read. Also, resizing this or + * any other kind of hash table is a relatively slow operation, so, when + * possible, it is a good idea to provide estimates of expected table sizes in + * constructors. + * + * <p> + * This class and its views and iterators implement all of the <em>optional</em> + * methods of the {@link Map} and {@link Iterator} interfaces. + * + * <p> + * Like {@link Hashtable} but unlike {@link HashMap}, this class does + * <em>not</em> allow <tt>null</tt> to be used as a key or value. + * + * <p> + * This class is a member of the <a href="{@docRoot}/../technotes/guides/collections/index.html"> + * Java Collections Framework</a>. + * + * @author Doug Lea + * @author Jason T. Greene + * @param <K> the type of keys maintained by this map + * @param <V> the type of mapped values + */ +class ConcurrentReferenceHashMap<K, V> extends AbstractMap<K, V> + implements java.util.concurrent.ConcurrentMap<K, V>, Serializable { + private static final long serialVersionUID = 7249069246763182397L; + + /* + * The basic strategy is to subdivide the table among Segments, + * each of which itself is a concurrently readable hash table. + */ + + /** + * An option specifying which Java reference type should be used to refer + * to a key and/or value. + */ + public static enum ReferenceType { + /** Indicates a normal Java strong reference should be used */ + STRONG, + /** Indicates a {@link WeakReference} should be used */ + WEAK, + /** Indicates a {@link SoftReference} should be used */ + SOFT + }; + + + public static enum Option { + /** Indicates that referential-equality (== instead of .equals()) should + * be used when locating keys. This offers similar behavior to {@link IdentityHashMap} */ + IDENTITY_COMPARISONS + }; + + /* ---------------- Constants -------------- */ + + static final ReferenceType DEFAULT_KEY_TYPE = ReferenceType.WEAK; + + static final ReferenceType DEFAULT_VALUE_TYPE = ReferenceType.STRONG; + + + /** + * The default initial capacity for this table, + * used when not otherwise specified in a constructor. + */ + static final int DEFAULT_INITIAL_CAPACITY = 16; + + /** + * The default load factor for this table, used when not + * otherwise specified in a constructor. + */ + static final float DEFAULT_LOAD_FACTOR = 0.75f; + + /** + * The default concurrency level for this table, used when not + * otherwise specified in a constructor. + */ + static final int DEFAULT_CONCURRENCY_LEVEL = 16; + + /** + * The maximum capacity, used if a higher value is implicitly + * specified by either of the constructors with arguments. MUST + * be a power of two <= 1<<30 to ensure that entries are indexable + * using ints. + */ + static final int MAXIMUM_CAPACITY = 1 << 30; + + /** + * The maximum number of segments to allow; used to bound + * constructor arguments. + */ + static final int MAX_SEGMENTS = 1 << 16; // slightly conservative + + /** + * Number of unsynchronized retries in size and containsValue + * methods before resorting to locking. This is used to avoid + * unbounded retries if tables undergo continuous modification + * which would make it impossible to obtain an accurate result. + */ + static final int RETRIES_BEFORE_LOCK = 2; + + /* ---------------- Fields -------------- */ + + /** + * Mask value for indexing into segments. The upper bits of a + * key's hash code are used to choose the segment. + */ + final int segmentMask; + + /** + * Shift value for indexing within segments. + */ + final int segmentShift; + + /** + * The segments, each of which is a specialized hash table + */ + final Segment<K,V>[] segments; + + boolean identityComparisons; + + transient Set<K> keySet; + transient Set<Map.Entry<K,V>> entrySet; + transient Collection<V> values; + + /* ---------------- Small Utilities -------------- */ + + /** + * Applies a supplemental hash function to a given hashCode, which + * defends against poor quality hash functions. This is critical + * because ConcurrentReferenceHashMap uses power-of-two length hash tables, + * that otherwise encounter collisions for hashCodes that do not + * differ in lower or upper bits. + */ + private static int hash(int h) { + // Spread bits to regularize both segment and index locations, + // using variant of single-word Wang/Jenkins hash. + h += (h << 15) ^ 0xffffcd7d; + h ^= (h >>> 10); + h += (h << 3); + h ^= (h >>> 6); + h += (h << 2) + (h << 14); + return h ^ (h >>> 16); + } + + /** + * Returns the segment that should be used for key with given hash + * @param hash the hash code for the key + * @return the segment + */ + final Segment<K,V> segmentFor(int hash) { + return segments[(hash >>> segmentShift) & segmentMask]; + } + + private int hashOf(Object key) { + return hash(identityComparisons ? + System.identityHashCode(key) : key.hashCode()); + } + + /* ---------------- Inner Classes -------------- */ + + static interface KeyReference { + int keyHash(); + Object keyRef(); + } + + /** + * A weak-key reference which stores the key hash needed for reclamation. + */ + static final class WeakKeyReference<K> extends WeakReference<K> implements KeyReference { + final int hash; + WeakKeyReference(K key, int hash, ReferenceQueue<Object> refQueue) { + super(key, refQueue); + this.hash = hash; + } + public final int keyHash() { + return hash; + } + + public final Object keyRef() { + return this; + } + } + + /** + * A soft-key reference which stores the key hash needed for reclamation. + */ + static final class SoftKeyReference<K> extends SoftReference<K> implements KeyReference { + final int hash; + SoftKeyReference(K key, int hash, ReferenceQueue<Object> refQueue) { + super(key, refQueue); + this.hash = hash; + } + public final int keyHash() { + return hash; + } + + public final Object keyRef() { + return this; + } + } + + static final class WeakValueReference<V> extends WeakReference<V> implements KeyReference { + final Object keyRef; + final int hash; + WeakValueReference(V value, Object keyRef, int hash, ReferenceQueue<Object> refQueue) { + super(value, refQueue); + this.keyRef = keyRef; + this.hash = hash; + } + + public final int keyHash() { + return hash; + } + + public final Object keyRef() { + return keyRef; + } + } + + static final class SoftValueReference<V> extends SoftReference<V> implements KeyReference { + final Object keyRef; + final int hash; + SoftValueReference(V value, Object keyRef, int hash, ReferenceQueue<Object> refQueue) { + super(value, refQueue); + this.keyRef = keyRef; + this.hash = hash; + } + public final int keyHash() { + return hash; + } + + public final Object keyRef() { + return keyRef; + } + } + + /** + * ConcurrentReferenceHashMap list entry. Note that this is never exported + * out as a user-visible Map.Entry. + * + * Because the value field is volatile, not final, it is legal wrt + * the Java Memory Model for an unsynchronized reader to see null + * instead of initial value when read via a data race. Although a + * reordering leading to this is not likely to ever actually + * occur, the Segment.readValueUnderLock method is used as a + * backup in case a null (pre-initialized) value is ever seen in + * an unsynchronized access method. + */ + static final class HashEntry<K,V> { + final Object keyRef; + final int hash; + volatile Object valueRef; + final HashEntry<K,V> next; + + HashEntry(K key, int hash, HashEntry<K,V> next, V value, + ReferenceType keyType, ReferenceType valueType, + ReferenceQueue<Object> refQueue) { + this.hash = hash; + this.next = next; + this.keyRef = newKeyReference(key, keyType, refQueue); + this.valueRef = newValueReference(value, valueType, refQueue); + } + + final Object newKeyReference(K key, ReferenceType keyType, + ReferenceQueue<Object> refQueue) { + if (keyType == ReferenceType.WEAK) + return new WeakKeyReference<K>(key, hash, refQueue); + if (keyType == ReferenceType.SOFT) + return new SoftKeyReference<K>(key, hash, refQueue); + + return key; + } + + final Object newValueReference(V value, ReferenceType valueType, + ReferenceQueue<Object> refQueue) { + if (valueType == ReferenceType.WEAK) + return new WeakValueReference<V>(value, keyRef, hash, refQueue); + if (valueType == ReferenceType.SOFT) + return new SoftValueReference<V>(value, keyRef, hash, refQueue); + + return value; + } + + @SuppressWarnings("unchecked") + final K key() { + if (keyRef instanceof Reference) + return ((Reference<K>)keyRef).get(); + + return (K) keyRef; + } + + final V value() { + return dereferenceValue(valueRef); + } + + @SuppressWarnings("unchecked") + final V dereferenceValue(Object value) { + if (value instanceof Reference) + return ((Reference<V>)value).get(); + + return (V) value; + } + + final void setValue(V value, ReferenceType valueType, ReferenceQueue<Object> refQueue) { + this.valueRef = newValueReference(value, valueType, refQueue); + } + + @SuppressWarnings("unchecked") + static final <K,V> HashEntry<K,V>[] newArray(int i) { + return new HashEntry[i]; + } + } + + /** + * Segments are specialized versions of hash tables. This + * subclasses from ReentrantLock opportunistically, just to + * simplify some locking and avoid separate construction. + */ + static final class Segment<K,V> extends ReentrantLock implements Serializable { + /* + * Segments maintain a table of entry lists that are ALWAYS + * kept in a consistent state, so can be read without locking. + * Next fields of nodes are immutable (final). All list + * additions are performed at the front of each bin. This + * makes it easy to check changes, and also fast to traverse. + * When nodes would otherwise be changed, new nodes are + * created to replace them. This works well for hash tables + * since the bin lists tend to be short. (The average length + * is less than two for the default load factor threshold.) + * + * Read operations can thus proceed without locking, but rely + * on selected uses of volatiles to ensure that completed + * write operations performed by other threads are + * noticed. For most purposes, the "count" field, tracking the + * number of elements, serves as that volatile variable + * ensuring visibility. This is convenient because this field + * needs to be read in many read operations anyway: + * + * - All (unsynchronized) read operations must first read the + * "count" field, and should not look at table entries if + * it is 0. + * + * - All (synchronized) write operations should write to + * the "count" field after structurally changing any bin. + * The operations must not take any action that could even + * momentarily cause a concurrent read operation to see + * inconsistent data. This is made easier by the nature of + * the read operations in Map. For example, no operation + * can reveal that the table has grown but the threshold + * has not yet been updated, so there are no atomicity + * requirements for this with respect to reads. + * + * As a guide, all critical volatile reads and writes to the + * count field are marked in code comments. + */ + + private static final long serialVersionUID = 2249069246763182397L; + + /** + * The number of elements in this segment's region. + */ + transient volatile int count; + + /** + * Number of updates that alter the size of the table. This is + * used during bulk-read methods to make sure they see a + * consistent snapshot: If modCounts change during a traversal + * of segments computing size or checking containsValue, then + * we might have an inconsistent view of state so (usually) + * must retry. + */ + transient int modCount; + + /** + * The table is rehashed when its size exceeds this threshold. + * (The value of this field is always <tt>(int)(capacity * + * loadFactor)</tt>.) + */ + transient int threshold; + + /** + * The per-segment table. + */ + transient volatile HashEntry<K,V>[] table; + + /** + * The load factor for the hash table. Even though this value + * is same for all segments, it is replicated to avoid needing + * links to outer object. + * @serial + */ + final float loadFactor; + + /** + * The collected weak-key reference queue for this segment. + * This should be (re)initialized whenever table is assigned, + */ + transient volatile ReferenceQueue<Object> refQueue; + + final ReferenceType keyType; + + final ReferenceType valueType; + + final boolean identityComparisons; + + Segment(int initialCapacity, float lf, ReferenceType keyType, + ReferenceType valueType, boolean identityComparisons) { + loadFactor = lf; + this.keyType = keyType; + this.valueType = valueType; + this.identityComparisons = identityComparisons; + setTable(HashEntry.<K,V>newArray(initialCapacity)); + } + + @SuppressWarnings("unchecked") + static final <K,V> Segment<K,V>[] newArray(int i) { + return new Segment[i]; + } + + private boolean keyEq(Object src, Object dest) { + return identityComparisons ? src == dest : src.equals(dest); + } + + /** + * Sets table to new HashEntry array. + * Call only while holding lock or in constructor. + */ + void setTable(HashEntry<K,V>[] newTable) { + threshold = (int)(newTable.length * loadFactor); + table = newTable; + refQueue = new ReferenceQueue<Object>(); + } + + /** + * Returns properly casted first entry of bin for given hash. + */ + HashEntry<K,V> getFirst(int hash) { + HashEntry<K,V>[] tab = table; + return tab[hash & (tab.length - 1)]; + } + + HashEntry<K,V> newHashEntry(K key, int hash, HashEntry<K, V> next, V value) { + return new HashEntry<K,V>(key, hash, next, value, keyType, valueType, refQueue); + } + + /** + * Reads value field of an entry under lock. Called if value + * field ever appears to be null. This is possible only if a + * compiler happens to reorder a HashEntry initialization with + * its table assignment, which is legal under memory model + * but is not known to ever occur. + */ + V readValueUnderLock(HashEntry<K,V> e) { + lock(); + try { + removeStale(); + return e.value(); + } finally { + unlock(); + } + } + + /* Specialized implementations of map methods */ + + V get(Object key, int hash) { + if (count != 0) { // read-volatile + HashEntry<K,V> e = getFirst(hash); + while (e != null) { + if (e.hash == hash && keyEq(key, e.key())) { + Object opaque = e.valueRef; + if (opaque != null) + return e.dereferenceValue(opaque); + + return readValueUnderLock(e); // recheck + } + e = e.next; + } + } + return null; + } + + boolean containsKey(Object key, int hash) { + if (count != 0) { // read-volatile + HashEntry<K,V> e = getFirst(hash); + while (e != null) { + if (e.hash == hash && keyEq(key, e.key())) + return true; + e = e.next; + } + } + return false; + } + + boolean containsValue(Object value) { + if (count != 0) { // read-volatile + HashEntry<K,V>[] tab = table; + int len = tab.length; + for (int i = 0 ; i < len; i++) { + for (HashEntry<K,V> e = tab[i]; e != null; e = e.next) { + Object opaque = e.valueRef; + V v; + + if (opaque == null) + v = readValueUnderLock(e); // recheck + else + v = e.dereferenceValue(opaque); + + if (value.equals(v)) + return true; + } + } + } + return false; + } + + boolean replace(K key, int hash, V oldValue, V newValue) { + lock(); + try { + removeStale(); + HashEntry<K,V> e = getFirst(hash); + while (e != null && (e.hash != hash || !keyEq(key, e.key()))) + e = e.next; + + boolean replaced = false; + if (e != null && oldValue.equals(e.value())) { + replaced = true; + e.setValue(newValue, valueType, refQueue); + } + return replaced; + } finally { + unlock(); + } + } + + V replace(K key, int hash, V newValue) { + lock(); + try { + removeStale(); + HashEntry<K,V> e = getFirst(hash); + while (e != null && (e.hash != hash || !keyEq(key, e.key()))) + e = e.next; + + V oldValue = null; + if (e != null) { + oldValue = e.value(); + e.setValue(newValue, valueType, refQueue); + } + return oldValue; + } finally { + unlock(); + } + } + + + V put(K key, int hash, V value, boolean onlyIfAbsent) { + lock(); + try { + removeStale(); + int c = count; + if (c++ > threshold) {// ensure capacity + int reduced = rehash(); + if (reduced > 0) // adjust from possible weak cleanups + count = (c -= reduced) - 1; // write-volatile + } + + HashEntry<K,V>[] tab = table; + int index = hash & (tab.length - 1); + HashEntry<K,V> first = tab[index]; + HashEntry<K,V> e = first; + while (e != null && (e.hash != hash || !keyEq(key, e.key()))) + e = e.next; + + V oldValue; + if (e != null) { + oldValue = e.value(); + if (!onlyIfAbsent) + e.setValue(value, valueType, refQueue); + } + else { + oldValue = null; + ++modCount; + tab[index] = newHashEntry(key, hash, first, value); + count = c; // write-volatile + } + return oldValue; + } finally { + unlock(); + } + } + + int rehash() { + HashEntry<K,V>[] oldTable = table; + int oldCapacity = oldTable.length; + if (oldCapacity >= MAXIMUM_CAPACITY) + return 0; + + /* + * Reclassify nodes in each list to new Map. Because we are + * using power-of-two expansion, the elements from each bin + * must either stay at same index, or move with a power of two + * offset. We eliminate unnecessary node creation by catching + * cases where old nodes can be reused because their next + * fields won't change. Statistically, at the default + * threshold, only about one-sixth of them need cloning when + * a table doubles. The nodes they replace will be garbage + * collectable as soon as they are no longer referenced by any + * reader thread that may be in the midst of traversing table + * right now. + */ + + HashEntry<K,V>[] newTable = HashEntry.newArray(oldCapacity<<1); + threshold = (int)(newTable.length * loadFactor); + int sizeMask = newTable.length - 1; + int reduce = 0; + for (int i = 0; i < oldCapacity ; i++) { + // We need to guarantee that any existing reads of old Map can + // proceed. So we cannot yet null out each bin. + HashEntry<K,V> e = oldTable[i]; + + if (e != null) { + HashEntry<K,V> next = e.next; + int idx = e.hash & sizeMask; + + // Single node on list + if (next == null) + newTable[idx] = e; + + else { + // Reuse trailing consecutive sequence at same slot + HashEntry<K,V> lastRun = e; + int lastIdx = idx; + for (HashEntry<K,V> last = next; + last != null; + last = last.next) { + int k = last.hash & sizeMask; + if (k != lastIdx) { + lastIdx = k; + lastRun = last; + } + } + newTable[lastIdx] = lastRun; + // Clone all remaining nodes + for (HashEntry<K,V> p = e; p != lastRun; p = p.next) { + // Skip GC'd weak refs + K key = p.key(); + if (key == null) { + reduce++; + continue; + } + int k = p.hash & sizeMask; + HashEntry<K,V> n = newTable[k]; + newTable[k] = newHashEntry(key, p.hash, n, p.value()); + } + } + } + } + table = newTable; + return reduce; + } + + /** + * Remove; match on key only if value null, else match both. + */ + V remove(Object key, int hash, Object value, boolean refRemove) { + lock(); + try { + if (!refRemove) + removeStale(); + int c = count - 1; + HashEntry<K,V>[] tab = table; + int index = hash & (tab.length - 1); + HashEntry<K,V> first = tab[index]; + HashEntry<K,V> e = first; + // a ref remove operation compares the Reference instance + while (e != null && key != e.keyRef + && (refRemove || hash != e.hash || !keyEq(key, e.key()))) + e = e.next; + + V oldValue = null; + if (e != null) { + V v = e.value(); + if (value == null || value.equals(v)) { + oldValue = v; + // All entries following removed node can stay + // in list, but all preceding ones need to be + // cloned. + ++modCount; + HashEntry<K,V> newFirst = e.next; + for (HashEntry<K,V> p = first; p != e; p = p.next) { + K pKey = p.key(); + if (pKey == null) { // Skip GC'd keys + c--; + continue; + } + + newFirst = newHashEntry(pKey, p.hash, newFirst, p.value()); + } + tab[index] = newFirst; + count = c; // write-volatile + } + } + return oldValue; + } finally { + unlock(); + } + } + + final void removeStale() { + KeyReference ref; + while ((ref = (KeyReference) refQueue.poll()) != null) { + remove(ref.keyRef(), ref.keyHash(), null, true); + } + } + + void clear() { + if (count != 0) { + lock(); + try { + HashEntry<K,V>[] tab = table; + for (int i = 0; i < tab.length ; i++) + tab[i] = null; + ++modCount; + // replace the reference queue to avoid unnecessary stale cleanups + refQueue = new ReferenceQueue<Object>(); + count = 0; // write-volatile + } finally { + unlock(); + } + } + } + } + + + + /* ---------------- Public operations -------------- */ + + /** + * Creates a new, empty map with the specified initial + * capacity, reference types, load factor and concurrency level. + * + * Behavioral changing options such as {@link Option#IDENTITY_COMPARISONS} + * can also be specified. + * + * @param initialCapacity the initial capacity. The implementation + * performs internal sizing to accommodate this many elements. + * @param loadFactor the load factor threshold, used to control resizing. + * Resizing may be performed when the average number of elements per + * bin exceeds this threshold. + * @param concurrencyLevel the estimated number of concurrently + * updating threads. The implementation performs internal sizing + * to try to accommodate this many threads. + * @param keyType the reference type to use for keys + * @param valueType the reference type to use for values + * @param options the behavioral options + * @throws IllegalArgumentException if the initial capacity is + * negative or the load factor or concurrencyLevel are + * nonpositive. + */ + public ConcurrentReferenceHashMap(int initialCapacity, + float loadFactor, int concurrencyLevel, + ReferenceType keyType, ReferenceType valueType, + EnumSet<Option> options) { + if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) + throw new IllegalArgumentException(); + + if (concurrencyLevel > MAX_SEGMENTS) + concurrencyLevel = MAX_SEGMENTS; + + // Find power-of-two sizes best matching arguments + int sshift = 0; + int ssize = 1; + while (ssize < concurrencyLevel) { + ++sshift; + ssize <<= 1; + } + segmentShift = 32 - sshift; + segmentMask = ssize - 1; + this.segments = Segment.newArray(ssize); + + if (initialCapacity > MAXIMUM_CAPACITY) + initialCapacity = MAXIMUM_CAPACITY; + int c = initialCapacity / ssize; + if (c * ssize < initialCapacity) + ++c; + int cap = 1; + while (cap < c) + cap <<= 1; + + identityComparisons = options != null && options.contains(Option.IDENTITY_COMPARISONS); + + for (int i = 0; i < this.segments.length; ++i) + this.segments[i] = new Segment<K,V>(cap, loadFactor, + keyType, valueType, identityComparisons); + } + + /** + * Creates a new, empty map with the specified initial + * capacity, load factor and concurrency level. + * + * @param initialCapacity the initial capacity. The implementation + * performs internal sizing to accommodate this many elements. + * @param loadFactor the load factor threshold, used to control resizing. + * Resizing may be performed when the average number of elements per + * bin exceeds this threshold. + * @param concurrencyLevel the estimated number of concurrently + * updating threads. The implementation performs internal sizing + * to try to accommodate this many threads. + * @throws IllegalArgumentException if the initial capacity is + * negative or the load factor or concurrencyLevel are + * nonpositive. + */ + public ConcurrentReferenceHashMap(int initialCapacity, + float loadFactor, int concurrencyLevel) { + this(initialCapacity, loadFactor, concurrencyLevel, + DEFAULT_KEY_TYPE, DEFAULT_VALUE_TYPE, null); + } + + /** + * Creates a new, empty map with the specified initial capacity + * and load factor and with the default reference types (weak keys, + * strong values), and concurrencyLevel (16). + * + * @param initialCapacity The implementation performs internal + * sizing to accommodate this many elements. + * @param loadFactor the load factor threshold, used to control resizing. + * Resizing may be performed when the average number of elements per + * bin exceeds this threshold. + * @throws IllegalArgumentException if the initial capacity of + * elements is negative or the load factor is nonpositive + * + * @since 1.6 + */ + public ConcurrentReferenceHashMap(int initialCapacity, float loadFactor) { + this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL); + } + + + /** + * Creates a new, empty map with the specified initial capacity, + * reference types and with default load factor (0.75) and concurrencyLevel (16). + * + * @param initialCapacity the initial capacity. The implementation + * performs internal sizing to accommodate this many elements. + * @param keyType the reference type to use for keys + * @param valueType the reference type to use for values + * @throws IllegalArgumentException if the initial capacity of + * elements is negative. + */ + public ConcurrentReferenceHashMap(int initialCapacity, + ReferenceType keyType, ReferenceType valueType) { + this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL, + keyType, valueType, null); + } + + /** + * Creates a new, empty map with the specified initial capacity, + * and with default reference types (weak keys, strong values), + * load factor (0.75) and concurrencyLevel (16). + * + * @param initialCapacity the initial capacity. The implementation + * performs internal sizing to accommodate this many elements. + * @throws IllegalArgumentException if the initial capacity of + * elements is negative. + */ + public ConcurrentReferenceHashMap(int initialCapacity) { + this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); + } + + /** + * Creates a new, empty map with a default initial capacity (16), + * reference types (weak keys, strong values), default + * load factor (0.75) and concurrencyLevel (16). + */ + public ConcurrentReferenceHashMap() { + this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); + } + + /** + * Creates a new map with the same mappings as the given map. + * The map is created with a capacity of 1.5 times the number + * of mappings in the given map or 16 (whichever is greater), + * and a default load factor (0.75) and concurrencyLevel (16). + * + * @param m the map + */ + public ConcurrentReferenceHashMap(Map<? extends K, ? extends V> m) { + this(Math.max((int) (m.size() / DEFAULT_LOAD_FACTOR) + 1, + DEFAULT_INITIAL_CAPACITY), + DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); + putAll(m); + } + + /** + * Returns <tt>true</tt> if this map contains no key-value mappings. + * + * @return <tt>true</tt> if this map contains no key-value mappings + */ + public boolean isEmpty() { + final Segment<K,V>[] segments = this.segments; + /* + * We keep track of per-segment modCounts to avoid ABA + * problems in which an element in one segment was added and + * in another removed during traversal, in which case the + * table was never actually empty at any point. Note the + * similar use of modCounts in the size() and containsValue() + * methods, which are the only other methods also susceptible + * to ABA problems. + */ + int[] mc = new int[segments.length]; + int mcsum = 0; + for (int i = 0; i < segments.length; ++i) { + if (segments[i].count != 0) + return false; + else + mcsum += mc[i] = segments[i].modCount; + } + // If mcsum happens to be zero, then we know we got a snapshot + // before any modifications at all were made. This is + // probably common enough to bother tracking. + if (mcsum != 0) { + for (int i = 0; i < segments.length; ++i) { + if (segments[i].count != 0 || + mc[i] != segments[i].modCount) + return false; + } + } + return true; + } + + /** + * Returns the number of key-value mappings in this map. If the + * map contains more than <tt>Integer.MAX_VALUE</tt> elements, returns + * <tt>Integer.MAX_VALUE</tt>. + * + * @return the number of key-value mappings in this map + */ + public int size() { + final Segment<K,V>[] segments = this.segments; + long sum = 0; + long check = 0; + int[] mc = new int[segments.length]; + // Try a few times to get accurate count. On failure due to + // continuous async changes in table, resort to locking. + for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) { + check = 0; + sum = 0; + int mcsum = 0; + for (int i = 0; i < segments.length; ++i) { + sum += segments[i].count; + mcsum += mc[i] = segments[i].modCount; + } + if (mcsum != 0) { + for (int i = 0; i < segments.length; ++i) { + check += segments[i].count; + if (mc[i] != segments[i].modCount) { + check = -1; // force retry + break; + } + } + } + if (check == sum) + break; + } + if (check != sum) { // Resort to locking all segments + sum = 0; + for (int i = 0; i < segments.length; ++i) + segments[i].lock(); + for (int i = 0; i < segments.length; ++i) + sum += segments[i].count; + for (int i = 0; i < segments.length; ++i) + segments[i].unlock(); + } + if (sum > Integer.MAX_VALUE) + return Integer.MAX_VALUE; + else + return (int)sum; + } + + /** + * Returns the value to which the specified key is mapped, + * or {@code null} if this map contains no mapping for the key. + * + * <p>More formally, if this map contains a mapping from a key + * {@code k} to a value {@code v} such that {@code key.equals(k)}, + * then this method returns {@code v}; otherwise it returns + * {@code null}. (There can be at most one such mapping.) + * + * @throws NullPointerException if the specified key is null + */ + public V get(Object key) { + int hash = hashOf(key); + return segmentFor(hash).get(key, hash); + } + + /** + * Tests if the specified object is a key in this table. + * + * @param key possible key + * @return <tt>true</tt> if and only if the specified object + * is a key in this table, as determined by the + * <tt>equals</tt> method; <tt>false</tt> otherwise. + * @throws NullPointerException if the specified key is null + */ + public boolean containsKey(Object key) { + int hash = hashOf(key); + return segmentFor(hash).containsKey(key, hash); + } + + /** + * Returns <tt>true</tt> if this map maps one or more keys to the + * specified value. Note: This method requires a full internal + * traversal of the hash table, and so is much slower than + * method <tt>containsKey</tt>. + * + * @param value value whose presence in this map is to be tested + * @return <tt>true</tt> if this map maps one or more keys to the + * specified value + * @throws NullPointerException if the specified value is null + */ + public boolean containsValue(Object value) { + if (value == null) + throw new NullPointerException(); + + // See explanation of modCount use above + + final Segment<K,V>[] segments = this.segments; + int[] mc = new int[segments.length]; + + // Try a few times without locking + for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) { + int sum = 0; + int mcsum = 0; + for (int i = 0; i < segments.length; ++i) { + int c = segments[i].count; + mcsum += mc[i] = segments[i].modCount; + if (segments[i].containsValue(value)) + return true; + } + boolean cleanSweep = true; + if (mcsum != 0) { + for (int i = 0; i < segments.length; ++i) { + int c = segments[i].count; + if (mc[i] != segments[i].modCount) { + cleanSweep = false; + break; + } + } + } + if (cleanSweep) + return false; + } + // Resort to locking all segments + for (int i = 0; i < segments.length; ++i) + segments[i].lock(); + boolean found = false; + try { + for (int i = 0; i < segments.length; ++i) { + if (segments[i].containsValue(value)) { + found = true; + break; + } + } + } finally { + for (int i = 0; i < segments.length; ++i) + segments[i].unlock(); + } + return found; + } + + /** + * Legacy method testing if some key maps into the specified value + * in this table. This method is identical in functionality to + * {@link #containsValue}, and exists solely to ensure + * full compatibility with class {@link java.util.Hashtable}, + * which supported this method prior to introduction of the + * Java Collections framework. + + * @param value a value to search for + * @return <tt>true</tt> if and only if some key maps to the + * <tt>value</tt> argument in this table as + * determined by the <tt>equals</tt> method; + * <tt>false</tt> otherwise + * @throws NullPointerException if the specified value is null + */ + public boolean contains(Object value) { + return containsValue(value); + } + + /** + * Maps the specified key to the specified value in this table. + * Neither the key nor the value can be null. + * + * <p> The value can be retrieved by calling the <tt>get</tt> method + * with a key that is equal to the original key. + * + * @param key key with which the specified value is to be associated + * @param value value to be associated with the specified key + * @return the previous value associated with <tt>key</tt>, or + * <tt>null</tt> if there was no mapping for <tt>key</tt> + * @throws NullPointerException if the specified key or value is null + */ + public V put(K key, V value) { + if (value == null) + throw new NullPointerException(); + int hash = hashOf(key); + return segmentFor(hash).put(key, hash, value, false); + } + + /** + * {@inheritDoc} + * + * @return the previous value associated with the specified key, + * or <tt>null</tt> if there was no mapping for the key + * @throws NullPointerException if the specified key or value is null + */ + public V putIfAbsent(K key, V value) { + if (value == null) + throw new NullPointerException(); + int hash = hashOf(key); + return segmentFor(hash).put(key, hash, value, true); + } + + /** + * Copies all of the mappings from the specified map to this one. + * These mappings replace any mappings that this map had for any of the + * keys currently in the specified map. + * + * @param m mappings to be stored in this map + */ + public void putAll(Map<? extends K, ? extends V> m) { + for (Map.Entry<? extends K, ? extends V> e : m.entrySet()) + put(e.getKey(), e.getValue()); + } + + /** + * Removes the key (and its corresponding value) from this map. + * This method does nothing if the key is not in the map. + * + * @param key the key that needs to be removed + * @return the previous value associated with <tt>key</tt>, or + * <tt>null</tt> if there was no mapping for <tt>key</tt> + * @throws NullPointerException if the specified key is null + */ + public V remove(Object key) { + int hash = hashOf(key); + return segmentFor(hash).remove(key, hash, null, false); + } + + /** + * {@inheritDoc} + * + * @throws NullPointerException if the specified key is null + */ + public boolean remove(Object key, Object value) { + int hash = hashOf(key); + if (value == null) + return false; + return segmentFor(hash).remove(key, hash, value, false) != null; + } + + /** + * {@inheritDoc} + * + * @throws NullPointerException if any of the arguments are null + */ + public boolean replace(K key, V oldValue, V newValue) { + if (oldValue == null || newValue == null) + throw new NullPointerException(); + int hash = hashOf(key); + return segmentFor(hash).replace(key, hash, oldValue, newValue); + } + + /** + * {@inheritDoc} + * + * @return the previous value associated with the specified key, + * or <tt>null</tt> if there was no mapping for the key + * @throws NullPointerException if the specified key or value is null + */ + public V replace(K key, V value) { + if (value == null) + throw new NullPointerException(); + int hash = hashOf(key); + return segmentFor(hash).replace(key, hash, value); + } + + /** + * Removes all of the mappings from this map. + */ + public void clear() { + for (int i = 0; i < segments.length; ++i) + segments[i].clear(); + } + + /** + * Removes any stale entries whose keys have been finalized. Use of this + * method is normally not necessary since stale entries are automatically + * removed lazily, when blocking operations are required. However, there + * are some cases where this operation should be performed eagerly, such + * as cleaning up old references to a ClassLoader in a multi-classloader + * environment. + * + * Note: this method will acquire locks, one at a time, across all segments + * of this table, so if it is to be used, it should be used sparingly. + */ + public void purgeStaleEntries() { + for (int i = 0; i < segments.length; ++i) + segments[i].removeStale(); + } + + + /** + * Returns a {@link Set} view of the keys contained in this map. + * The set is backed by the map, so changes to the map are + * reflected in the set, and vice-versa. The set supports element + * removal, which removes the corresponding mapping from this map, + * via the <tt>Iterator.remove</tt>, <tt>Set.remove</tt>, + * <tt>removeAll</tt>, <tt>retainAll</tt>, and <tt>clear</tt> + * operations. It does not support the <tt>add</tt> or + * <tt>addAll</tt> operations. + * + * <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator + * that will never throw {@link ConcurrentModificationException}, + * and guarantees to traverse elements as they existed upon + * construction of the iterator, and may (but is not guaranteed to) + * reflect any modifications subsequent to construction. + */ + public Set<K> keySet() { + Set<K> ks = keySet; + return (ks != null) ? ks : (keySet = new KeySet()); + } + + /** + * Returns a {@link Collection} view of the values contained in this map. + * The collection is backed by the map, so changes to the map are + * reflected in the collection, and vice-versa. The collection + * supports element removal, which removes the corresponding + * mapping from this map, via the <tt>Iterator.remove</tt>, + * <tt>Collection.remove</tt>, <tt>removeAll</tt>, + * <tt>retainAll</tt>, and <tt>clear</tt> operations. It does not + * support the <tt>add</tt> or <tt>addAll</tt> operations. + * + * <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator + * that will never throw {@link ConcurrentModificationException}, + * and guarantees to traverse elements as they existed upon + * construction of the iterator, and may (but is not guaranteed to) + * reflect any modifications subsequent to construction. + */ + public Collection<V> values() { + Collection<V> vs = values; + return (vs != null) ? vs : (values = new Values()); + } + + /** + * Returns a {@link Set} view of the mappings contained in this map. + * The set is backed by the map, so changes to the map are + * reflected in the set, and vice-versa. The set supports element + * removal, which removes the corresponding mapping from the map, + * via the <tt>Iterator.remove</tt>, <tt>Set.remove</tt>, + * <tt>removeAll</tt>, <tt>retainAll</tt>, and <tt>clear</tt> + * operations. It does not support the <tt>add</tt> or + * <tt>addAll</tt> operations. + * + * <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator + * that will never throw {@link ConcurrentModificationException}, + * and guarantees to traverse elements as they existed upon + * construction of the iterator, and may (but is not guaranteed to) + * reflect any modifications subsequent to construction. + */ + public Set<Map.Entry<K,V>> entrySet() { + Set<Map.Entry<K,V>> es = entrySet; + return (es != null) ? es : (entrySet = new EntrySet()); + } + + /** + * Returns an enumeration of the keys in this table. + * + * @return an enumeration of the keys in this table + * @see #keySet() + */ + public Enumeration<K> keys() { + return new KeyIterator(); + } + + /** + * Returns an enumeration of the values in this table. + * + * @return an enumeration of the values in this table + * @see #values() + */ + public Enumeration<V> elements() { + return new ValueIterator(); + } + + /* ---------------- Iterator Support -------------- */ + + abstract class HashIterator { + int nextSegmentIndex; + int nextTableIndex; + HashEntry<K,V>[] currentTable; + HashEntry<K, V> nextEntry; + HashEntry<K, V> lastReturned; + K currentKey; // Strong reference to weak key (prevents gc) + + HashIterator() { + nextSegmentIndex = segments.length - 1; + nextTableIndex = -1; + advance(); + } + + public boolean hasMoreElements() { return hasNext(); } + + final void advance() { + if (nextEntry != null && (nextEntry = nextEntry.next) != null) + return; + + while (nextTableIndex >= 0) { + if ( (nextEntry = currentTable[nextTableIndex--]) != null) + return; + } + + while (nextSegmentIndex >= 0) { + Segment<K,V> seg = segments[nextSegmentIndex--]; + if (seg.count != 0) { + currentTable = seg.table; + for (int j = currentTable.length - 1; j >= 0; --j) { + if ( (nextEntry = currentTable[j]) != null) { + nextTableIndex = j - 1; + return; + } + } + } + } + } + + public boolean hasNext() { + while (nextEntry != null) { + if (nextEntry.key() != null) + return true; + advance(); + } + + return false; + } + + HashEntry<K,V> nextEntry() { + do { + if (nextEntry == null) + throw new NoSuchElementException(); + + lastReturned = nextEntry; + currentKey = lastReturned.key(); + advance(); + } while (currentKey == null); // Skip GC'd keys + + return lastReturned; + } + + public void remove() { + if (lastReturned == null) + throw new IllegalStateException(); + ConcurrentReferenceHashMap.this.remove(currentKey); + lastReturned = null; + } + } + + final class KeyIterator + extends HashIterator + implements Iterator<K>, Enumeration<K> + { + public K next() { return super.nextEntry().key(); } + public K nextElement() { return super.nextEntry().key(); } + } + + final class ValueIterator + extends HashIterator + implements Iterator<V>, Enumeration<V> + { + public V next() { return super.nextEntry().value(); } + public V nextElement() { return super.nextEntry().value(); } + } + + /* + * This class is needed for JDK5 compatibility. + */ + static class SimpleEntry<K, V> implements Entry<K, V>, + java.io.Serializable { + private static final long serialVersionUID = -8499721149061103585L; + + private final K key; + private V value; + + public SimpleEntry(K key, V value) { + this.key = key; + this.value = value; + } + + public SimpleEntry(Entry<? extends K, ? extends V> entry) { + this.key = entry.getKey(); + this.value = entry.getValue(); + } + + public K getKey() { + return key; + } + + public V getValue() { + return value; + } + + public V setValue(V value) { + V oldValue = this.value; + this.value = value; + return oldValue; + } + + public boolean equals(Object o) { + if (!(o instanceof Map.Entry)) + return false; + @SuppressWarnings("unchecked") + Map.Entry e = (Map.Entry) o; + return eq(key, e.getKey()) && eq(value, e.getValue()); + } + + public int hashCode() { + return (key == null ? 0 : key.hashCode()) + ^ (value == null ? 0 : value.hashCode()); + } + + public String toString() { + return key + "=" + value; + } + + private static boolean eq(Object o1, Object o2) { + return o1 == null ? o2 == null : o1.equals(o2); + } + } + + + /** + * Custom Entry class used by EntryIterator.next(), that relays setValue + * changes to the underlying map. + */ + final class WriteThroughEntry extends SimpleEntry<K,V> + { + private static final long serialVersionUID = -7900634345345313646L; + + WriteThroughEntry(K k, V v) { + super(k,v); + } + + /** + * Set our entry's value and write through to the map. The + * value to return is somewhat arbitrary here. Since a + * WriteThroughEntry does not necessarily track asynchronous + * changes, the most recent "previous" value could be + * different from what we return (or could even have been + * removed in which case the put will re-establish). We do not + * and cannot guarantee more. + */ + public V setValue(V value) { + if (value == null) throw new NullPointerException(); + V v = super.setValue(value); + ConcurrentReferenceHashMap.this.put(getKey(), value); + return v; + } + } + + final class EntryIterator + extends HashIterator + implements Iterator<Entry<K,V>> + { + public Map.Entry<K,V> next() { + HashEntry<K,V> e = super.nextEntry(); + return new WriteThroughEntry(e.key(), e.value()); + } + } + + final class KeySet extends AbstractSet<K> { + public Iterator<K> iterator() { + return new KeyIterator(); + } + public int size() { + return ConcurrentReferenceHashMap.this.size(); + } + public boolean isEmpty() { + return ConcurrentReferenceHashMap.this.isEmpty(); + } + public boolean contains(Object o) { + return ConcurrentReferenceHashMap.this.containsKey(o); + } + public boolean remove(Object o) { + return ConcurrentReferenceHashMap.this.remove(o) != null; + } + public void clear() { + ConcurrentReferenceHashMap.this.clear(); + } + } + + final class Values extends AbstractCollection<V> { + public Iterator<V> iterator() { + return new ValueIterator(); + } + public int size() { + return ConcurrentReferenceHashMap.this.size(); + } + public boolean isEmpty() { + return ConcurrentReferenceHashMap.this.isEmpty(); + } + public boolean contains(Object o) { + return ConcurrentReferenceHashMap.this.containsValue(o); + } + public void clear() { + ConcurrentReferenceHashMap.this.clear(); + } + } + + final class EntrySet extends AbstractSet<Map.Entry<K,V>> { + public Iterator<Map.Entry<K,V>> iterator() { + return new EntryIterator(); + } + public boolean contains(Object o) { + if (!(o instanceof Map.Entry)) + return false; + Map.Entry<?,?> e = (Map.Entry<?,?>)o; + V v = ConcurrentReferenceHashMap.this.get(e.getKey()); + return v != null && v.equals(e.getValue()); + } + public boolean remove(Object o) { + if (!(o instanceof Map.Entry)) + return false; + Map.Entry<?,?> e = (Map.Entry<?,?>)o; + return ConcurrentReferenceHashMap.this.remove(e.getKey(), e.getValue()); + } + public int size() { + return ConcurrentReferenceHashMap.this.size(); + } + public boolean isEmpty() { + return ConcurrentReferenceHashMap.this.isEmpty(); + } + public void clear() { + ConcurrentReferenceHashMap.this.clear(); + } + } + + /* ---------------- Serialization Support -------------- */ + + /** + * Save the state of the <tt>ConcurrentReferenceHashMap</tt> instance to a + * stream (i.e., serialize it). + * @param s the stream + * @serialData + * the key (Object) and value (Object) + * for each key-value mapping, followed by a null pair. + * The key-value mappings are emitted in no particular order. + */ + private void writeObject(java.io.ObjectOutputStream s) throws IOException { + s.defaultWriteObject(); + + for (int k = 0; k < segments.length; ++k) { + Segment<K,V> seg = segments[k]; + seg.lock(); + try { + HashEntry<K,V>[] tab = seg.table; + for (int i = 0; i < tab.length; ++i) { + for (HashEntry<K,V> e = tab[i]; e != null; e = e.next) { + K key = e.key(); + if (key == null) // Skip GC'd keys + continue; + + s.writeObject(key); + s.writeObject(e.value()); + } + } + } finally { + seg.unlock(); + } + } + s.writeObject(null); + s.writeObject(null); + } + + /** + * Reconstitute the <tt>ConcurrentReferenceHashMap</tt> instance from a + * stream (i.e., deserialize it). + * @param s the stream + */ + @SuppressWarnings("unchecked") + private void readObject(java.io.ObjectInputStream s) + throws IOException, ClassNotFoundException { + s.defaultReadObject(); + + // Initialize each segment to be minimally sized, and let grow. + for (int i = 0; i < segments.length; ++i) { + segments[i].setTable(new HashEntry[1]); + } + + // Read the keys and values, and put the mappings in the table + for (;;) { + K key = (K) s.readObject(); + V value = (V) s.readObject(); + if (key == null) + break; + put(key, value); + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/EndpointImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,463 @@ +package org.jboss.remoting; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.ref.WeakReference; +import java.net.URI; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +import org.jboss.remoting.Client; +import org.jboss.remoting.ClientSource; +import org.jboss.remoting.CloseHandler; +import org.jboss.remoting.Endpoint; +import org.jboss.remoting.EndpointPermission; +import org.jboss.remoting.LocalServiceConfiguration; +import org.jboss.remoting.RemoteServiceConfiguration; +import org.jboss.remoting.RequestListener; +import org.jboss.remoting.ServiceListener; +import org.jboss.remoting.ServiceURI; +import org.jboss.remoting.SimpleCloseable; +import org.jboss.remoting.Version; +import org.jboss.remoting.spi.AbstractHandleableCloseable; +import org.jboss.remoting.spi.AbstractSimpleCloseable; +import org.jboss.remoting.spi.Handle; +import org.jboss.remoting.spi.RequestHandler; +import org.jboss.remoting.spi.RequestHandlerSource; +import org.jboss.xnio.FailedIoFuture; +import org.jboss.xnio.FinishedIoFuture; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.WeakCloseable; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class EndpointImpl extends AbstractHandleableCloseable<Endpoint> implements Endpoint { + + static { + // Print Remoting "greeting" message + Logger.getLogger("org.jboss.remoting").info("JBoss Remoting version %s", Version.VERSION); + } + + static <K, V> Map<K, V> hashMap() { + return new HashMap<K, V>(); + } + + static <T> Set<T> hashSet() { + return new HashSet<T>(); + } + + private static final Logger log = Logger.getLogger("org.jboss.remoting.endpoint"); + + private final String name; + + private final ConcurrentMap<Object, Object> endpointMap = new ConcurrentHashMap<Object, Object>(); + + private final Object serviceLock = new Object(); + private final Map<Object, ServiceListenerRegistration> serviceListenerMap = hashMap(); + private final Set<ServiceRegistration> serviceRegistrations = hashSet(); + + private static final EndpointPermission CREATE_ENDPOINT_PERM = new EndpointPermission("createEndpoint"); + private static final EndpointPermission CREATE_REQUEST_HANDLER_PERM = new EndpointPermission("createRequestHandler"); + private static final EndpointPermission REGISTER_SERVICE_PERM = new EndpointPermission("registerService"); + private static final EndpointPermission CREATE_CLIENT_PERM = new EndpointPermission("createClient"); + private static final EndpointPermission CREATE_CLIENT_SOURCE_PERM = new EndpointPermission("createClientSource"); + private static final EndpointPermission REGISTER_REMOTE_SERVICE_PERM = new EndpointPermission("registerRemoteService"); + private static final EndpointPermission ADD_SERVICE_LISTENER_PERM = new EndpointPermission("addServiceListener"); + + public EndpointImpl(final Executor executor, final String name) { + super(executor); + final SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(CREATE_ENDPOINT_PERM); + } + this.executor = executor; + this.name = name; + } + + private final Executor executor; + + protected Executor getOrderedExecutor() { + return new OrderedExecutor(executor); + } + + protected Executor getExecutor() { + return executor; + } + + // Endpoint implementation + + public String getName() { + return name; + } + + public ConcurrentMap<Object, Object> getAttributes() { + return endpointMap; + } + + public <I, O> Handle<RequestHandler> createRequestHandler(final RequestListener<I, O> requestListener, final Class<I> requestClass, final Class<O> replyClass) throws IOException { + SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(CREATE_REQUEST_HANDLER_PERM); + } + LocalRequestHandler.Config<I, O> config = new LocalRequestHandler.Config<I, O>(requestClass, replyClass); + config.setExecutor(executor); + config.setRequestListener(requestListener); + config.setClientContext(new ClientContextImpl(executor)); + final LocalRequestHandler<I, O> localRequestHandler = new LocalRequestHandler<I, O>(config); + final WeakCloseable lrhCloseable = new WeakCloseable(localRequestHandler); + final Key key = addCloseHandler(new CloseHandler<Endpoint>() { + public void handleClose(final Endpoint closed) { + IoUtils.safeClose(lrhCloseable); + } + }); + localRequestHandler.addCloseHandler(new CloseHandler<RequestHandler>() { + public void handleClose(final RequestHandler closed) { + key.remove(); + } + }); + localRequestHandler.open(); + return localRequestHandler.getHandle(); + } + + public <I, O> Handle<RequestHandlerSource> registerService(final LocalServiceConfiguration<I, O> configuration) throws IOException { + SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(REGISTER_SERVICE_PERM); + } + final String serviceType = configuration.getServiceType(); + final String groupName = configuration.getGroupName(); + final int metric = configuration.getMetric(); + if (serviceType == null) { + throw new NullPointerException("serviceType is null"); + } + if (groupName == null) { + throw new NullPointerException("groupName is null"); + } + if (serviceType.length() == 0) { + throw new IllegalArgumentException("serviceType is empty"); + } + if (groupName.length() == 0) { + throw new IllegalArgumentException("groupName is empty"); + } + if (metric < 0) { + throw new IllegalArgumentException("metric must be greater than or equal to zero"); + } + final LocalRequestHandlerSource.Config<I, O> config = new LocalRequestHandlerSource.Config<I,O>(configuration.getRequestClass(), configuration.getReplyClass()); + config.setRequestListener(configuration.getRequestListener()); + config.setExecutor(executor); + final LocalRequestHandlerSource<I, O> localRequestHandlerSource = new LocalRequestHandlerSource<I, O>(config); + final ServiceRegistration registration = new ServiceRegistration(serviceType, groupName, name, localRequestHandlerSource); + final AbstractSimpleCloseable newHandle = new AbstractSimpleCloseable(executor) { + protected void closeAction() throws IOException { + synchronized (serviceLock) { + serviceRegistrations.remove(registration); + } + } + }; + registration.setHandle(newHandle); + synchronized (serviceLock) { + serviceRegistrations.add(registration); + for (final ServiceListenerRegistration slr : serviceListenerMap.values()) { + final ServiceListener listener = slr.getServiceListener(); + try { + final ServiceListener.ServiceInfo serviceInfo = new ServiceListener.ServiceInfo(); + serviceInfo.setEndpointName(name); + serviceInfo.setGroupName(groupName); + serviceInfo.setServiceType(serviceType); + serviceInfo.setMetric(metric); + serviceInfo.setRegistrationHandle(newHandle); + serviceInfo.setRemote(false); + serviceInfo.setRequestHandlerSource(localRequestHandlerSource); + listener.serviceRegistered(slr.handle, serviceInfo); + } catch (Throwable t) { + logListenerError(t); + } + } + } + final WeakCloseable lrhCloseable = new WeakCloseable(new WeakReference<Closeable>(localRequestHandlerSource)); + final Key key = addCloseHandler(new CloseHandler<Endpoint>() { + public void handleClose(final Endpoint closed) { + IoUtils.safeClose(lrhCloseable); + } + }); + localRequestHandlerSource.addCloseHandler(new CloseHandler<RequestHandlerSource>() { + public void handleClose(final RequestHandlerSource closed) { + key.remove(); + } + }); + localRequestHandlerSource.open(); + return localRequestHandlerSource.getHandle(); + } + + private static void logListenerError(final Throwable t) { + log.error(t, "Service listener threw an exception"); + } + + public <I, O> Client<I, O> createClient(final RequestHandler requestHandler, final Class<I> requestType, final Class<O> replyType) throws IOException { + SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(CREATE_CLIENT_PERM); + } + boolean ok = false; + final Handle<RequestHandler> handle = requestHandler.getHandle(); + try { + final ClientImpl<I, O> client = ClientImpl.create(handle, executor, requestType, replyType); + final WeakCloseable lrhCloseable = new WeakCloseable(new WeakReference<Closeable>(client)); + final Key key = addCloseHandler(new CloseHandler<Endpoint>() { + public void handleClose(final Endpoint closed) { + IoUtils.safeClose(lrhCloseable); + } + }); + client.addCloseHandler(new CloseHandler<Client>() { + public void handleClose(final Client closed) { + key.remove(); + } + }); + ok = true; + return client; + } finally { + if (! ok) { + IoUtils.safeClose(handle); + } + } + } + + public <I, O> ClientSource<I, O> createClientSource(final RequestHandlerSource requestHandlerSource, final Class<I> requestClass, final Class<O> replyClass) throws IOException { + SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(CREATE_CLIENT_SOURCE_PERM); + } + boolean ok = false; + final Handle<RequestHandlerSource> handle = requestHandlerSource.getHandle(); + try { + final ClientSourceImpl<I, O> clientSource = ClientSourceImpl.create(handle, this, requestClass, replyClass); + final WeakCloseable lrhCloseable = new WeakCloseable(new WeakReference<Closeable>(clientSource)); + final Key key = addCloseHandler(new CloseHandler<Endpoint>() { + public void handleClose(final Endpoint closed) { + IoUtils.safeClose(lrhCloseable); + } + }); + clientSource.addCloseHandler(new CloseHandler<ClientSource>() { + public void handleClose(final ClientSource closed) { + key.remove(); + } + }); + ok = true; + return clientSource; + } finally { + if (! ok) { + IoUtils.safeClose(handle); + } + } + } + + public <I, O> IoFuture<ClientSource<I, O>> locateService(final URI serviceUri, final Class<I> requestClass, final Class<O> replyClass) throws IllegalArgumentException { + if (serviceUri == null) { + throw new NullPointerException("serviceUri is null"); + } + if (! ServiceURI.isRemotingServiceUri(serviceUri)) { + throw new IllegalArgumentException("Not a valid remoting service URI"); + } + final String endpointName = ServiceURI.getEndpointName(serviceUri); + final String groupName = ServiceURI.getGroupName(serviceUri); + final String serviceType = ServiceURI.getServiceType(serviceUri); + synchronized (serviceLock) { + int bestMetric = Integer.MAX_VALUE; + List<ServiceRegistration> candidates = new ArrayList<ServiceRegistration>(); + for (ServiceRegistration svc : serviceRegistrations) { + if (svc.matches(serviceType, groupName, endpointName)) { + final int metric = svc.getMetric(); + if (metric < bestMetric) { + candidates.clear(); + candidates.add(svc); + } else if (metric == bestMetric) { + candidates.add(svc); + } + } + } + final int size = candidates.size(); + if (size == 0) { + final FutureClientSource<I, O> futureClientSource = new FutureClientSource<I, O>(); + final SimpleCloseable listenerHandle = addServiceListener(new ServiceListener() { + public void serviceRegistered(final SimpleCloseable listenerHandle, final ServiceInfo info) { + final String addedEndpointName = info.getEndpointName(); + final String addedServiceType = info.getServiceType(); + final String addedGroupName = info.getGroupName(); + final RequestHandlerSource requestHandlerSource = info.getRequestHandlerSource(); + if (endpointName != null && endpointName.length() > 0 && !endpointName.equals(addedEndpointName)) { + // no match + return; + } + if (serviceType != null && serviceType.length() > 0 && !serviceType.equals(addedServiceType)) { + // no match + return; + } + if (groupName != null && groupName.length() > 0 && !groupName.equals(addedGroupName)) { + // no match + return; + } + try { + // match! + final ClientSource<I, O> clientSource = createClientSource(requestHandlerSource, requestClass, replyClass); + futureClientSource.setResult(clientSource); + } catch (IOException e) { + futureClientSource.setException(e); + } finally { + IoUtils.safeClose(listenerHandle); + } + } + }, true); + futureClientSource.setListenerHandle(listenerHandle); + return futureClientSource; + } + final RequestHandlerSource handlerSource; + if (size == 1) { + handlerSource = candidates.get(0).getHandlerSource(); + } else { + int idx = (int) ((double) candidates.size() * Math.random()); + handlerSource = candidates.get(idx).getHandlerSource(); + } + try { + return new FinishedIoFuture<ClientSource<I,O>>(createClientSource(handlerSource, requestClass, replyClass)); + } catch (IOException e) { + return new FailedIoFuture<ClientSource<I,O>>(e); + } + } + } + + public SimpleCloseable registerRemoteService(final RemoteServiceConfiguration configuration) throws IllegalArgumentException, IOException { + SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(REGISTER_REMOTE_SERVICE_PERM); + } + final RequestHandlerSource handlerSource = configuration.getRequestHandlerSource(); + final String serviceType = configuration.getServiceType(); + final String groupName = configuration.getGroupName(); + final String endpointName = configuration.getEndpointName(); + final int metric = configuration.getMetric(); + if (handlerSource == null) { + throw new NullPointerException("handlerSource is null"); + } + if (serviceType == null) { + throw new NullPointerException("serviceType is null"); + } + if (groupName == null) { + throw new NullPointerException("groupName is null"); + } + if (endpointName == null) { + throw new NullPointerException("endpointName is null"); + } + if (serviceType.length() == 0) { + throw new IllegalArgumentException("serviceType is empty"); + } + if (groupName.length() == 0) { + throw new IllegalArgumentException("groupName is empty"); + } + if (endpointName.length() == 0) { + throw new IllegalArgumentException("endpointName is empty"); + } + if (endpointName.equals(name)) { + throw new IllegalArgumentException("remote endpoint has the same name as the local endpoint"); + } + if (metric < 1) { + throw new IllegalArgumentException("metric must be greater than zero"); + } + final ServiceRegistration registration = new ServiceRegistration(serviceType, groupName, endpointName, metric, handlerSource); + final AbstractSimpleCloseable newHandle = new AbstractSimpleCloseable(executor) { + protected void closeAction() throws IOException { + synchronized (serviceLock) { + serviceRegistrations.remove(registration); + } + } + }; + registration.setHandle(newHandle); + synchronized (serviceLock) { + serviceRegistrations.add(registration); + for (final ServiceListenerRegistration slr : serviceListenerMap.values()) { + final ServiceListener listener = slr.getServiceListener(); + try { + final ServiceListener.ServiceInfo info = new ServiceListener.ServiceInfo(); + info.setEndpointName(endpointName); + info.setGroupName(groupName); + info.setMetric(metric); + info.setRegistrationHandle(newHandle); + info.setRemote(true); + info.setRequestHandlerSource(handlerSource); + info.setServiceType(serviceType); + listener.serviceRegistered(slr.handle, info); + } catch (Throwable t) { + logListenerError(t); + } + } + } + return newHandle; + } + + public SimpleCloseable addServiceListener(final ServiceListener serviceListener, final boolean onlyNew) { + SecurityManager sm = System.getSecurityManager(); + if (sm != null) { + sm.checkPermission(ADD_SERVICE_LISTENER_PERM); + } + final Object key = new Object(); + synchronized (serviceLock) { + final ServiceListenerRegistration registration = new ServiceListenerRegistration(serviceListener); + serviceListenerMap.put(key, registration); + final AbstractSimpleCloseable handle = new AbstractSimpleCloseable(executor) { + protected void closeAction() throws IOException { + synchronized (serviceLock) { + serviceListenerMap.remove(key); + } + } + }; + registration.setHandle(handle); + if (! onlyNew) { + for (final ServiceRegistration reg : serviceRegistrations) { + try { + final ServiceListener.ServiceInfo info = new ServiceListener.ServiceInfo(); + info.setEndpointName(reg.getEndpointName()); + info.setGroupName(reg.getGroupName()); + info.setMetric(reg.getMetric()); + info.setRegistrationHandle(reg.getHandle()); + info.setRemote(reg.isRemote()); + info.setRequestHandlerSource(reg.getHandlerSource()); + info.setServiceType(reg.getServiceType()); + serviceListener.serviceRegistered(handle, info); + } catch (Throwable t) { + logListenerError(t); + } + } + } + return handle; + } + } + + private static final class ServiceListenerRegistration { + private final ServiceListener serviceListener; + private volatile SimpleCloseable handle; + + private ServiceListenerRegistration(final ServiceListener serviceListener) { + this.serviceListener = serviceListener; + } + + ServiceListener getServiceListener() { + return serviceListener; + } + + void setHandle(final SimpleCloseable handle) { + this.handle = handle; + } + } + + public String toString() { + return "endpoint \"" + name + "\" <" + Integer.toString(hashCode()) + ">"; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureClientSource.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/FutureClientSource.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureClientSource.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureClientSource.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,55 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import org.jboss.remoting.ClientSource; +import org.jboss.remoting.SimpleCloseable; +import org.jboss.xnio.AbstractIoFuture; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.IoUtils; + +/** + * + */ +final class FutureClientSource<I, O> extends AbstractIoFuture<ClientSource<I, O>> { + + private volatile SimpleCloseable listenerHandle; + + protected boolean setException(final IOException exception) { + return super.setException(exception); + } + + protected boolean setResult(final ClientSource<I, O> result) { + return super.setResult(result); + } + + public IoFuture<ClientSource<I, O>> cancel() { + IoUtils.safeClose(listenerHandle); + return this; + } + + void setListenerHandle(final SimpleCloseable listenerHandle) { + this.listenerHandle = listenerHandle; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureReplyImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/FutureReplyImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureReplyImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/FutureReplyImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,87 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.util.concurrent.Executor; +import org.jboss.remoting.ReplyException; +import org.jboss.remoting.spi.RemoteRequestContext; +import org.jboss.remoting.spi.ReplyHandler; +import org.jboss.xnio.AbstractIoFuture; +import org.jboss.xnio.IoFuture; + +/** + * + */ +final class FutureReplyImpl<O> extends AbstractIoFuture<O> { + + private final Executor executor; + private final Class<O> replyType; + private final ReplyHandler replyHandler = new Handler(); + private volatile RemoteRequestContext remoteRequestContext; + + FutureReplyImpl(final Executor executor, final Class<O> replyType) { + this.executor = executor; + this.replyType = replyType; + } + + void setRemoteRequestContext(final RemoteRequestContext remoteRequestContext) { + this.remoteRequestContext = remoteRequestContext; + } + + public IoFuture<O> cancel() { + // must not be called before setRemoteRequestContext + remoteRequestContext.cancel(); + return this; + } + + protected Executor getNotifierExecutor() { + return executor; + } + + ReplyHandler getReplyHandler() { + return replyHandler; + } + + private final class Handler implements ReplyHandler { + + public void handleReply(final Object reply) { + final O actualReply; + try { + actualReply = replyType.cast(reply); + } catch (ClassCastException e) { + setException(new ReplyException("Reply was of the wrong type (got <" + reply.getClass().getName() + ">; expected <? extends " + replyType.getName() + ">")); + return; + } + setResult(actualReply); + } + + public void handleException(final IOException exception) { + setException(exception); + } + + public void handleCancellation() { + finishCancel(); + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandler.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandler.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandler.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandler.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,157 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import org.jboss.remoting.RemoteExecutionException; +import org.jboss.remoting.RemoteRequestException; +import org.jboss.remoting.RequestListener; +import org.jboss.remoting.spi.AbstractAutoCloseable; +import org.jboss.remoting.spi.RemoteRequestContext; +import org.jboss.remoting.spi.ReplyHandler; +import org.jboss.remoting.spi.RequestHandler; +import org.jboss.remoting.spi.SpiUtils; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class LocalRequestHandler<I, O> extends AbstractAutoCloseable<RequestHandler> implements RequestHandler { + + private final RequestListener<I, O> requestListener; + private final ClientContextImpl clientContext; + private final Class<I> requestClass; + private final Class<O> replyClass; + + private static final Logger log = Logger.getLogger("org.jboss.remoting.listener"); + + LocalRequestHandler(Config<I, O> config) { + super(config.getExecutor()); + requestListener = config.getRequestListener(); + clientContext = config.getClientContext(); + requestClass = config.getRequestClass(); + replyClass = config.getReplyClass(); + } + + public RemoteRequestContext receiveRequest(final Object request, final ReplyHandler replyHandler) { + final RequestContextImpl<O> context = new RequestContextImpl<O>(replyHandler, clientContext, replyClass); + try { + final I castRequest; + try { + castRequest = requestClass.cast(request); + } catch (ClassCastException e) { + SpiUtils.safeHandleException(replyHandler, new RemoteRequestException("Request is the wrong type; expected " + requestClass + " but got " + request.getClass())); + return SpiUtils.getBlankRemoteRequestContext(); + } + context.execute(new Runnable() { + public void run() { + try { + requestListener.handleRequest(context, castRequest); + } catch (RemoteExecutionException e) { + SpiUtils.safeHandleException(replyHandler, e); + } catch (Throwable t) { + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Request handler threw an exception", t)); + } + } + }); + } catch (RejectedExecutionException e) { + SpiUtils.safeHandleException(replyHandler, new RemoteRequestException("Execution was rejected (server may be too busy)", e)); + return SpiUtils.getBlankRemoteRequestContext(); + } + return new RemoteRequestContext() { + public void cancel() { + context.cancel(); + } + }; + } + + protected void closeAction() throws IOException { + try { + requestListener.handleClientClose(clientContext); + } catch (Throwable t) { + log.error(t, "Unexpected exception in request listener client close handler method"); + } + } + + void open() throws IOException { + try { + requestListener.handleClientOpen(clientContext); + } catch (Throwable t) { + final IOException ioe = new IOException("Failed to open client context"); + ioe.initCause(t); + throw ioe; + } + } + + public String toString() { + return "local request handler <" + Integer.toHexString(hashCode()) + "> (request listener = " + String.valueOf(requestListener) + ")"; + } + + static class Config<I, O> { + private final Class<I> requestClass; + private final Class<O> replyClass; + + private Executor executor; + private RequestListener<I, O> requestListener; + private ClientContextImpl clientContext; + + Config(final Class<I> requestClass, final Class<O> replyClass) { + this.requestClass = requestClass; + this.replyClass = replyClass; + } + + public Class<I> getRequestClass() { + return requestClass; + } + + public Class<O> getReplyClass() { + return replyClass; + } + + public Executor getExecutor() { + return executor; + } + + public void setExecutor(final Executor executor) { + this.executor = executor; + } + + public RequestListener<I, O> getRequestListener() { + return requestListener; + } + + public void setRequestListener(final RequestListener<I, O> requestListener) { + this.requestListener = requestListener; + } + + public ClientContextImpl getClientContext() { + return clientContext; + } + + public void setClientContext(final ClientContextImpl clientContext) { + this.clientContext = clientContext; + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandlerSource.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandlerSource.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandlerSource.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalRequestHandlerSource.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,133 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.util.concurrent.Executor; +import org.jboss.remoting.CloseHandler; +import org.jboss.remoting.RequestListener; +import org.jboss.remoting.spi.AbstractAutoCloseable; +import org.jboss.remoting.spi.Handle; +import org.jboss.remoting.spi.RequestHandler; +import org.jboss.remoting.spi.RequestHandlerSource; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class LocalRequestHandlerSource<I, O> extends AbstractAutoCloseable<RequestHandlerSource> implements RequestHandlerSource { + + private final RequestListener<I, O> requestListener; + private final ServiceContextImpl serviceContext; + private final Executor executor; + private final Class<I> requestClass; + private final Class<O> replyClass; + + private static final Logger log = Logger.getLogger("org.jboss.remoting.listener-source"); + + LocalRequestHandlerSource(final Config<I, O> config) { + super(config.getExecutor()); + requestClass = config.getRequestClass(); + replyClass = config.getReplyClass(); + requestListener = config.getRequestListener(); + executor = config.getExecutor(); + serviceContext = new ServiceContextImpl(executor); + } + + public Handle<RequestHandler> createRequestHandler() throws IOException { + if (isOpen()) { + final LocalRequestHandler.Config<I, O> config = new LocalRequestHandler.Config<I, O>(requestClass, replyClass); + config.setExecutor(executor); + config.setRequestListener(requestListener); + config.setClientContext(new ClientContextImpl(serviceContext)); + final LocalRequestHandler<I, O> localRequestHandler = new LocalRequestHandler<I, O>(config); + localRequestHandler.open(); + return localRequestHandler.getHandle(); + } else { + throw new IOException("LocalRequestHandlerSource is closed"); + } + } + + void open() throws IOException { + try { + requestListener.handleServiceOpen(serviceContext); + addCloseHandler(new CloseHandler<RequestHandlerSource>() { + public void handleClose(final RequestHandlerSource closed) { + try { + requestListener.handleServiceClose(serviceContext); + } catch (Throwable t) { + log.error(t, "Unexpected exception in request listener client close handler method"); + } + } + }); + } catch (Throwable t) { + final IOException ioe = new IOException("Failed to open client context"); + ioe.initCause(t); + throw ioe; + } + } + + ServiceContextImpl getServiceContext() { + return serviceContext; + } + + public String toString() { + return "local request handler source <" + Integer.toHexString(hashCode()) + "> (request listener = " + String.valueOf(requestListener) + ")"; + } + + static class Config<I, O> { + private final Class<I> requestClass; + private final Class<O> replyClass; + private Executor executor; + private RequestListener<I, O> requestListener; + + Config(final Class<I> requestClass, final Class<O> replyClass) { + this.requestClass = requestClass; + this.replyClass = replyClass; + } + + public Class<I> getRequestClass() { + return requestClass; + } + + public Class<O> getReplyClass() { + return replyClass; + } + + public Executor getExecutor() { + return executor; + } + + public void setExecutor(final Executor executor) { + this.executor = executor; + } + + public RequestListener<I, O> getRequestListener() { + return requestListener; + } + + public void setRequestListener(final RequestListener<I, O> requestListener) { + this.requestListener = requestListener; + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/OrderedExecutor.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/OrderedExecutor.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/OrderedExecutor.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/OrderedExecutor.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,95 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.util.LinkedList; +import java.util.concurrent.Executor; +import org.jboss.xnio.log.Logger; + +/** + * An executor that always runs all tasks in order, using a delegate executor to run the tasks. + * <p/> + * More specifically, any call B to the {@link #execute(Runnable)} method that happens-after another call A to the + * same method, will result in B's task running after A's. + */ +final class OrderedExecutor implements Executor { + private static final Logger log = Logger.getLogger(OrderedExecutor.class); + + // @protectedby tasks + private final LinkedList<Runnable> tasks = new LinkedList<Runnable>(); + // @protectedby tasks + private boolean running; + private final Executor parent; + private final Runnable runner; + + /** + * Construct a new instance. + * + * @param parent the parent executor + */ + OrderedExecutor(final Executor parent) { + this.parent = parent; + runner = new Runnable() { + public void run() { + for (;;) { + final Runnable task; + synchronized(tasks) { + task = tasks.poll(); + if (task == null) { + running = false; + return; + } + } + try { + task.run(); + } catch (Throwable t) { + log.error(t, "Runnable task %s failed", task); + } + } + } + }; + } + + /** + * Run a task. + * + * @param command the task to run. + */ + public void execute(Runnable command) { + synchronized(tasks) { + tasks.add(command); + if (! running) { + running = true; + boolean ok = false; + try { + parent.execute(runner); + ok = true; + } finally { + if (! ok) { + running = false; + } + } + } + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/QueueExecutor.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/QueueExecutor.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/QueueExecutor.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/QueueExecutor.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,101 @@ +package org.jboss.remoting; + +import java.util.LinkedList; +import java.util.Queue; +import org.jboss.xnio.CloseableExecutor; +import org.jboss.xnio.log.Logger; + +/** + * An executor designed to run all submitted tasks in the current thread. The queue is run continuously + * until the {@code close()} method is invoked. Jobs may be submitted to the queue from any thread. + * Only one thread should invoke the {@code runQueue()} method, which will run until the executor is + * shut down. + */ +final class QueueExecutor implements CloseableExecutor { + private static final Logger log = org.jboss.xnio.log.Logger.getLogger(QueueExecutor.class); + + private final Queue<Runnable> queue = new LinkedList<Runnable>(); + + private State state = State.WAITING; + + private enum State { + RUNNING, + WAITING, + STOPPING, + DOWN, + } + + public void execute(Runnable command) { + synchronized(queue) { + switch (state) { + case WAITING: + state = State.RUNNING; + queue.notify(); + // fall thru + case STOPPING: + case RUNNING: + queue.add(command); + break; + default: + throw new IllegalStateException("Executor is no longer available"); + } + } + } + + public void runQueue() { + boolean intr = Thread.interrupted(); + try { + for (;;) { + final State newState; + synchronized(queue) { + while (state == State.WAITING) { + try { + queue.wait(); + } catch (InterruptedException e) { + intr = true; + } + } + if (state == State.DOWN) { + throw new IllegalStateException("DOWN"); + } + newState = state; + } + for (;;) { + final Runnable runnable; + synchronized(queue) { + runnable = queue.poll(); + if (runnable == null) { + break; + } + } + try { + runnable.run(); + } catch (Throwable t) { + log.trace(t, "Error occurred while processing run queue"); + } + } + if (newState == State.STOPPING) { + synchronized(queue) { + state = State.DOWN; + return; + } + } + } + } finally { + if (intr) { + Thread.currentThread().interrupt(); + } + } + } + + public void close() { + synchronized(queue) { + switch (state) { + case WAITING: + queue.notify(); + case RUNNING: + state = State.STOPPING; + } + } + } +} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/Remoting.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/Remoting.java 2009-02-26 17:58:02 UTC (rev 4854) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/Remoting.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -87,11 +87,7 @@ * @return the endpoint */ public static Endpoint createEndpoint(final Executor executor, final String name) throws IOException { - try { - return (Endpoint) Class.forName("org.jboss.remoting.core.EndpointImpl").getConstructor(Executor.class, String.class).newInstance(executor, name); - } catch (Exception e) { - throw new EndpointException("Unable to create endpoint", e); - } + return new EndpointImpl(executor, name); } /** Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/RequestContextImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,164 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.jboss.remoting.ClientContext; +import org.jboss.remoting.IndeterminateOutcomeException; +import org.jboss.remoting.RemoteExecutionException; +import org.jboss.remoting.RemoteReplyException; +import org.jboss.remoting.RequestCancelHandler; +import org.jboss.remoting.RequestContext; +import org.jboss.remoting.spi.ReplyHandler; +import org.jboss.remoting.spi.SpiUtils; + +/** + * + */ +final class RequestContextImpl<O> implements RequestContext<O> { + + private final AtomicBoolean closed = new AtomicBoolean(); + private final Object cancelLock = new Object(); + private final ReplyHandler replyHandler; + private final ClientContextImpl clientContext; + private final AtomicInteger taskCount = new AtomicInteger(); + + // @protectedby cancelLock + private boolean cancelled; + // @protectedby cancelLock + private Set<RequestCancelHandler<O>> cancelHandlers; + private final RequestListenerExecutor interruptingExecutor; + private final Class<O> replyClass; + + RequestContextImpl(final ReplyHandler replyHandler, final ClientContextImpl clientContext, final Class<O> replyClass) { + this.replyHandler = replyHandler; + this.clientContext = clientContext; + this.replyClass = replyClass; + final Executor executor = clientContext.getExecutor(); + //noinspection ThisEscapedInObjectConstruction + interruptingExecutor = new RequestListenerExecutor(executor, this); + } + + public ClientContext getContext() { + return clientContext; + } + + public boolean isCancelled() { + synchronized (cancelLock) { + return cancelled; + } + } + + public void sendReply(final O reply) throws IOException, IllegalStateException { + if (! closed.getAndSet(true)) { + final O actualReply; + try { + actualReply = replyClass.cast(reply); + } catch (ClassCastException e) { + SpiUtils.safeHandleException(replyHandler, new RemoteReplyException("Remote reply was the wrong type", e)); + throw e; + } + try { + replyHandler.handleReply(actualReply); + } catch (IOException e) { + SpiUtils.safeHandleException(replyHandler, new RemoteReplyException("Remote reply failed", e)); + throw e; + } + } else { + throw new IllegalStateException("Reply already sent"); + } + } + + public void sendFailure(final String msg, final Throwable cause) throws IOException, IllegalStateException { + if (! closed.getAndSet(true)) { + replyHandler.handleException(new RemoteExecutionException(msg, cause)); + } else { + throw new IllegalStateException("Reply already sent"); + } + } + + public void sendCancelled() throws IOException, IllegalStateException { + if (! closed.getAndSet(true)) { + try { + replyHandler.handleCancellation(); + } catch (IOException e) { + // this is highly unlikely to succeed + SpiUtils.safeHandleException(replyHandler, new RemoteReplyException("Remote cancellation acknowledgement failed", e)); + } + } else { + throw new IllegalStateException("Reply already sent"); + } + } + + public void addCancelHandler(final RequestCancelHandler<O> handler) { + synchronized (cancelLock) { + if (cancelled) { + SpiUtils.safeNotifyCancellation(handler, this); + } else { + if (cancelHandlers == null) { + cancelHandlers = new HashSet<RequestCancelHandler<O>>(); + } + cancelHandlers.add(handler); + } + } + } + + public void execute(final Runnable command) { + interruptingExecutor.execute(command); + } + + protected void cancel() { + synchronized (cancelLock) { + if (! cancelled) { + cancelled = true; + if (cancelHandlers != null) { + for (final RequestCancelHandler<O> handler : cancelHandlers) { + interruptingExecutor.execute(new Runnable() { + public void run() { + SpiUtils.safeNotifyCancellation(handler, RequestContextImpl.this); + } + }); + } + cancelHandlers = null; + } + interruptingExecutor.interruptAll(); + } + } + } + + void startTask() { + taskCount.incrementAndGet(); + } + + void finishTask() { + if (taskCount.decrementAndGet() == 0 && ! closed.getAndSet(true)) { + // no response sent! send back IndeterminateOutcomeException + SpiUtils.safeHandleException(replyHandler, new IndeterminateOutcomeException("No reply was sent by the request listener")); + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestListenerExecutor.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/RequestListenerExecutor.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestListenerExecutor.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestListenerExecutor.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,81 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.Executor; + +/** + * + */ +final class RequestListenerExecutor implements Executor { + + private final Set<Task> tasks = Collections.synchronizedSet(new HashSet<Task>()); + private final Executor executor; + private final RequestContextImpl<?> requestContext; + + public RequestListenerExecutor(final Executor executor, final RequestContextImpl<?> context) { + this.executor = executor; + requestContext = context; + } + + private final class Task implements Runnable { + private volatile Thread thread; + private final Runnable runnable; + + private Task(final Runnable runnable) { + this.runnable = runnable; + } + + public void run() { + requestContext.startTask(); + thread = Thread.currentThread(); + tasks.add(this); + try { + runnable.run(); + } finally { + requestContext.finishTask(); + tasks.remove(this); + thread = null; + } + } + } + + public void execute(final Runnable command) { + executor.execute(new Task(command)); + } + + public void interruptAll() { + synchronized (tasks) { + final Thread currentThread = Thread.currentThread(); + for (Task task : tasks) { + final Thread thread = task.thread; + if (thread != null && thread != currentThread) { + thread.interrupt(); + } + } + } + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceContextImpl.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ServiceContextImpl.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceContextImpl.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceContextImpl.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,42 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import java.util.concurrent.Executor; +import org.jboss.remoting.ServiceContext; +import org.jboss.xnio.log.Logger; + +/** + * + */ +final class ServiceContextImpl extends AbstractContextImpl<ServiceContext> implements ServiceContext { + private static final Logger log = Logger.getLogger("org.jboss.remoting.service-context"); + + protected ServiceContextImpl(final Executor executor) { + super(executor); + } + + public String toString() { + return "service context instance <" + Integer.toHexString(hashCode()) + ">"; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceRegistration.java (from rev 4854, remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ServiceRegistration.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceRegistration.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ServiceRegistration.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,95 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import org.jboss.remoting.SimpleCloseable; +import org.jboss.remoting.spi.RequestHandlerSource; + +/** + * + */ +final class ServiceRegistration { + private final boolean remote; + private final String serviceType; + private final String groupName; + private final String endpointName; + private final int metric; + private final RequestHandlerSource handlerSource; + private volatile SimpleCloseable handle; + + ServiceRegistration(final String serviceType, final String groupName, final String endpointName, final int metric, final RequestHandlerSource handlerSource) { + remote = true; + this.serviceType = serviceType; + this.groupName = groupName; + this.endpointName = endpointName; + this.metric = metric; + this.handlerSource = handlerSource; + } + + ServiceRegistration(final String serviceType, final String groupName, final String endpointName, final RequestHandlerSource handlerSource) { + remote = false; + metric = 0; + this.serviceType = serviceType; + this.groupName = groupName; + this.endpointName = endpointName; + this.handlerSource = handlerSource; + } + + public boolean matches(final String serviceType, final String groupName, final String endpointName) { + return (serviceType == null || serviceType.length() == 0 || serviceType.equals(this.serviceType)) && + (groupName == null || groupName.length() == 0 || groupName.equals(this.groupName)) && + (endpointName == null || endpointName.length() == 0 || endpointName.equals(this.endpointName)); + } + + public boolean isRemote() { + return remote; + } + + public String getServiceType() { + return serviceType; + } + + public String getGroupName() { + return groupName; + } + + public String getEndpointName() { + return endpointName; + } + + public int getMetric() { + return metric; + } + + public RequestHandlerSource getHandlerSource() { + return handlerSource; + } + + public SimpleCloseable getHandle() { + return handle; + } + + void setHandle(final SimpleCloseable handle) { + this.handle = handle; + } +} Copied: remoting3/trunk/api/src/test/java/org/jboss/remoting/EndpointTestCase.java (from rev 4853, remoting3/trunk/core/src/test/java/org/jboss/remoting/core/EndpointTestCase.java) =================================================================== --- remoting3/trunk/api/src/test/java/org/jboss/remoting/EndpointTestCase.java (rev 0) +++ remoting3/trunk/api/src/test/java/org/jboss/remoting/EndpointTestCase.java 2009-02-26 18:30:20 UTC (rev 4855) @@ -0,0 +1,277 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting; + +import junit.framework.TestCase; +import java.util.concurrent.Executors; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.io.IOException; +import org.jboss.remoting.AbstractRequestListener; +import org.jboss.remoting.RequestContext; +import org.jboss.remoting.RemoteExecutionException; +import org.jboss.remoting.CloseHandler; +import org.jboss.remoting.Client; +import org.jboss.remoting.IndeterminateOutcomeException; +import org.jboss.remoting.test.support.LoggingHelper; +import org.jboss.remoting.spi.RequestHandler; +import org.jboss.remoting.spi.Handle; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.log.Logger; + +/** + * + */ +public final class EndpointTestCase extends TestCase { + static { + LoggingHelper.init(); + } + + private static final Logger log = Logger.getLogger(EndpointTestCase.class); + + public void testCreate() throws Throwable { + final ExecutorService executorService = Executors.newCachedThreadPool(); + final EndpointImpl endpoint = new EndpointImpl(executorService, "foo"); + try { + endpoint.close(); + executorService.shutdown(); + assertTrue(executorService.awaitTermination(1L, TimeUnit.SECONDS)); + } finally { + executorService.shutdownNow(); + } + } + + public void testLocalClientInvoke() throws Throwable { + final AtomicBoolean clientEndpointClosed = new AtomicBoolean(false); + final AtomicBoolean clientClosed = new AtomicBoolean(false); + final ExecutorService executorService = Executors.newCachedThreadPool(); + try { + final EndpointImpl endpoint = new EndpointImpl(executorService, "test-endpoint"); + final Object requestObj = new Object(); + final Object replyObj = new Object(); + try { + final Handle<RequestHandler> handle = endpoint.createRequestHandler(new AbstractRequestListener<Object, Object>() { + public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException { + assertEquals(request, requestObj); + try { + context.sendReply(replyObj); + } catch (IOException e) { + log.error(e, "Error sending reply!"); + } + } + }, Object.class, Object.class); + try { + final RequestHandler requestHandler = handle.getResource(); + try { + requestHandler.addCloseHandler(new CloseHandler<RequestHandler>() { + public void handleClose(final RequestHandler closed) { + clientEndpointClosed.set(true); + } + }); + final Client<Object,Object> client = endpoint.createClient(requestHandler, Object.class, Object.class); + try { + client.addCloseHandler(new CloseHandler<Client<Object, Object>>() { + public void handleClose(final Client<Object, Object> closed) { + clientClosed.set(true); + } + }); + handle.close(); + assertEquals(replyObj, client.invoke(requestObj)); + client.close(); + executorService.shutdown(); + assertTrue(executorService.awaitTermination(1L, TimeUnit.SECONDS)); + assertTrue(clientEndpointClosed.get()); + assertTrue(clientClosed.get()); + } finally { + IoUtils.safeClose(client); + } + } finally { + IoUtils.safeClose(requestHandler); + } + } finally { + IoUtils.safeClose(handle); + } + } finally { + IoUtils.safeClose(endpoint); + } + } finally { + executorService.shutdownNow(); + } + } + + public void testLocalClientSend() throws Throwable { + final AtomicBoolean clientEndpointClosed = new AtomicBoolean(false); + final AtomicBoolean clientClosed = new AtomicBoolean(false); + final ExecutorService executorService = Executors.newCachedThreadPool(); + try { + final EndpointImpl endpoint = new EndpointImpl(executorService, "test-endpoint"); + try { + final Object requestObj = new Object(); + final Object replyObj = new Object(); + final Handle<RequestHandler> handle = endpoint.createRequestHandler(new AbstractRequestListener<Object, Object>() { + public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException { + assertEquals(request, requestObj); + try { + context.sendReply(replyObj); + } catch (IOException e) { + log.error(e, "Error sending reply!"); + } + } + }, Object.class, Object.class); + try { + final RequestHandler requestHandler = handle.getResource(); + try { + requestHandler.addCloseHandler(new CloseHandler<RequestHandler>() { + public void handleClose(final RequestHandler closed) { + clientEndpointClosed.set(true); + } + }); + final Client<Object,Object> client = endpoint.createClient(requestHandler, Object.class, Object.class); + try { + client.addCloseHandler(new CloseHandler<Client<Object, Object>>() { + public void handleClose(final Client<Object, Object> closed) { + clientClosed.set(true); + } + }); + handle.close(); + final IoFuture<? extends Object> futureReply = client.send(requestObj); + assertEquals(IoFuture.Status.DONE, futureReply.await(1L, TimeUnit.SECONDS)); + assertEquals(replyObj, futureReply.get()); + client.close(); + executorService.shutdown(); + assertTrue(executorService.awaitTermination(1L, TimeUnit.SECONDS)); + assertTrue(clientEndpointClosed.get()); + assertTrue(clientClosed.get()); + } finally { + IoUtils.safeClose(client); + } + } finally { + IoUtils.safeClose(requestHandler); + } + } finally { + IoUtils.safeClose(handle); + } + } finally { + IoUtils.safeClose(endpoint); + } + } finally { + executorService.shutdownNow(); + } + } + + public void testUnsentReply() throws Throwable { + final ExecutorService executorService = Executors.newCachedThreadPool(); + try { + final EndpointImpl endpoint = new EndpointImpl(executorService, "test-endpoint"); + try { + final Object requestObj = new Object(); + final Handle<RequestHandler> handle = endpoint.createRequestHandler(new AbstractRequestListener<Object, Object>() { + public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException { + assertEquals(request, requestObj); + // don't send a reply!! + } + }, Object.class, Object.class); + try { + final RequestHandler requestHandler = handle.getResource(); + try { + final Client<Object,Object> client = endpoint.createClient(requestHandler, Object.class, Object.class); + try { + final IoFuture<? extends Object> futureReply = client.send(requestObj); + assertEquals(IoFuture.Status.FAILED, futureReply.await(500L, TimeUnit.MILLISECONDS)); + assertTrue(futureReply.getException() instanceof IndeterminateOutcomeException); + } finally { + IoUtils.safeClose(client); + } + } finally { + IoUtils.safeClose(requestHandler); + } + } finally { + IoUtils.safeClose(handle); + } + } finally { + IoUtils.safeClose(endpoint); + } + } finally { + executorService.shutdownNow(); + } + } + + public void testUnsentReply2() throws Throwable { + final ExecutorService executorService = Executors.newCachedThreadPool(); + try { + final EndpointImpl endpoint = new EndpointImpl(executorService, "test-endpoint"); + try { + final Object requestObj = new Object(); + final Handle<RequestHandler> handle = endpoint.createRequestHandler(new AbstractRequestListener<Object, Object>() { + public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException { + assertEquals(request, requestObj); + context.execute(new Runnable() { + public void run() { + context.execute(new Runnable() { + public void run() { + context.execute(new Runnable() { + public void run() { + } + }); + } + }); + context.execute(new Runnable() { + public void run() { + } + }); + } + }); + context.execute(new Runnable() { + public void run() { + } + }); + // don't send a reply!! + } + }, Object.class, Object.class); + try { + final RequestHandler requestHandler = handle.getResource(); + try { + final Client<Object,Object> client = endpoint.createClient(requestHandler, Object.class, Object.class); + try { + final IoFuture<? extends Object> futureReply = client.send(requestObj); + assertEquals(IoFuture.Status.FAILED, futureReply.await(500L, TimeUnit.MILLISECONDS)); + assertTrue(futureReply.getException() instanceof IndeterminateOutcomeException); + } finally { + IoUtils.safeClose(client); + } + } finally { + IoUtils.safeClose(requestHandler); + } + } finally { + IoUtils.safeClose(handle); + } + } finally { + IoUtils.safeClose(endpoint); + } + } finally { + executorService.shutdownNow(); + } + } +} Modified: remoting3/trunk/build.xml =================================================================== --- remoting3/trunk/build.xml 2009-02-26 17:58:02 UTC (rev 4854) +++ remoting3/trunk/build.xml 2009-02-26 18:30:20 UTC (rev 4855) @@ -216,117 +216,6 @@ <property name="version" value="UNKNOWN"/> </target> - <!-- core module --> - - <target name="core.compile.depcheck"> - <mkdir dir="core/target/main"/> - <uptodate property="core.compile.uptodate" targetfile="core/target/main/.lastcompile"> - <srcfiles dir="core/src/main/java"> - <include name="**/"/> - <include name="**/*.java"/> - <exclude name="**/.*"/> - </srcfiles> - </uptodate> - </target> - - <target name="core.compile" depends="core.compile.depcheck" unless="core.compile.uptodate"> - <mkdir dir="core/target/main/classes"/> - <javac - source="${javac.source}" - target="${javac.target}" - srcdir="core/src/main/java" - destdir="core/target/main/classes" - debug="true"> - <compilerarg value="-Xlint:unchecked"/> - <classpath> - <path refid="api.classpath"/> - </classpath> - </javac> - <touch file="core/target/main/.lastcompile" verbose="false"/> - </target> - - <target name="core.test.compile.depcheck"> - <mkdir dir="core/target/test"/> - <uptodate property="core.compile.uptodate" targetfile="core/target/test/.lastcompile"> - <srcfiles dir="core/src/test/java"> - <include name="**/"/> - <include name="**/*.java"/> - <exclude name="**/.*"/> - </srcfiles> - </uptodate> - </target> - - <target name="core.test.compile" depends="lib.junit,core.compile,core.test.compile.depcheck" unless="core.test.compile.uptodate"> - <mkdir dir="core/target/test/classes"/> - <javac - source="${javac.source}" - target="${javac.target}" - srcdir="core/src/test/java" - destdir="core/target/test/classes" - debug="true"> - <compilerarg value="-Xlint:unchecked"/> - <classpath> - <path refid="api.classpath"/> - <path refid="core.classpath"/> - <path refid="testing-support.classpath"/> - <pathelement location="${lib.junit.local}"/> - </classpath> - </javac> - <touch file="core/target/test/.lastcompile" verbose="false"/> - </target> - - <target name="core.test.pseudotarget"> - <echo message="============================================="/> - <echo message="${message}"/> - <echo message="============================================="/> - <mkdir dir="core/target/test-results"/> - <junit printsummary="true" fork="yes" includeantruntime="true"> - <sysproperty key="build.home" value="${basedir}"/> - <sysproperty key="ant.library.dir" value="${ant.home}/lib"/> - <sysproperty key="lib.junit.local" value="${lib.junit.local}"/> - <sysproperty key="lib.marshalling-api.local" value="${lib.marshalling-api.local}"/> - <sysproperty key="lib.xnio-api.local" value="${lib.xnio-api.local}"/> - <jvmarg line="${test.jvmargs}"/> - <formatter type="plain" extension="${extension}"/> - <classpath> - <path refid="api.classpath"/> - <path refid="core.classpath"/> - <path refid="testing-support.classpath"/> - <pathelement location="core/target/test/classes"/> - <path refid="lib.junit.classpath"/> - </classpath> - <batchtest fork="yes" todir="core/target/test-results" - haltonfailure="no"> - <fileset dir="core/target/test/classes"> - <include name="**/*TestCase.class"/> - </fileset> - </batchtest> - </junit> - </target> - - <target name="core.test" depends="api,core,testing-support,core.test.compile"> - <antcall inheritall="true" inheritrefs="true" target="core.test.pseudotarget"> - <param name="extension" value=".txt"/> - <param name="message" value="Running with no security manager"/> - <param name="test.jvmargs" value="-Ddummy=dummy"/> - </antcall> - <antcall inheritall="true" inheritrefs="true" target="core.test.pseudotarget"> - <param name="extension" value="-security.txt"/> - <param name="message" value="Running with security manager"/> - <param name="test.jvmargs" value="-Djava.security.manager=org.jboss.remoting.test.support.LoggingSecurityManager -Djava.security.policy=${basedir}/testing-support/src/main/resources/testing.policy"/> - </antcall> - </target> - - <target name="core.clean"> - <delete dir="core/target"/> - </target> - - <target name="core" description="Build the core module" depends="api,core.compile"> - <path id="core.classpath"> - <pathelement location="core/target/main/classes"/> - </path> - </target> - <!-- samples module --> <target name="samples.compile.depcheck"> @@ -413,8 +302,8 @@ <!-- ============================================== --> <target name="api-jar" description="Build the API JAR" depends="api"> - <delete file="jboss-remoting-api.jar"/> - <jar jarfile="jboss-remoting-api.jar"> + <delete file="jboss-remoting.jar"/> + <jar jarfile="jboss-remoting.jar"> <manifest> <attribute name="Created-By" value="${java.vm.version} (${java.vm.vendor})"/> <attribute name="Specification-Title" value="JBoss Remoting"/> @@ -433,8 +322,8 @@ </target> <target name="api-source-jar" description="Build the API source JAR" depends="api"> - <delete file="jboss-remoting-api-source.jar"/> - <jar jarfile="jboss-remoting-api-source.jar"> + <delete file="jboss-remoting-source.jar"/> + <jar jarfile="jboss-remoting-source.jar"> <manifest> <attribute name="Created-By" value="${java.vm.version} (${java.vm.vendor})"/> <attribute name="Specification-Title" value="JBoss Remoting Source"/> @@ -452,51 +341,11 @@ </jar> </target> - <target name="core-jar" description="Build the Core JAR" depends="core"> - <delete file="jboss-remoting-core.jar"/> - <jar jarfile="jboss-remoting-core.jar"> - <manifest> - <attribute name="Created-By" value="${java.vm.version} (${java.vm.vendor})"/> - <attribute name="Specification-Title" value="JBoss Remoting"/> - <attribute name="Specification-Version" value="${version}"/> - <attribute name="Specification-Vendor" value="JBoss (http://www.jboss.org/)"/> - <attribute name="Implementation-Title" value="JBoss Remoting (Core)"/> - <attribute name="Implementation-URL" value="http://labs.jboss.org/jbossremoting/"/> - <attribute name="Implementation-Version" value="${version}"/> - <attribute name="Implementation-Vendor" value="JBoss, a division of Red Hat, Inc."/> - <attribute name="Implementation-Vendor-Id" value="http://www.jboss.org"/> - </manifest> - <zipfileset dir="core/target/main/classes"> - <include name="**/*.class"/> - </zipfileset> - </jar> - </target> - - <target name="core-source-jar" description="Build the Core source JAR" depends="core"> - <delete file="jboss-remoting-core-source.jar"/> - <jar jarfile="jboss-remoting-core-source.jar"> - <manifest> - <attribute name="Created-By" value="${java.vm.version} (${java.vm.vendor})"/> - <attribute name="Specification-Title" value="JBoss Remoting Source"/> - <attribute name="Specification-Version" value="${version}"/> - <attribute name="Specification-Vendor" value="JBoss (http://www.jboss.org/)"/> - <attribute name="Implementation-Title" value="JBoss Remoting Source (Core)"/> - <attribute name="Implementation-URL" value="http://labs.jboss.org/jbossremoting/"/> - <attribute name="Implementation-Version" value="${version}"/> - <attribute name="Implementation-Vendor" value="JBoss, a division of Red Hat, Inc."/> - <attribute name="Implementation-Vendor-Id" value="http://www.jboss.org"/> - </manifest> - <zipfileset dir="core/src/main/java"> - <include name="**/*.java"/> - </zipfileset> - </jar> - </target> - <!-- ============================================== --> <!-- JAVADOCS --> <!-- ============================================== --> - <target name="api-javadoc" description="Build the API JavaDoc" depends="api,core,lib.apiviz,lib.marshalling-api,lib.xnio-api"> + <target name="api-javadoc" description="Build the API JavaDoc" depends="api,lib.apiviz,lib.marshalling-api,lib.xnio-api"> <delete dir="api/target/main/docs"/> <mkdir dir="api/target/main/docs"/> <javadoc destdir="api/target/main/docs" author="false" version="false" use="false" windowtitle="JBoss Remoting API"> @@ -510,7 +359,6 @@ <link href="http://docs.jboss.org/xnio/1.2.0.GA/api/"/> <link href="http://docs.jboss.org/river/1.0.0.GA/api/"/> <classpath> - <path refid="core.classpath"/> <path refid="api.classpath"/> </classpath> </javadoc> @@ -532,15 +380,13 @@ <!-- Distribution target --> <!-- ============================================== --> - <target name="dist" description="Build distribution zip file" depends="api-jar,api-source-jar,api-javadoc-zip,core-jar,core-source-jar"> + <target name="dist" description="Build distribution zip file" depends="api-jar,api-source-jar,api-javadoc-zip"> <delete file="jboss-remoting.zip"/> <zip zipfile="jboss-remoting.zip"> <zipfileset dir="${basedir}"> <include name="COPYING.txt"/> <include name="jboss-remoting-api.jar"/> <include name="jboss-remoting-api-source.jar"/> - <include name="jboss-remoting-core.jar"/> - <include name="jboss-remoting-core-source.jar"/> <include name="jboss-remoting-javadoc.zip"/> </zipfileset> </zip> @@ -552,18 +398,18 @@ <!-- core --> <!--<target name="all-core" description="Build all core targets" depends="api,core,samples,testing-support"/>--> - <target name="all-core" description="Build all core targets" depends="api,core,testing-support"/> + <target name="all-core" description="Build all core targets" depends="api,testing-support"/> <!-- JARs: These should be the second-to-last targets in the file --> - <target name="all-jars" description="Build all the JARs" depends="api-jar,core-jar"/> + <target name="all-jars" description="Build all the JARs" depends="api-jar"/> <!-- all: These should be the last targets in the file --> <target name="all" description="Build everything" depends="all-core,all-jars,api-javadoc"/> - <target name="clean" description="Clean out all build files" depends="api.clean,core.clean,samples.clean,testing-support.clean"/> + <target name="clean" description="Clean out all build files" depends="api.clean,samples.clean,testing-support.clean"/> - <target name="test" description="Run all tests" depends="api.test,core.test"/> + <target name="test" description="Run all tests" depends="api.test"/> </project> Modified: remoting3/trunk/testing-support/src/main/resources/testing.policy =================================================================== --- remoting3/trunk/testing-support/src/main/resources/testing.policy 2009-02-26 17:58:02 UTC (rev 4854) +++ remoting3/trunk/testing-support/src/main/resources/testing.policy 2009-02-26 18:30:20 UTC (rev 4855) @@ -4,34 +4,11 @@ grant codeBase "file:${build.home}/api/target/test/classes/-" { permission java.lang.RuntimePermission "modifyThread"; // for executor control -}; - -grant codeBase "file:${build.home}/core/target/test/classes/-" -{ - permission java.lang.RuntimePermission "modifyThread"; // for executor control permission org.jboss.remoting.EndpointPermission "createEndpoint"; permission org.jboss.remoting.EndpointPermission "createRequestHandler"; permission org.jboss.remoting.EndpointPermission "createClient"; }; -grant codeBase "file:${build.home}/protocol/basic/target/test/classes/-" -{ - permission java.lang.RuntimePermission "modifyThread"; // for executor control - permission org.jboss.remoting.EndpointPermission "createEndpoint"; - permission org.jboss.remoting.EndpointPermission "createRequestHandler"; - permission org.jboss.remoting.EndpointPermission "createClient"; -}; - -grant codeBase "file:${build.home}/protocol/multiplex/target/test/classes/-" -{ - permission java.lang.RuntimePermission "modifyThread"; // for executor control - permission org.jboss.remoting.EndpointPermission "createEndpoint"; - permission org.jboss.remoting.EndpointPermission "createRequestHandler"; - permission org.jboss.remoting.EndpointPermission "createClient"; - permission org.jboss.remoting.EndpointPermission "createClientSource"; - permission org.jboss.remoting.EndpointPermission "registerService"; -}; - // Permissions for Remoting itself grant codeBase "file:${build.home}/api/target/main/classes/-" @@ -40,17 +17,6 @@ permission org.jboss.remoting.EndpointPermission "*"; }; -grant codeBase "file:${build.home}/core/target/main/classes/-" -{ - permission java.util.PropertyPermission "jboss.remoting.*", "read"; - permission org.jboss.remoting.EndpointPermission "*"; -}; - -grant codeBase "file:${build.home}/protocol/basic/target/main/classes/-" -{ - permission java.util.PropertyPermission "jboss.remoting.*", "read"; -}; - // Support classes grant codeBase "file:${build.home}/testing-support/target/main/classes/-"