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

Author: david.lloyd(a)jboss.com Date: 2008-11-03 21:34:36 -0500 (Mon, 03 Nov 2008) New Revision: 4642 Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointPermission.java remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalServiceConfiguration.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteRequestException.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteServiceConfiguration.java remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AutoCloseable.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerBiMap.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerResourceBiMap.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerBiMap.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerResourceBiMap.java Removed: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceWrapper.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientWrapper.java remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointWrapper.java remoting3/trunk/api/src/main/java/org/jboss/remoting/IOExceptionCarrier.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextWrapper.java remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSinkWrapper.java remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSourceWrapper.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityIntMap.java remoting3/trunk/util/src/main/java/org/jboss/remoting/util/ConcurrentIntegerMap.java remoting3/trunk/util/src/main/java/org/jboss/remoting/util/EmulatedConcurrentIntegerHashMap.java Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/Endpoint.java remoting3/trunk/api/src/main/java/org/jboss/remoting/HandleableCloseable.java remoting3/trunk/api/src/main/java/org/jboss/remoting/IndeterminateOutcomeException.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteReplyException.java remoting3/trunk/api/src/main/java/org/jboss/remoting/ReplyException.java remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContext.java remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractAutoCloseable.java remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractHandleableCloseable.java remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/Handle.java remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandler.java remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandlerSource.java remoting3/trunk/api/src/test/java/org/jboss/remoting/spi/CloseableTestCase.java remoting3/trunk/build.properties remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientExternalizer.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientImpl.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceExternalizer.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceImpl.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/EndpointImpl.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/FutureReplyImpl.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandler.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandlerSource.java remoting3/trunk/core/src/main/java/org/jboss/remoting/core/RequestContextImpl.java remoting3/trunk/core/src/test/java/org/jboss/remoting/core/EndpointTestCase.java remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicConfiguration.java remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicProtocol.java remoting3/trunk/protocol/basic/src/test/java/org/jboss/remoting/protocol/basic/BasicTestCase.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MessageType.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConfiguration.java remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexHandler.java remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalBasicExampleMain.java remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalStreamExampleMain.java remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java remoting3/trunk/transporter/src/main/java/org/jboss/remoting/transporter/Transporter.java remoting3/trunk/util/src/main/java/org/jboss/remoting/util/CollectionUtil.java Log: More massive cleanup; more marshaller integration work; type safety strengthening; exception cleanup; various bug fixes... Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceWrapper.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientSourceWrapper.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,50 +0,0 @@ -package org.jboss.remoting; - -import org.jboss.remoting.CloseHandler; -import org.jboss.remoting.Client; -import org.jboss.remoting.ClientSource; -import java.io.IOException; - -/** - * A simple delegating wrapper for client sources. - * - * @param <I> the request type - * @param <O> the reply type - */ -public class ClientSourceWrapper<I, O> implements ClientSource<I, O> { - private final ClientSource<I, O> delegate; - - /** - * Construct a new instance. Calls will be sent to the given {@code delegate} by default. - * - * @param delegate the delegate client instance - */ - protected ClientSourceWrapper(ClientSource<I, O> delegate) { - this.delegate = delegate; - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void close() throws IOException { - delegate.close(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void addCloseHandler(final CloseHandler<? super ClientSource<I, O>> closeHandler) { - delegate.addCloseHandler(new CloseHandler<ClientSource<I, O>>() { - public void handleClose(final ClientSource<I, O> closed) { - closeHandler.handleClose(ClientSourceWrapper.this); - } - }); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public Client<I, O> createClient() throws IOException { - return delegate.createClient(); - } -} Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientWrapper.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ClientWrapper.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,65 +0,0 @@ -package org.jboss.remoting; - -import java.util.concurrent.ConcurrentMap; -import java.io.IOException; -import org.jboss.remoting.CloseHandler; -import org.jboss.remoting.Client; -import org.jboss.xnio.IoFuture; - -/** - * A simple delegating wrapper for clients. - * - * @param <I> the request type - * @param <O> the reply type - */ -public class ClientWrapper<I, O> implements Client<I, O> { - protected final Client<I, O> delegate; - - /** - * Construct a new instance. Calls will be sent to the given {@code delegate} by default. - * - * @param delegate the delegate client instance - */ - protected ClientWrapper(final Client<I, O> delegate) { - this.delegate = delegate; - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void close() throws IOException { - delegate.close(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void addCloseHandler(final CloseHandler<? super Client<I, O>> closeHandler) { - delegate.addCloseHandler(new CloseHandler<Client<I, O>>() { - public void handleClose(final Client<I, O> closed) { - closeHandler.handleClose(ClientWrapper.this); - } - }); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public O invoke(final I request) throws IOException { - return delegate.invoke(request); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public IoFuture<O> send(final I request) throws IOException { - return delegate.send(request); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public ConcurrentMap<Object, Object> getAttributes() { - return delegate.getAttributes(); - } -} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/Endpoint.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/Endpoint.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/Endpoint.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -33,54 +33,61 @@ * Create a request handler that can be used to receive incoming requests on this endpoint. The client may be passed to a * remote endpoint as part of a request or a reply, or it may be used locally. * - * You must have the TODO permission to invoke this method. + * You must have the {@link org.jboss.remoting.EndpointPermission createRequestHandler EndpointPermission} to invoke this method. * * @param <I> the request type * @param <O> the reply type * @param requestListener the request listener + * @param requestClass the class of requests sent to this request listener + * @param replyClass the class of replies received back from this request listener * @return a handle for the client * @throws IOException if an error occurs */ - <I, O> Handle<RequestHandler> createRequestHandler(RequestListener<I, O> requestListener) throws IOException; + <I, O> Handle<RequestHandler> createRequestHandler(RequestListener<I, O> requestListener, final Class<I> requestClass, final Class<O> replyClass) throws IOException; /** * Create a request handler source that can be used to acquire clients associated with a request listener on this endpoint. - * The request handler source may be passed to a remote endpoint as part of a request or a reply, or it may be used locally. + * The request handler source may be ignored, passed to a remote endpoint as part of a request or a reply, or used locally. * The objects that are produced by this method may be used to mass-produce {@code RequestHandler} instances. * - * You must have the TODO permission to invoke this method. + * You must have the {@link org.jboss.remoting.EndpointPermission registerService EndpointPermission} to invoke this method. * * @param <I> the request type * @param <O> the reply type - * @param requestListener the request listener - * @param serviceType the type of service to advertise - * @param groupName the name of the group of this type to be part of - * @return a handle for the client source + * @param configuration the configuration to use * @throws IOException if an error occurs */ - <I, O> Handle<RequestHandlerSource> createRequestHandlerSource(RequestListener<I, O> requestListener, String serviceType, String groupName) throws IOException; + <I, O> Handle<RequestHandlerSource> registerService(LocalServiceConfiguration<I, O> configuration) throws IOException; /** * Create a client that uses the given request handler to handle its requests. * + * You must have the {@link org.jboss.remoting.EndpointPermission createClient EndpointPermission} to invoke this method. + * * @param <I> the request type * @param <O> the reply type * @param handler the request handler + * @param requestClass the class of requests sent through this client + * @param replyClass the class of replies received back through this client * @return the client * @throws IOException if an error occurs */ - <I, O> Client<I, O> createClient(RequestHandler handler) throws IOException; + <I, O> Client<I, O> createClient(RequestHandler handler, Class<I> requestClass, Class<O> replyClass) throws IOException; /** * Create a client source that uses the given request handler source to generate clients. * + * You must have the {@link org.jboss.remoting.EndpointPermission createClientSource EndpointPermission} to invoke this method. + * * @param <I> the request type * @param <O> the reply type * @param handlerSource the request handler source + * @param requestClass the class of requests sent through this client source + * @param replyClass the class of replies received back through this client source * @return the client source * @throws IOException if an error occurs */ - <I, O> ClientSource<I, O> createClientSource(RequestHandlerSource handlerSource) throws IOException; + <I, O> ClientSource<I, O> createClientSource(RequestHandlerSource handlerSource, Class<I> requestClass, Class<O> replyClass) throws IOException; /** * Attempt to locate a service. The return value then be queried for the service's {@code ClientSource}. @@ -88,32 +95,33 @@ * @param <I> the request type * @param <O> the reply type * @param serviceUri the URI of the service + * @param requestClass the class of requests sent through the client source + * @param replyClass the class of replies received back through the client source * @return the future service * @throws IllegalArgumentException if the given URI is not a valid Remoting service URI */ - <I, O> IoFuture<ClientSource<I, O>> locateService(URI serviceUri) throws IllegalArgumentException; + <I, O> IoFuture<ClientSource<I, O>> locateService(URI serviceUri, Class<I> requestClass, Class<O> replyClass) throws IllegalArgumentException; /** * Register a remotely available service.<p> * The remote endpoint may not have the same name as this endpoint. The group name and service type must be * non-{@code null} and non-empty. The metric must be greater than zero. * - * @param serviceType the service type string - * @param groupName the group name - * @param endpointName the name of the remote endpoint - * @param handlerSource the remote handler source - * @param metric the preference metric, lower is more preferred - * @return a handle corresponding to the registration + * You must have the {@link org.jboss.remoting.EndpointPermission registerRemoteService EndpointPermission} to invoke this method. + * + * @param configuration the remote service configuration * @throws IllegalArgumentException if one of the given arguments was not valid * @throws IOException if an error occurs with the registration */ - SimpleCloseable registerRemoteService(String serviceType, String groupName, String endpointName, RequestHandlerSource handlerSource, int metric) throws IllegalArgumentException, IOException; + SimpleCloseable registerRemoteService(RemoteServiceConfiguration configuration) throws IllegalArgumentException, IOException; /** * Add a listener for observing when local and remote services are added. The caller may specify whether the listener * should be notified of the complete list of currently registered services (set {@code onlyNew} to {@code false}) * or only services registered after the time of calling this method (set {@code onlyNew} to {@code true}). * + * You must have the {@link org.jboss.remoting.EndpointPermission addServiceListener EndpointPermission} to invoke this method. + * * @param serviceListener the listener * @param onlyNew {@code true} if only new registrations should be sent to the listener * @return a handle which may be used to unregister the listener Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointPermission.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointPermission.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointPermission.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,61 @@ +/* + * 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.security.BasicPermission; + +/** + * This class is for permissions relating to Remoting endpoints. + */ +public class EndpointPermission extends BasicPermission { + + private static final long serialVersionUID = 4984517897378387571L; + + /** + * Creates a new {@code EndpointPermission} object with the specified name. + * The name is the symbolic name of the {@code EndpointPermission}. + * + * @param name the name of the {@code EndpointPermission} + * + * @throws NullPointerException if {@code name} is {@code null} + * @throws IllegalArgumentException if {@code name} is empty + */ + public EndpointPermission(String name) throws NullPointerException, IllegalArgumentException { + super(name); + } + + /** + * Creates a new {@code EndpointPermission} object with the specified name. + * The name is the symbolic name of the {@code EndpointPermission}, and the + * actions string is currently unused. + * + * @param name the name of the {@code EndpointPermission} + * @param actions ignored + * + * @throws NullPointerException if {@code name} is {@code null} + * @throws IllegalArgumentException if {@code name} is empty + */ + public EndpointPermission(String name, String actions) throws NullPointerException, IllegalArgumentException { + super(name, actions); + } +} Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointWrapper.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/EndpointWrapper.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,94 +0,0 @@ -package org.jboss.remoting; - -import java.util.concurrent.ConcurrentMap; -import java.io.IOException; -import java.net.URI; -import org.jboss.remoting.Endpoint; -import org.jboss.remoting.RequestListener; -import org.jboss.remoting.Client; -import org.jboss.remoting.ClientSource; -import org.jboss.remoting.SimpleCloseable; -import org.jboss.remoting.ServiceListener; -import org.jboss.remoting.spi.RequestHandler; -import org.jboss.remoting.spi.RequestHandlerSource; -import org.jboss.remoting.spi.Handle; -import org.jboss.xnio.IoFuture; - -/** - * A simple delegating wrapper for endpoints. - */ -public class EndpointWrapper implements Endpoint { - protected final Endpoint delegate; - - /** - * Construct a new instance. Calls will be sent to the given {@code delegate} by default. - * - * @param delegate the delegate client instance - */ - protected EndpointWrapper(final Endpoint delegate) { - this.delegate = delegate; - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public ConcurrentMap<Object, Object> getAttributes() { - return delegate.getAttributes(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public String getName() { - return delegate.getName(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public <I, O> Handle<RequestHandler> createRequestHandler(final RequestListener<I, O> requestListener) throws IOException { - return delegate.createRequestHandler(requestListener); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public <I, O> Handle<RequestHandlerSource> createRequestHandlerSource(final RequestListener<I, O> requestListener, final String serviceType, final String groupName) throws IOException { - return delegate.createRequestHandlerSource(requestListener, serviceType, groupName); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public <I, O> Client<I, O> createClient(final RequestHandler handler) throws IOException { - return delegate.createClient(handler); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public <I, O> ClientSource<I, O> createClientSource(final RequestHandlerSource handlerSource) throws IOException { - return delegate.createClientSource(handlerSource); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public <I, O> IoFuture<ClientSource<I, O>> locateService(final URI serviceUri) throws IllegalArgumentException { - return delegate.locateService(serviceUri); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public SimpleCloseable registerRemoteService(final String serviceType, final String groupName, final String endpointName, final RequestHandlerSource handlerSource, final int metric) throws IllegalArgumentException, IOException { - return delegate.registerRemoteService(serviceType, groupName, endpointName, handlerSource, metric); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public SimpleCloseable addServiceListener(final ServiceListener serviceListener, final boolean onlyNew) { - return delegate.addServiceListener(serviceListener, true); - } -} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/HandleableCloseable.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/HandleableCloseable.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/HandleableCloseable.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -22,6 +22,18 @@ * takes place. * * @param handler the close handler + * @return a key which may be used to later remove this handler */ - void addCloseHandler(CloseHandler<? super T> handler); + Key addCloseHandler(CloseHandler<? super T> handler); + + /** + * A key which may be used to remove this handler. + */ + interface Key { + + /** + * Remove the registered handler. Calling this method more than once has no additional effect. + */ + void remove(); + } } Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/IOExceptionCarrier.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/IOExceptionCarrier.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/IOExceptionCarrier.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,29 +0,0 @@ -package org.jboss.remoting; - -import java.io.IOException; - -/** - * A runtime exception that carries an {@link java.io.IOException} as a cause. - */ -public class IOExceptionCarrier extends RuntimeException { - - private static final long serialVersionUID = -1602940590696531671L; - - /** - * Construct a new carrier. - * - * @param cause the nested cause - */ - public IOExceptionCarrier(IOException cause) { - super(cause); - } - - /** - * Get the cause. - * - * @return the cause - */ - public IOException getCause() { - return (IOException) super.getCause(); - } -} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/IndeterminateOutcomeException.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/IndeterminateOutcomeException.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/IndeterminateOutcomeException.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -3,7 +3,7 @@ /** * An exception that is thrown when an operation terminates in such a way that the outcome cannot be known. */ -public class IndeterminateOutcomeException extends RemoteExecutionException { +public class IndeterminateOutcomeException extends RemotingException { private static final long serialVersionUID = 6304843915977033800L; Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalServiceConfiguration.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalServiceConfiguration.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/LocalServiceConfiguration.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,77 @@ +/* + * 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; + +/** + * A configuration for a service to be deployed into the endpoint. + */ +public final class LocalServiceConfiguration<I, O> { + private final RequestListener<I, O> requestListener; + private final Class<I> requestClass; + private final Class<O> replyClass; + private String serviceType; + private String groupName; + private int metric; + + public LocalServiceConfiguration(final RequestListener<I, O> requestListener, final Class<I> requestClass, final Class<O> replyClass) { + this.requestListener = requestListener; + this.requestClass = requestClass; + this.replyClass = replyClass; + } + + public RequestListener<I, O> getRequestListener() { + return requestListener; + } + + public Class<I> getRequestClass() { + return requestClass; + } + + public Class<O> getReplyClass() { + return replyClass; + } + + public String getServiceType() { + return serviceType; + } + + public void setServiceType(final String serviceType) { + this.serviceType = serviceType; + } + + public String getGroupName() { + return groupName; + } + + public void setGroupName(final String groupName) { + this.groupName = groupName; + } + + public int getMetric() { + return metric; + } + + public void setMetric(final int metric) { + this.metric = metric; + } +} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteReplyException.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteReplyException.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteReplyException.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -23,7 +23,7 @@ package org.jboss.remoting; /** - * An exception indicating that a the remote side tried and failed to send a reply message; the remote side would be + * An exception indicating that a the remote side tried and failed to send a reply message. The remote side would be * aware of this type of failure, so the outcome is determinate; thus it extends {@code RemoteExecutionException}. */ public class RemoteReplyException extends RemoteExecutionException { Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteRequestException.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteRequestException.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteRequestException.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,71 @@ +/* + * 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; + +/** + * Remote request exception. Indicates that the request arrived at the remote side, but the handler could not be + * executed for some reason (for example, the request sent might not be of the correct type). + */ +public final class RemoteRequestException extends RemotingException { + + private static final long serialVersionUID = 5494334026096542700L; + + /** + * Constructs a <tt>RemoteRequestException</tt> with no detail message. The cause is not initialized, and may + * subsequently be initialized by a call to {@link #initCause(Throwable) initCause}. + */ + public RemoteRequestException() { + } + + /** + * Constructs a <tt>RemoteRequestException</tt> with the specified detail message. The cause is not initialized, and may + * subsequently be initialized by a call to {@link #initCause(Throwable) initCause}. + * + * @param msg the detail message + */ + public RemoteRequestException(String msg) { + super(msg); + } + + /** + * Constructs a <tt>RemoteRequestException</tt> with the specified cause. The detail message is set to: + * <pre> + * (cause == null ? null : cause.toString())</pre> + * (which typically contains the class and detail message of <tt>cause</tt>). + * + * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method) + */ + public RemoteRequestException(Throwable cause) { + super(cause); + } + + /** + * Constructs a <tt>RemoteRequestException</tt> with the specified detail message and cause. + * + * @param msg the detail message + * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method) + */ + public RemoteRequestException(String msg, Throwable cause) { + super(msg, cause); + } +} Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteServiceConfiguration.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteServiceConfiguration.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RemoteServiceConfiguration.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,76 @@ +/* + * 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.spi.RequestHandlerSource; + +/** + * + */ +public final class RemoteServiceConfiguration { + private String serviceType; + private String groupName; + private String endpointName; + private RequestHandlerSource requestHandlerSource; + private int metric; + + public String getServiceType() { + return serviceType; + } + + public void setServiceType(final String serviceType) { + this.serviceType = serviceType; + } + + public String getGroupName() { + return groupName; + } + + public void setGroupName(final String groupName) { + this.groupName = groupName; + } + + public String getEndpointName() { + return endpointName; + } + + public void setEndpointName(final String endpointName) { + this.endpointName = endpointName; + } + + public RequestHandlerSource getRequestHandlerSource() { + return requestHandlerSource; + } + + public void setRequestHandlerSource(final RequestHandlerSource requestHandlerSource) { + this.requestHandlerSource = requestHandlerSource; + } + + public int getMetric() { + return metric; + } + + public void setMetric(final int metric) { + this.metric = metric; + } +} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/ReplyException.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/ReplyException.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/ReplyException.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -23,24 +23,49 @@ package org.jboss.remoting; /** - * + * An exception type indicating that an operation completed and a reply was received, but it could not be delivered + * to the client. Possible causes include (but are not limited to) class cast problems and unmarshalling problems. */ public class ReplyException extends RemotingException { private static final long serialVersionUID = 5562116026829381932L; + /** + * Constructs a <tt>ReplyException</tt> with no detail message. The cause is not initialized, and may subsequently be + * initialized by a call to {@link #initCause(Throwable) initCause}. + */ public ReplyException() { } - public ReplyException(final String msg) { + /** + * Constructs a <tt>ReplyException</tt> with the specified detail message. The cause is not initialized, and may + * subsequently be initialized by a call to {@link #initCause(Throwable) initCause}. + * + * @param msg the detail message + */ + public ReplyException(String msg) { super(msg); } - public ReplyException(final Throwable cause) { + /** + * Constructs a <tt>ReplyException</tt> with the specified cause. The detail message is set to: + * <pre> + * (cause == null ? null : cause.toString())</pre> + * (which typically contains the class and detail message of <tt>cause</tt>). + * + * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method) + */ + public ReplyException(Throwable cause) { super(cause); } - public ReplyException(final String msg, final Throwable cause) { + /** + * Constructs a <tt>ReplyException</tt> with the specified detail message and cause. + * + * @param msg the detail message + * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method) + */ + public ReplyException(String msg, Throwable cause) { super(msg, cause); } } Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContext.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContext.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContext.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,6 +1,7 @@ package org.jboss.remoting; import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; import java.io.IOException; /** @@ -74,5 +75,5 @@ * * @param command the task to execute */ - void execute(Runnable command); + void execute(Runnable command) throws RejectedExecutionException; } Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextWrapper.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/RequestContextWrapper.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,73 +0,0 @@ -package org.jboss.remoting; - -import org.jboss.remoting.RequestCancelHandler; -import org.jboss.remoting.RequestContext; -import org.jboss.remoting.ClientContext; -import java.io.IOException; - -/** - * A simple delegating wrapper for request context instances. - * - * @param <O> the reply type - */ -public class RequestContextWrapper<O> implements RequestContext<O> { - protected final RequestContext<O> delegate; - - /** - * Construct a new instance. Calls will be sent to the given {@code delegate} by default. - * - * @param delegate the delegate client instance - */ - protected RequestContextWrapper(final RequestContext<O> delegate) { - this.delegate = delegate; - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public ClientContext getContext() { - return delegate.getContext(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public boolean isCancelled() { - return delegate.isCancelled(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void sendReply(O reply) throws IOException, IllegalStateException { - delegate.sendReply(reply); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void sendFailure(String msg, Throwable cause) throws IOException, IllegalStateException { - delegate.sendFailure(msg, cause); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void sendCancelled() throws IOException, IllegalStateException { - delegate.sendCancelled(); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void addCancelHandler(final RequestCancelHandler<O> requestCancelHandler) { - delegate.addCancelHandler(requestCancelHandler); - } - - /** - * {@inheritDoc} This implementation calls the same method on the delegate object. - */ - public void execute(final Runnable command) { - delegate.execute(command); - } -} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractAutoCloseable.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractAutoCloseable.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractAutoCloseable.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -26,13 +26,15 @@ import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicInteger; import org.jboss.remoting.RemotingException; +import org.jboss.remoting.CloseHandler; import org.jboss.xnio.log.Logger; +import org.jboss.xnio.IoUtils; /** * A closeable implementation that supports reference counting. Since the initial reference count is zero, implementors * must be careful to ensure that the first operation invoked is a call to {@link #getHandle()}. */ -public abstract class AbstractAutoCloseable<T> extends AbstractHandleableCloseable<T> { +public abstract class AbstractAutoCloseable<T> extends AbstractHandleableCloseable<T> implements AutoCloseable<T> { private final AtomicInteger refcount = new AtomicInteger(0); private final Executor executor; @@ -100,12 +102,20 @@ } private final class HandleImpl extends AbstractHandleableCloseable<Handle<T>> implements Handle<T> { + private final Key key; + private HandleImpl() throws IOException { super(AbstractAutoCloseable.this.executor); + key = AbstractAutoCloseable.this.addCloseHandler(new CloseHandler<T>() { + public void handleClose(final T closed) { + IoUtils.safeClose(HandleImpl.this); + } + }); inc(); } protected void closeAction() throws IOException { + key.remove(); dec(); } Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractHandleableCloseable.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractHandleableCloseable.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AbstractHandleableCloseable.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -126,12 +126,17 @@ /** * {@inheritDoc} */ - public void addCloseHandler(final CloseHandler<? super T> handler) { + public Key addCloseHandler(final CloseHandler<? super T> handler) { synchronized (closeLock) { if (closeHandlers == null) { closeHandlers = new HashSet<CloseHandler<? super T>>(); } closeHandlers.add(handler); + return new Key() { + public void remove() { + closeHandlers.remove(handler); + } + }; } } Added: remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AutoCloseable.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AutoCloseable.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/AutoCloseable.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,33 @@ +/* + * 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.spi; + +import org.jboss.remoting.HandleableCloseable; +import java.io.IOException; + +/** + * + */ +public interface AutoCloseable<T> extends HandleableCloseable<T> { + Handle<T> getHandle() throws IOException; +} Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/Handle.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/Handle.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/Handle.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -50,5 +50,5 @@ * * @param handler the handler */ - void addCloseHandler(final CloseHandler<? super Handle<T>> handler); + Key addCloseHandler(final CloseHandler<? super Handle<T>> handler); } Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandler.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandler.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandler.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -24,14 +24,12 @@ import java.io.IOException; import org.jboss.remoting.CloseHandler; -import org.jboss.remoting.HandleableCloseable; -import org.jboss.remoting.RemotingException; /** * A request handler, which can be passed to remote endpoints. Remote systems can then use the handler * to make invocations, or they may forward a handler on to other remote systems. */ -public interface RequestHandler extends HandleableCloseable<RequestHandler> { +public interface RequestHandler extends AutoCloseable<RequestHandler> { /** * Receive a request from a remote system. This method is intended to be called by protocol handlers. If the @@ -60,7 +58,7 @@ * Close this request handler. The outcome of any outstanding requests is not defined, though implementations * should make an effort to cancel any outstanding requests. * - * @throws RemotingException if the client endpoint could not be closed + * @throws java.io.IOException if the client endpoint could not be closed */ void close() throws IOException; @@ -69,5 +67,5 @@ * * @param handler the handler to be called */ - void addCloseHandler(final CloseHandler<? super RequestHandler> handler); + Key addCloseHandler(final CloseHandler<? super RequestHandler> handler); } Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandlerSource.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandlerSource.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/spi/RequestHandlerSource.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -24,7 +24,6 @@ import java.io.IOException; import org.jboss.remoting.CloseHandler; -import org.jboss.remoting.HandleableCloseable; import org.jboss.remoting.RemotingException; /** @@ -33,7 +32,7 @@ * has the advantage that a round trip to the remote side is not necessary; the local side can spawn a request handler * and simply notify the remote side of the change. */ -public interface RequestHandlerSource extends HandleableCloseable<RequestHandlerSource> { +public interface RequestHandlerSource extends AutoCloseable<RequestHandlerSource> { /** * Create a request handler for the service corresponding to this request handler source. @@ -64,5 +63,5 @@ * * @param handler the handler to be called */ - void addCloseHandler(final CloseHandler<? super RequestHandlerSource> handler); + Key addCloseHandler(final CloseHandler<? super RequestHandlerSource> handler); } Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSinkWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSinkWrapper.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSinkWrapper.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,26 +0,0 @@ -package org.jboss.remoting.stream; - -import java.io.IOException; - -/** - * - */ -public class ObjectSinkWrapper<T> implements ObjectSink<T> { - private final ObjectSink<T> target; - - public ObjectSinkWrapper(final ObjectSink<T> target) { - this.target = target; - } - - public void accept(final T instance) throws IOException { - target.accept(instance); - } - - public void flush() throws IOException { - target.flush(); - } - - public void close() throws IOException { - target.close(); - } -} Deleted: remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSourceWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSourceWrapper.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/main/java/org/jboss/remoting/stream/ObjectSourceWrapper.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,26 +0,0 @@ -package org.jboss.remoting.stream; - -import java.io.IOException; - -/** - * - */ -public class ObjectSourceWrapper<T> implements ObjectSource<T> { - private final ObjectSource<T> target; - - public ObjectSourceWrapper(final ObjectSource<T> target) { - this.target = target; - } - - public boolean hasNext() throws IOException { - return target.hasNext(); - } - - public T next() throws IOException { - return target.next(); - } - - public void close() throws IOException { - target.close(); - } -} Modified: remoting3/trunk/api/src/test/java/org/jboss/remoting/spi/CloseableTestCase.java =================================================================== --- remoting3/trunk/api/src/test/java/org/jboss/remoting/spi/CloseableTestCase.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/api/src/test/java/org/jboss/remoting/spi/CloseableTestCase.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -27,11 +27,11 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicBoolean; import org.jboss.xnio.IoUtils; import org.jboss.remoting.CloseHandler; import org.jboss.remoting.test.support.LoggingHelper; -import org.jboss.remoting.spi.Handle; /** * @@ -179,4 +179,17 @@ executorService.shutdownNow(); } } + + public void testHandlerRemoval() throws Throwable { + final Executor executor = IoUtils.directExecutor(); + final AbstractAutoCloseable<Object> closeable = new AbstractAutoCloseable<Object>(executor) { + // empty + }; + final Handle<Object> rootHandle = closeable.getHandle(); + try { + + } finally { + IoUtils.safeClose(closeable); + } + } } Modified: remoting3/trunk/build.properties =================================================================== --- remoting3/trunk/build.properties 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/build.properties 2008-11-04 02:34:36 UTC (rev 4642) @@ -115,7 +115,7 @@ lib.jboss-managed.local=${local.repository}/${lib.jboss-managed.local-path} lib.jboss-managed.remote=${remote.repository}/${lib.jboss-managed.remote-path} -lib.marshalling-api.version=1.0.0.Beta2 +lib.marshalling-api.version=1.0.0.CR1 lib.marshalling-api.name=marshalling-api.jar lib.marshalling-api.license=lgpl lib.marshalling-api.dir=jboss/marshalling/${lib.marshalling-api.version}/lib Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientExternalizer.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientExternalizer.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientExternalizer.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -42,15 +42,25 @@ this.endpoint = endpoint; } - @SuppressWarnings({ "unchecked" }) + 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 { - output.writeObject(((ClientImpl)o).getRequestHandlerHandle().getResource()); + doWriteExternal((ClientImpl<?, ?>) o, output); } - @SuppressWarnings({ "unchecked" }) + private <I, O> ClientImpl<I, O> doCreateExternal(Class<I> requestClass, Class<O> replyClass, RequestHandler handler) throws IOException { + return new ClientImpl<I, O>(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 new ClientImpl(handler.getHandle(), endpoint.getExecutor()); + return doCreateExternal(requestClass, replyClass, handler); } public void readExternal(final Object o, final ObjectInput input) throws IOException, ClassNotFoundException { Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientImpl.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientImpl.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientImpl.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -39,10 +39,14 @@ public final class ClientImpl<I, O> extends AbstractContextImpl<Client<I, O>> implements Client<I, O> { private final Handle<RequestHandler> handle; + private final Class<I> requestClass; + private final Class<O> replyClass; - ClientImpl(final Handle<RequestHandler> handle, final Executor executor) { + 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; } protected void closeAction() throws IOException { @@ -53,10 +57,11 @@ if (! isOpen()) { throw new IOException("Client is not open"); } + final I actualRequest = requestClass.cast(request); final QueueExecutor executor = new QueueExecutor(); - final FutureReplyImpl<O> futureReply = new FutureReplyImpl<O>(executor); + final FutureReplyImpl<O> futureReply = new FutureReplyImpl<O>(executor, replyClass); final ReplyHandler replyHandler = futureReply.getReplyHandler(); - final RemoteRequestContext requestContext = handle.getResource().receiveRequest(request, replyHandler); + final RemoteRequestContext requestContext = handle.getResource().receiveRequest(actualRequest, replyHandler); futureReply.setRemoteRequestContext(requestContext); futureReply.addNotifier(new IoFuture.Notifier<O>() { public void notify(final IoFuture<O> future) { @@ -80,7 +85,7 @@ if (! isOpen()) { throw new IOException("Client is not open"); } - final FutureReplyImpl<O> futureReply = new FutureReplyImpl<O>(executor); + final FutureReplyImpl<O> futureReply = new FutureReplyImpl<O>(executor, replyClass); final ReplyHandler replyHandler = futureReply.getReplyHandler(); final RemoteRequestContext requestContext = handle.getResource().receiveRequest(request, replyHandler); futureReply.setRemoteRequestContext(requestContext); @@ -94,4 +99,12 @@ Handle<RequestHandler> getRequestHandlerHandle() { return handle; } + + Class<I> getRequestClass() { + return requestClass; + } + + Class<O> getReplyClass() { + return replyClass; + } } Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceExternalizer.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceExternalizer.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceExternalizer.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -42,15 +42,25 @@ this.endpoint = endpoint; } - @SuppressWarnings({ "unchecked" }) + 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 { - output.writeObject(((ClientSourceImpl) o).getRequestHandlerSourceHandle().getResource()); + doWriteExternal((ClientSourceImpl<?, ?>) o, output); } - @SuppressWarnings({ "unchecked" }) + private <I, O> ClientSourceImpl<I, O> doCreateExternal(Class<I> requestClass, Class<O> replyClass, RequestHandlerSource handlerSource) throws IOException { + return new ClientSourceImpl<I, O>(handlerSource.getHandle(), endpoint, requestClass, replyClass); + } + public Object createExternal(final Class<?> aClass, final ObjectInput input, final Creator creator) throws IOException, ClassNotFoundException { - final RequestHandlerSource handler = (RequestHandlerSource) input.readObject(); - return new ClientSourceImpl(handler.getHandle(), endpoint); + 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 { Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceImpl.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceImpl.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/ClientSourceImpl.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -39,11 +39,15 @@ private final Handle<RequestHandlerSource> handle; private final Endpoint endpoint; + private final Class<I> requestClass; + private final Class<O> replyClass; - ClientSourceImpl(final Handle<RequestHandlerSource> handle, final EndpointImpl endpoint) { + 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; } protected void closeAction() throws IOException { @@ -56,7 +60,7 @@ } final Handle<RequestHandler> clientHandle = handle.getResource().createRequestHandler(); try { - return endpoint.createClient(clientHandle.getResource()); + return endpoint.createClient(clientHandle.getResource(), requestClass, replyClass); } finally { IoUtils.safeClose(clientHandle); } @@ -69,4 +73,12 @@ Handle<RequestHandlerSource> getRequestHandlerSourceHandle() { return handle; } + + Class<I> getRequestClass() { + return requestClass; + } + + Class<O> getReplyClass() { + return replyClass; + } } Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/EndpointImpl.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/EndpointImpl.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/EndpointImpl.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -11,6 +11,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.security.AccessController; import org.jboss.remoting.Client; import org.jboss.remoting.ClientSource; import org.jboss.remoting.CloseHandler; @@ -18,6 +19,9 @@ import org.jboss.remoting.RequestListener; import org.jboss.remoting.ServiceListener; import org.jboss.remoting.SimpleCloseable; +import org.jboss.remoting.LocalServiceConfiguration; +import org.jboss.remoting.EndpointPermission; +import org.jboss.remoting.RemoteServiceConfiguration; import org.jboss.remoting.spi.AbstractSimpleCloseable; import org.jboss.remoting.spi.Handle; import org.jboss.remoting.spi.RequestHandler; @@ -36,13 +40,15 @@ /** * */ -public class EndpointImpl implements Endpoint { +public final class EndpointImpl implements Endpoint { static { // Print Remoting "greeting" message Logger.getLogger("org.jboss.remoting").info("JBoss Remoting version %s", Version.VERSION); } + private static final Logger log = Logger.getLogger(Endpoint.class); + private String name; private OrderedExecutorFactory orderedExecutorFactory; @@ -55,6 +61,13 @@ private final Map<Object, ServiceListenerRegistration> serviceListenerMap = CollectionUtil.hashMap(); private final Set<ServiceRegistration> serviceRegistrations = CollectionUtil.hashSet(); + 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() { } @@ -135,14 +148,23 @@ return endpointMap; } - public <I, O> Handle<RequestHandler> createRequestHandler(final RequestListener<I, O> requestListener) throws IOException { - final LocalRequestHandler<I, O> localRequestHandler = new LocalRequestHandler<I, O>(executor, requestListener); + public <I, O> Handle<RequestHandler> createRequestHandler(final RequestListener<I, O> requestListener, final Class<I> requestClass, final Class<O> replyClass) throws IOException { + AccessController.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); localRequestHandler.addCloseHandler(remover); localRequestHandler.open(); return localRequestHandler.getHandle(); } - public <I, O> Handle<RequestHandlerSource> createRequestHandlerSource(final RequestListener<I, O> requestListener, final String serviceType, final String groupName) throws IOException { + public <I, O> Handle<RequestHandlerSource> registerService(final LocalServiceConfiguration<I, O> configuration) throws IOException { + AccessController.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"); } @@ -155,7 +177,13 @@ if (groupName.length() == 0) { throw new IllegalArgumentException("groupName is empty"); } - final LocalRequestHandlerSource<I, O> localRequestHandlerSource = new LocalRequestHandlerSource<I, O>(executor, requestListener); + 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 { @@ -169,11 +197,11 @@ serviceRegistrations.add(registration); for (final ServiceListenerRegistration slr : serviceListenerMap.values()) { final ServiceListener listener = slr.getServiceListener(); - slr.getExecutor().execute(new Runnable() { - public void run() { - listener.localServiceCreated(slr.handle, serviceType, groupName, localRequestHandlerSource); - } - }); + try { + listener.localServiceCreated(slr.handle, serviceType, groupName, localRequestHandlerSource); + } catch (Throwable t) { + logListenerError(t); + } } } localRequestHandlerSource.addCloseHandler(remover); @@ -181,11 +209,16 @@ return localRequestHandlerSource.getHandle(); } - public <I, O> Client<I, O> createClient(final RequestHandler requestHandler) throws IOException { + 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 { + AccessController.checkPermission(CREATE_CLIENT_PERM); boolean ok = false; final Handle<RequestHandler> handle = requestHandler.getHandle(); try { - final ClientImpl<I, O> client = new ClientImpl<I, O>(handle, executor); + final ClientImpl<I, O> client = new ClientImpl<I, O>(handle, executor, requestType, replyType); client.addCloseHandler(new CloseHandler<Client<I, O>>() { public void handleClose(final Client<I, O> closed) { IoUtils.safeClose(handle); @@ -200,11 +233,12 @@ } } - public <I, O> ClientSource<I, O> createClientSource(final RequestHandlerSource requestHandlerSource) throws IOException { + public <I, O> ClientSource<I, O> createClientSource(final RequestHandlerSource requestHandlerSource, final Class<I> requestClass, final Class<O> replyClass) throws IOException { + AccessController.checkPermission(CREATE_CLIENT_SOURCE_PERM); boolean ok = false; final Handle<RequestHandlerSource> handle = requestHandlerSource.getHandle(); try { - final ClientSourceImpl<I, O> clientSource = new ClientSourceImpl<I, O>(handle, this); + final ClientSourceImpl<I, O> clientSource = new ClientSourceImpl<I, O>(handle, this, requestClass, replyClass); ok = true; return clientSource; } finally { @@ -214,7 +248,7 @@ } } - public <I, O> IoFuture<ClientSource<I, O>> locateService(final URI serviceUri) throws IllegalArgumentException { + 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"); } @@ -262,7 +296,7 @@ } try { // match! - final ClientSource<I, O> clientSource = createClientSource(requestHandlerSource); + final ClientSource<I, O> clientSource = createClientSource(requestHandlerSource, requestClass, replyClass); futureClientSource.setResult(clientSource); } catch (IOException e) { futureClientSource.setException(e); @@ -282,14 +316,23 @@ handlerSource = candidates.get(idx).getHandlerSource(); } try { - return new FinishedIoFuture<ClientSource<I,O>>(EndpointImpl.this.<I, O>createClientSource(handlerSource)); + return new FinishedIoFuture<ClientSource<I,O>>(createClientSource(handlerSource, requestClass, replyClass)); } catch (IOException e) { return new FailedIoFuture<ClientSource<I,O>>(e); } } } - public SimpleCloseable registerRemoteService(final String serviceType, final String groupName, final String endpointName, final RequestHandlerSource handlerSource, final int metric) throws IllegalArgumentException, IOException { + public SimpleCloseable registerRemoteService(final RemoteServiceConfiguration configuration) throws IllegalArgumentException, IOException { + AccessController.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"); } @@ -327,21 +370,21 @@ serviceRegistrations.add(registration); for (final ServiceListenerRegistration slr : serviceListenerMap.values()) { final ServiceListener listener = slr.getServiceListener(); - slr.getExecutor().execute(new Runnable() { - public void run() { - listener.remoteServiceRegistered(slr.handle, endpointName, serviceType, groupName, metric, handlerSource, newHandle); - } - }); + try { + listener.remoteServiceRegistered(slr.handle, endpointName, serviceType, groupName, metric, handlerSource, newHandle); + } catch (Throwable t) { + logListenerError(t); + } } } return newHandle; } public SimpleCloseable addServiceListener(final ServiceListener serviceListener, final boolean onlyNew) { + AccessController.checkPermission(ADD_SERVICE_LISTENER_PERM); final Object key = new Object(); synchronized (serviceLock) { - final Executor orderedExecutor = getOrderedExecutor(); - final ServiceListenerRegistration registration = new ServiceListenerRegistration(serviceListener, orderedExecutor); + final ServiceListenerRegistration registration = new ServiceListenerRegistration(serviceListener); serviceListenerMap.put(key, registration); final AbstractSimpleCloseable handle = new AbstractSimpleCloseable(executor) { protected void closeAction() throws IOException { @@ -353,18 +396,14 @@ registration.setHandle(handle); if (! onlyNew) { for (final ServiceRegistration reg : serviceRegistrations) { - if (reg.isRemote()) { // x is remote - orderedExecutor.execute(new Runnable() { - public void run() { - serviceListener.remoteServiceRegistered(handle, reg.getEndpointName(), reg.getServiceType(), reg.getGroupName(), reg.getMetric(), reg.getHandlerSource(), reg.getHandle()); - } - }); - } else { // x is local - orderedExecutor.execute(new Runnable() { - public void run() { - serviceListener.localServiceCreated(handle, reg.getServiceType(), reg.getGroupName(), reg.getHandlerSource()); - } - }); + try { + if (reg.isRemote()) { // x is remote + serviceListener.remoteServiceRegistered(handle, reg.getEndpointName(), reg.getServiceType(), reg.getGroupName(), reg.getMetric(), reg.getHandlerSource(), reg.getHandle()); + } else { // x is local + serviceListener.localServiceCreated(handle, reg.getServiceType(), reg.getGroupName(), reg.getHandlerSource()); + } + } catch (Throwable t) { + logListenerError(t); } } } @@ -374,22 +413,16 @@ private static final class ServiceListenerRegistration { private final ServiceListener serviceListener; - private final Executor executor; private volatile SimpleCloseable handle; - private ServiceListenerRegistration(final ServiceListener serviceListener, final Executor executor) { + private ServiceListenerRegistration(final ServiceListener serviceListener) { this.serviceListener = serviceListener; - this.executor = executor; } ServiceListener getServiceListener() { return serviceListener; } - Executor getExecutor() { - return executor; - } - void setHandle(final SimpleCloseable handle) { this.handle = handle; } Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/FutureReplyImpl.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/FutureReplyImpl.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/FutureReplyImpl.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -26,6 +26,7 @@ import java.util.concurrent.Executor; import org.jboss.remoting.spi.RemoteRequestContext; import org.jboss.remoting.spi.ReplyHandler; +import org.jboss.remoting.ReplyException; import org.jboss.xnio.AbstractIoFuture; import org.jboss.xnio.IoFuture; @@ -35,11 +36,13 @@ public 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; - public FutureReplyImpl(final Executor executor) { + public FutureReplyImpl(final Executor executor, final Class<O> replyType) { this.executor = executor; + this.replyType = replyType; } void setRemoteRequestContext(final RemoteRequestContext remoteRequestContext) { @@ -62,9 +65,15 @@ private final class Handler implements ReplyHandler { - @SuppressWarnings({ "unchecked" }) public void handleReply(final Object reply) { - setResult((O) 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) { Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandler.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandler.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandler.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -24,9 +24,11 @@ import java.io.IOException; import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; import org.jboss.remoting.CloseHandler; import org.jboss.remoting.RemoteExecutionException; import org.jboss.remoting.RequestListener; +import org.jboss.remoting.RemoteRequestException; import org.jboss.remoting.spi.AbstractAutoCloseable; import org.jboss.remoting.spi.RemoteRequestContext; import org.jboss.remoting.spi.ReplyHandler; @@ -41,37 +43,44 @@ 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(LocalRequestHandler.class); - private LocalRequestHandler(final Executor executor, final RequestListener<I, O> requestListener, final ClientContextImpl clientContext) { - super(executor); - this.requestListener = requestListener; - this.clientContext = clientContext; + LocalRequestHandler(Config<I, O> config) { + super(config.getExecutor()); + requestListener = config.getRequestListener(); + clientContext = config.getClientContext(); + requestClass = config.getRequestClass(); + replyClass = config.getReplyClass(); } - LocalRequestHandler(final Executor executor, final LocalRequestHandlerSource<I, O> service, final RequestListener<I, O> requestListener) { - this(executor, requestListener, new ClientContextImpl(service.getServiceContext())); - } - - LocalRequestHandler(final Executor executor, final RequestListener<I, O> requestListener) { - this(executor, requestListener, new ClientContextImpl(executor)); - } - public RemoteRequestContext receiveRequest(final Object request, final ReplyHandler replyHandler) { - final RequestContextImpl<O> context = new RequestContextImpl<O>(replyHandler, clientContext); - context.execute(new Runnable() { - @SuppressWarnings({ "unchecked" }) - public void run() { - try { - requestListener.handleRequest(context, (I) request); - } catch (RemoteExecutionException e) { - SpiUtils.safeHandleException(replyHandler, e); - } catch (Throwable t) { - SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Request handler threw an exception", t)); + 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(); @@ -101,4 +110,50 @@ public String toString() { return "local request handler <" + Integer.toString(hashCode(), 16) + "> (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; + } + } } Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandlerSource.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandlerSource.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/LocalRequestHandlerSource.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -40,19 +40,27 @@ 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(LocalRequestHandlerSource.class); - LocalRequestHandlerSource(final Executor executor, final RequestListener<I, O> requestListener) { - super(executor); - this.requestListener = requestListener; - this.executor = executor; + 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<I, O> localRequestHandler = new LocalRequestHandler<I, O>(executor, this, requestListener); + 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 { @@ -86,4 +94,40 @@ public String toString() { return "local request handler source <" + Integer.toString(hashCode(), 16) + "> (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; + } + } } Modified: remoting3/trunk/core/src/main/java/org/jboss/remoting/core/RequestContextImpl.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/remoting/core/RequestContextImpl.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/main/java/org/jboss/remoting/core/RequestContextImpl.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -27,6 +27,7 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.Executor; import org.jboss.remoting.ClientContext; import org.jboss.remoting.IndeterminateOutcomeException; import org.jboss.remoting.RemoteExecutionException; @@ -51,13 +52,16 @@ private boolean cancelled; // @protectedby cancelLock private Set<RequestCancelHandler<O>> cancelHandlers; - private final RequestListenerExecutor executor; + private final RequestListenerExecutor interruptingExecutor; + private final Class<O> replyClass; - RequestContextImpl(final ReplyHandler replyHandler, final ClientContextImpl clientContext) { + 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 - executor = new RequestListenerExecutor(clientContext.getExecutor(), this); + interruptingExecutor = new RequestListenerExecutor(executor, this); } public ClientContext getContext() { @@ -72,8 +76,15 @@ public void sendReply(final O reply) throws IOException, IllegalStateException { if (! closed.getAndSet(true)) { + final O actualReply; try { - replyHandler.handleReply(reply); + 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; @@ -118,7 +129,7 @@ } public void execute(final Runnable command) { - executor.execute(command); + interruptingExecutor.execute(command); } protected void cancel() { @@ -127,7 +138,7 @@ cancelled = true; if (cancelHandlers != null) { for (final RequestCancelHandler<O> handler : cancelHandlers) { - executor.execute(new Runnable() { + interruptingExecutor.execute(new Runnable() { public void run() { SpiUtils.safeNotifyCancellation(handler, RequestContextImpl.this); } @@ -135,7 +146,7 @@ } cancelHandlers = null; } - executor.interruptAll(); + interruptingExecutor.interruptAll(); } } } Modified: remoting3/trunk/core/src/test/java/org/jboss/remoting/core/EndpointTestCase.java =================================================================== --- remoting3/trunk/core/src/test/java/org/jboss/remoting/core/EndpointTestCase.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/core/src/test/java/org/jboss/remoting/core/EndpointTestCase.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -92,7 +92,7 @@ } } } - }); + }, INIT_ME, INIT_ME2); final RequestHandler requestHandler = handle.getResource(); try { requestHandler.addCloseHandler(new CloseHandler<RequestHandler>() { @@ -100,7 +100,7 @@ clientEndpointClosed.set(true); } }); - final Client<Object,Object> client = endpoint.createClient(requestHandler); + final Client<Object,Object> client = endpoint.createClient(requestHandler, requestType, replyType); try { client.addCloseHandler(new CloseHandler<Client<Object, Object>>() { public void handleClose(final Client<Object, Object> closed) { @@ -151,7 +151,7 @@ } } } - }); + }, INIT_ME, INIT_ME2); final RequestHandler requestHandler = handle.getResource(); try { requestHandler.addCloseHandler(new CloseHandler<RequestHandler>() { @@ -159,7 +159,7 @@ clientEndpointClosed.set(true); } }); - final Client<Object,Object> client = endpoint.createClient(requestHandler); + final Client<Object,Object> client = endpoint.createClient(requestHandler, requestType, replyType); try { client.addCloseHandler(new CloseHandler<Client<Object, Object>>() { public void handleClose(final Client<Object, Object> closed) { Modified: remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicConfiguration.java =================================================================== --- remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicConfiguration.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicConfiguration.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -23,7 +23,7 @@ package org.jboss.remoting.protocol.basic; import org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.Configuration; +import org.jboss.marshalling.MarshallingConfiguration; import org.jboss.xnio.BufferAllocator; import java.util.concurrent.Executor; import java.nio.ByteBuffer; @@ -33,7 +33,7 @@ */ public final class BasicConfiguration { private MarshallerFactory marshallerFactory; - private Configuration marshallingConfiguration; + private MarshallingConfiguration marshallingConfiguration; private int linkMetric; private Executor executor; private BufferAllocator<ByteBuffer> allocator; @@ -46,11 +46,11 @@ this.marshallerFactory = marshallerFactory; } - public Configuration getMarshallingConfiguration() { + public MarshallingConfiguration getMarshallingConfiguration() { return marshallingConfiguration; } - public void setMarshallingConfiguration(final Configuration marshallingConfiguration) { + public void setMarshallingConfiguration(final MarshallingConfiguration marshallingConfiguration) { this.marshallingConfiguration = marshallingConfiguration; } Modified: remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicProtocol.java =================================================================== --- remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicProtocol.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/basic/src/main/java/org/jboss/remoting/protocol/basic/BasicProtocol.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -28,7 +28,7 @@ import org.jboss.xnio.channels.StreamChannel; import org.jboss.xnio.channels.ChannelOutputStream; import org.jboss.xnio.channels.ChannelInputStream; -import org.jboss.marshalling.Configuration; +import org.jboss.marshalling.MarshallingConfiguration; import org.jboss.marshalling.MarshallerFactory; import org.jboss.marshalling.Unmarshaller; import org.jboss.marshalling.Marshaller; @@ -52,7 +52,7 @@ public static final void createServer(final Handle<RequestHandler> requestHandlerHandle, final StreamChannel streamChannel, final BasicConfiguration configuration) throws IOException { final RequestHandler requestHandler = requestHandlerHandle.getResource(); - final Configuration marshallerConfiguration = configuration.getMarshallingConfiguration(); + final MarshallingConfiguration marshallerConfiguration = configuration.getMarshallingConfiguration(); final MarshallerFactory marshallerFactory = configuration.getMarshallerFactory(); final Marshaller marshaller = marshallerFactory.createMarshaller(marshallerConfiguration); final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallerConfiguration); @@ -60,12 +60,14 @@ marshaller.start(Marshalling.createByteOutput(new ChannelOutputStream(streamChannel))); unmarshaller.start(Marshalling.createByteInput(new ChannelInputStream(streamChannel))); final BlockingQueue<FutureBasicReply> replyQueue = new LinkedBlockingQueue<FutureBasicReply>(); + // todo - handle rejected execution... executor.execute(new BasicServerReplyTransmitter(replyQueue, marshaller, streamChannel, requestHandlerHandle)); + // todo - handle rejected execution... executor.execute(new BasicServerRequestConsumer(unmarshaller, requestHandler, replyQueue, streamChannel, requestHandlerHandle)); } public static final Handle<RequestHandler> createClient(final StreamChannel streamChannel, final BasicConfiguration configuration) throws IOException { - final Configuration marshallerConfiguration = configuration.getMarshallingConfiguration(); + final MarshallingConfiguration marshallerConfiguration = configuration.getMarshallingConfiguration(); final MarshallerFactory marshallerFactory = configuration.getMarshallerFactory(); final Marshaller marshaller = marshallerFactory.createMarshaller(marshallerConfiguration); final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallerConfiguration); @@ -74,6 +76,7 @@ unmarshaller.start(Marshalling.createByteInput(new ChannelInputStream(streamChannel))); final Lock reqLock = new ReentrantLock(); final Queue<ReplyHandler> replyQueue = new LinkedList<ReplyHandler>(); + // todo - handle rejected execution... executor.execute(new BasicHandlerReplyConsumer(unmarshaller, streamChannel, reqLock, replyQueue)); return new BasicRequestHandler(reqLock, marshaller, replyQueue, streamChannel, executor).getHandle(); } Modified: remoting3/trunk/protocol/basic/src/test/java/org/jboss/remoting/protocol/basic/BasicTestCase.java =================================================================== --- remoting3/trunk/protocol/basic/src/test/java/org/jboss/remoting/protocol/basic/BasicTestCase.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/basic/src/test/java/org/jboss/remoting/protocol/basic/BasicTestCase.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -40,7 +40,7 @@ import org.jboss.remoting.test.support.LoggingHelper; import org.jboss.remoting.spi.RequestHandler; import org.jboss.remoting.spi.Handle; -import org.jboss.marshalling.Configuration; +import org.jboss.marshalling.MarshallingConfiguration; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.LinkedBlockingQueue; @@ -61,7 +61,7 @@ final BasicConfiguration configuration = new BasicConfiguration(); configuration.setExecutor(executor); configuration.setMarshallerFactory(new RiverMarshallerFactory()); - final Configuration marshallingConfiguration = new Configuration(); + final MarshallingConfiguration marshallingConfiguration = new MarshallingConfiguration(); configuration.setMarshallingConfiguration(marshallingConfiguration); final Endpoint endpoint = Remoting.createEndpoint("test"); final Handle<RequestHandler> requestHandlerHandle = endpoint.createRequestHandler(new AbstractRequestListener<Object, Object>() { @@ -77,7 +77,7 @@ } } } - }); + }, INIT_ME, INIT_ME2); final ChannelSource<StreamChannel> channelSource = xnio.createPipeServer(executor, IoUtils.singletonHandlerFactory(new IoHandler<StreamChannel>() { public void handleOpened(final StreamChannel channel) { try { @@ -101,7 +101,7 @@ })); final IoFuture<StreamChannel> futureChannel = channelSource.open(IoUtils.nullHandler()); final Handle<RequestHandler> clientHandlerHandle = BasicProtocol.createClient(futureChannel.get(), configuration); - final Client<Object,Object> client = endpoint.createClient(clientHandlerHandle.getResource()); + final Client<Object,Object> client = endpoint.createClient(clientHandlerHandle.getResource(), requestType, replyType); System.out.println("Reply is:" + client.invoke("GORBA!")); } Added: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerBiMap.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerBiMap.java (rev 0) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerBiMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,79 @@ +/* + * 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.protocol.multiplex; + +import java.util.HashMap; +import java.util.IdentityHashMap; + +/** + * + */ +final class IdentityHashIntegerBiMap<T> implements IntegerBiMap<T> { + + private final HashMap<Integer, T> leftMap; + private final IdentityHashMap<T, Integer> rightMap; + + public IdentityHashIntegerBiMap(int initialCapacity, float loadFactor) { + leftMap = new HashMap<Integer, T>(initialCapacity, loadFactor); + rightMap = new IdentityHashMap<T, Integer>((int) (initialCapacity / loadFactor)); + } + + public IdentityHashIntegerBiMap() { + this(256, 0.4f); + } + + public int get(final T key, final int defValue) { + final Integer v = rightMap.get(key); + return v == null ? defValue : v.intValue(); + } + + public T get(final int key) { + return leftMap.get(Integer.valueOf(key)); + } + + public void put(final int key1, final T key2) { + final Integer key1Obj = Integer.valueOf(key1); + final T oldKey2 = leftMap.put(key1Obj, key2); + final Integer oldKey1Obj = rightMap.put(key2, key1Obj); + rightMap.remove(oldKey2); + leftMap.remove(oldKey1Obj); + } + + public T remove(final int key) { + final T oldRightKey = leftMap.remove(Integer.valueOf(key)); + rightMap.remove(oldRightKey); + return oldRightKey; + } + + public void remove(final T key) { + leftMap.remove(rightMap.remove(key)); + } + + public static <T> IntegerBiMap<T> create() { + return new IdentityHashIntegerBiMap<T>(); + } + + public static <T> IntegerBiMap<T> createSynchronizing() { + return IntegerBiMap.Util.synchronizing(new IdentityHashIntegerBiMap<T>()); + } +} Added: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerResourceBiMap.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerResourceBiMap.java (rev 0) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerResourceBiMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,104 @@ +/* + * 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.protocol.multiplex; + +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.Map; +import org.jboss.remoting.spi.AutoCloseable; +import org.jboss.remoting.spi.Handle; + +/** + * + */ +final class IdentityHashIntegerResourceBiMap<T extends AutoCloseable<T>> implements IntegerResourceBiMap<T> { + + private final HashMap<Integer, Handle<T>> leftMap; + private final IdentityHashMap<T, Integer> rightMap; + + public IdentityHashIntegerResourceBiMap(int initialCapacity, float loadFactor) { + leftMap = new HashMap<Integer, Handle<T>>(initialCapacity, loadFactor); + rightMap = new IdentityHashMap<T, Integer>((int) (initialCapacity / loadFactor)); + } + + public IdentityHashIntegerResourceBiMap() { + this(256, 0.4f); + } + + public int get(final T key, final int defValue) { + final Integer v = rightMap.get(key); + return v == null ? defValue : v.intValue(); + } + + public Handle<T> get(final int key) { + return leftMap.get(Integer.valueOf(key)); + } + + public void put(final int key1, final Handle<T> key2) { + final Integer key1Obj = Integer.valueOf(key1); + final Handle<T> oldKey2 = leftMap.put(key1Obj, key2); + final Integer oldKey1Obj = rightMap.put(key2.getResource(), key1Obj); + if (oldKey2 != null) rightMap.remove(oldKey2.getResource()); + if (oldKey1Obj != null) leftMap.remove(oldKey1Obj); + } + + public Handle<T> remove(final int key) { + final Handle<T> oldRightKey = leftMap.remove(Integer.valueOf(key)); + if (oldRightKey != null) rightMap.remove(oldRightKey.getResource()); + return oldRightKey; + } + + public void remove(final T key) { + leftMap.remove(rightMap.remove(key)); + } + + public static <T extends AutoCloseable<T>> IntegerResourceBiMap<T> create() { + return new IdentityHashIntegerResourceBiMap<T>(); + } + + public static <T extends AutoCloseable<T>> IntegerResourceBiMap<T> createSynchronizing() { + return Util.synchronizing(new IdentityHashIntegerResourceBiMap<T>()); + } + + public Iterator<Handle<T>> iterator() { + final Iterator<Map.Entry<Integer, Handle<T>>> delegate = leftMap.entrySet().iterator(); + return new Iterator<Handle<T>>() { + private Map.Entry<Integer, Handle<T>> current; + + public boolean hasNext() { + return delegate.hasNext(); + } + + public Handle<T> next() { + current = delegate.next(); + return current.getValue(); + } + + public void remove() { + delegate.remove(); + rightMap.remove(current.getValue().getResource()); + } + }; + } +} \ No newline at end of file Deleted: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityIntMap.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityIntMap.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityIntMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,153 +0,0 @@ -/* - * 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.protocol.multiplex; - -import java.util.Arrays; - -/** - * - */ -final class IdentityIntMap<T> { - - private int[] values; - private Object[] keys; - private int count; - private int resizeCount; - - public IdentityIntMap(int initialCapacity, final float loadFactor) { - if (initialCapacity < 1) { - throw new IllegalArgumentException("initialCapacity must be > 0"); - } - if (loadFactor <= 0.0f || loadFactor >= 1.0f) { - throw new IllegalArgumentException("loadFactor must be > 0.0 and < 1.0"); - } - if (initialCapacity < 16) { - initialCapacity = 16; - } else { - // round up - final int c = Integer.highestOneBit(initialCapacity) - 1; - initialCapacity = Integer.highestOneBit(initialCapacity + c); - } - keys = new Object[initialCapacity]; - values = new int[initialCapacity]; - resizeCount = (int) ((double) initialCapacity * (double) loadFactor); - } - - public IdentityIntMap(final float loadFactor) { - this(64, loadFactor); - } - - public IdentityIntMap(final int initialCapacity) { - this(initialCapacity, 0.5f); - } - - public IdentityIntMap() { - this(0.5f); - } - - public int get(T key, int defVal) { - if (key == null) { - throw new NullPointerException("key is null"); - } - final Object[] keys = this.keys; - final int mask = keys.length - 1; - int hc = System.identityHashCode(key) & mask; - Object v; - for (;;) { - v = keys[hc]; - if (v == key) { - return values[hc]; - } - if (v == null) { - // not found - return defVal; - } - hc = (hc + 1) & mask; - } - } - - public void put(T key, int value) { - if (key == null) { - throw new NullPointerException("key is null"); - } - final Object[] keys = this.keys; - final int mask = keys.length - 1; - final int[] values = this.values; - Object v; - int hc = System.identityHashCode(key) & mask; - for (int idx = hc;; idx = hc++ & mask) { - v = keys[idx]; - if (v == null) { - keys[idx] = key; - values[idx] = value; - if (++count > resizeCount) { - resize(); - } - return; - } - if (v == key) { - values[idx] = value; - return; - } - } - } - - private final void resize() { - final Object[] oldKeys = keys; - final int oldsize = oldKeys.length; - final int[] oldValues = values; - if (oldsize >= 0x40000000) { - throw new IllegalStateException("Table full"); - } - final int newsize = oldsize << 1; - final int mask = newsize - 1; - final Object[] newKeys = new Object[newsize]; - final int[] newValues = new int[newsize]; - keys = newKeys; - values = newValues; - if ((resizeCount <<= 1) == 0) { - resizeCount = Integer.MAX_VALUE; - } - for (int oi = 0; oi < oldsize; oi ++) { - final Object key = oldKeys[oi]; - if (key != null) { - int ni = System.identityHashCode(key) & mask; - for (;;) { - final Object v = newKeys[ni]; - if (v == null) { - // found - newKeys[ni] = key; - newValues[ni] = oldValues[oi]; - break; - } - ni = (ni + 1) & mask; - } - } - } - } - - public void clear() { - Arrays.fill(keys, null); - count = 0; - } -} Added: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerBiMap.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerBiMap.java (rev 0) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerBiMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,89 @@ +/* + * 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.protocol.multiplex; + +/** + * + */ +interface IntegerBiMap<T> { + int get(T key, int defValue); + + T get(int key); + + void put(int key1, T key2); + + T remove(int key); + + void remove(T key); + + class Util { + + private Util() { + } + + private static class SyncWrapper<T> implements IntegerBiMap<T> { + + private final IntegerBiMap<T> orig; + private final Object lock; + + private SyncWrapper(IntegerBiMap<T> orig, Object lock) { + this.orig = orig; + this.lock = lock; + } + + public int get(final T key, final int defValue) { + synchronized (lock) { + return orig.get(key, defValue); + } + } + + public T get(final int key) { + synchronized (lock) { + return orig.get(key); + } + } + + public void put(final int key1, final T key2) { + synchronized (lock) { + orig.put(key1, key2); + } + } + + public T remove(final int key) { + synchronized (lock) { + return orig.remove(key); + } + } + + public void remove(final T key) { + synchronized (lock) { + orig.remove(key); + } + } + } + + public static <T> IntegerBiMap<T> synchronizing(IntegerBiMap<T> orig) { + return new SyncWrapper<T>(orig, new Object()); + } + } +} Added: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerResourceBiMap.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerResourceBiMap.java (rev 0) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerResourceBiMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -0,0 +1,97 @@ +/* + * 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.protocol.multiplex; + +import org.jboss.remoting.spi.AutoCloseable; +import org.jboss.remoting.spi.Handle; +import java.util.Iterator; + +/** + * + */ +interface IntegerResourceBiMap<T extends AutoCloseable<T>> extends Iterable<Handle<T>> { + int get(T key, int defValue); + + Handle<T> get(int key); + + void put(int key1, Handle<T> key2); + + Handle<T> remove(int key); + + void remove(T key); + + class Util { + + private Util() { + } + + private static class SyncWrapper<T extends AutoCloseable<T>> implements IntegerResourceBiMap<T> { + + private final IntegerResourceBiMap<T> orig; + private final Object lock; + + private SyncWrapper(IntegerResourceBiMap<T> orig, Object lock) { + this.orig = orig; + this.lock = lock; + } + + public int get(final T key, final int defValue) { + synchronized (lock) { + return orig.get(key, defValue); + } + } + + public Handle<T> get(final int key) { + synchronized (lock) { + return orig.get(key); + } + } + + public void put(final int key1, final Handle<T> key2) { + synchronized (lock) { + orig.put(key1, key2); + } + } + + public Handle<T> remove(final int key) { + synchronized (lock) { + return orig.remove(key); + } + } + + public void remove(final T key) { + synchronized (lock) { + orig.remove(key); + } + } + + public Iterator<Handle<T>> iterator() { + return null; + } + } + + public static <T extends AutoCloseable<T>> IntegerResourceBiMap<T> synchronizing(IntegerResourceBiMap<T> orig) { + return new SyncWrapper<T>(orig, new Object()); + } + } +} Modified: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MessageType.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MessageType.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MessageType.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -27,16 +27,26 @@ */ enum MessageType { - // One-way request, no return value may be sent - // Two-way request, return value is expected + /** + * The request part of a request-response sequence, sent from the Client to the RequestListener. + */ REQUEST(2), - // Reply + /** + * The reply part of a request-response sequence, sent from the RequestListener to the Client. + */ REPLY(3), - // Attempt to cancel a request + /** + * A cancellation request for an outstanding request, sent from the Client to the RequestListener. + */ CANCEL_REQUEST(4), - // Acknowledge that a request was cancelled + /** + * Acknowlegement that a request was cancelled, sent from the RequestListener to the Client. + */ CANCEL_ACK(5), - // Request failed due to protocol or unmarshalling problem + /** + * Message that the request could not be received on the remote end, sent from to the Client from the + * protocol handler as a + */ REQUEST_RECEIVE_FAILED(6), // Request failed due to exception REQUEST_FAILED(7), Modified: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConfiguration.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConfiguration.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConfiguration.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -26,14 +26,14 @@ import java.nio.ByteBuffer; import org.jboss.xnio.BufferAllocator; import org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.Configuration; +import org.jboss.marshalling.MarshallingConfiguration; /** * A configuration object for the multiplex protocol. */ public final class MultiplexConfiguration { private MarshallerFactory marshallerFactory; - private Configuration marshallingConfiguration; + private MarshallingConfiguration marshallingConfiguration; private int linkMetric; private Executor executor; private BufferAllocator<ByteBuffer> allocator; @@ -67,7 +67,7 @@ * * @return the configuration */ - public Configuration getMarshallingConfiguration() { + public MarshallingConfiguration getMarshallingConfiguration() { return marshallingConfiguration; } @@ -76,7 +76,7 @@ * * @param marshallingConfiguration the configuration */ - public void setMarshallingConfiguration(final Configuration marshallingConfiguration) { + public void setMarshallingConfiguration(final MarshallingConfiguration marshallingConfiguration) { this.marshallingConfiguration = marshallingConfiguration; } Modified: remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexHandler.java =================================================================== --- remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexHandler.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/protocol/multiplex/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexHandler.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -34,19 +34,19 @@ import org.jboss.remoting.spi.Handle; import org.jboss.remoting.spi.SpiUtils; import org.jboss.remoting.spi.AbstractAutoCloseable; -import static org.jboss.remoting.util.CollectionUtil.concurrentIntegerMap; import org.jboss.remoting.util.CollectionUtil; -import org.jboss.remoting.util.ConcurrentIntegerMap; import org.jboss.remoting.CloseHandler; import org.jboss.remoting.Endpoint; import org.jboss.remoting.SimpleCloseable; import org.jboss.remoting.RemoteExecutionException; import org.jboss.remoting.IndeterminateOutcomeException; +import org.jboss.remoting.ReplyException; +import org.jboss.remoting.RemoteServiceConfiguration; import org.jboss.marshalling.MarshallerFactory; import org.jboss.marshalling.Unmarshaller; import org.jboss.marshalling.ByteOutput; import org.jboss.marshalling.Marshaller; -import org.jboss.marshalling.Configuration; +import org.jboss.marshalling.MarshallingConfiguration; import org.jboss.marshalling.ObjectTable; import org.jboss.marshalling.Marshalling; import java.util.concurrent.BlockingQueue; @@ -60,6 +60,7 @@ import java.io.IOException; import java.io.InvalidClassException; import java.io.InterruptedIOException; +import java.io.InvalidObjectException; /** * Protocol handler for the basic message-oriented Remoting protocol. @@ -70,40 +71,41 @@ //--== Connection configuration items ==-- private final MarshallerFactory marshallerFactory; - private final Configuration marshallingConfiguration; + private final MarshallingConfiguration marshallingConfiguration; private final int linkMetric; private final Executor executor; // buffer allocator for outbound message assembly private final BufferAllocator<ByteBuffer> allocator; // running on remote node - private final ConcurrentIntegerMap<ReplyHandler> remoteRequests = concurrentIntegerMap(); + private final IntegerBiMap<ReplyHandler> remoteRequests = IdentityHashIntegerBiMap.createSynchronizing(); // running on local node - private final ConcurrentIntegerMap<RemoteRequestContext> localRequests = concurrentIntegerMap(); + private final IntegerBiMap<RemoteRequestContext> localRequests = IdentityHashIntegerBiMap.createSynchronizing(); // sequence for remote requests private final AtomicInteger requestSequence = new AtomicInteger(); // clients whose requests get forwarded to the remote side // even #s were opened from services forwarded to us (our sequence) // odd #s were forwarded directly to us (remote sequence) - private final ConcurrentIntegerMap<RequestHandler> remoteClients = concurrentIntegerMap(); + private final IntegerBiMap<RequestHandler> remoteClients = IdentityHashIntegerBiMap.createSynchronizing(); // forwarded to remote side (handled on this side) - private final ConcurrentIntegerMap<Handle<RequestHandler>> forwardedClients = concurrentIntegerMap(); - // sequence for forwarded clients (unsigned; shift left one bit, add one) + private final IntegerResourceBiMap<RequestHandler> forwardedClients = IdentityHashIntegerResourceBiMap.createSynchronizing(); + // sequence for forwarded clients (shift left one bit, add one, limit is 2^30) private final AtomicInteger forwardedClientSequence = new AtomicInteger(); - // sequence for clients created from services forwarded to us (unsigned; shift left one bit) + // sequence for clients created from services forwarded to us (shift left one bit, limit is 2^30) private final AtomicInteger remoteClientSequence = new AtomicInteger(); // services forwarded to us - private final ConcurrentIntegerMap<RequestHandlerSource> remoteServices = concurrentIntegerMap(); + private final IntegerBiMap<RequestHandlerSource> remoteServices = IdentityHashIntegerBiMap.createSynchronizing(); // forwarded to remote side (handled on this side) - private final ConcurrentIntegerMap<Handle<RequestHandlerSource>> forwardedServices = concurrentIntegerMap(); + private final IntegerResourceBiMap<RequestHandlerSource> forwardedServices = IdentityHashIntegerResourceBiMap.createSynchronizing(); // sequence for forwarded services - private final AtomicInteger serviceSequence = new AtomicInteger(); + private final AtomicInteger forwardedServiceSequence = new AtomicInteger(); private final Endpoint endpoint; private volatile AllocatedMessageChannel channel; + private static final StackTraceElement[] emptyStackTraceElements = new StackTraceElement[0]; public MultiplexHandler(final Endpoint endpoint, final MultiplexConfiguration configuration) { this.endpoint = endpoint; @@ -114,6 +116,28 @@ linkMetric = configuration.getLinkMetric(); } + // sequence methods + + int nextRequest() { + return requestSequence.getAndIncrement() & 0x7fffffff; + } + + int nextForwardedClient() { + return (forwardedClientSequence.getAndIncrement() << 1 | 1) & 0x7fffffff; + } + + int nextRemoteClient() { + return remoteClientSequence.getAndIncrement() << 1 & 0x7fffffff; + } + + int nextForwardedService() { + return forwardedServiceSequence.getAndIncrement() & 0x7fffffff; + } + + void setChannel(final AllocatedMessageChannel channel) { + this.channel = channel; + } + public void handleOpened(final AllocatedMessageChannel channel) { channel.resumeReads(); } @@ -165,18 +189,40 @@ payload = unmarshaller.readObject(); unmarshaller.finish(); } catch (ClassNotFoundException e) { - log.trace("Class not found in request ID %d for client ID %d", Integer.valueOf(requestId), Integer.valueOf(clientId)); - // todo - send request receive failed message break; } } finally { IoUtils.safeClose(unmarshaller); } - } catch (IOException ex) { + } catch (Exception ex) { + // IOException | ClassNotFoundException log.trace("Failed to unmarshal a request (%s), sending %s", ex, MessageType.REQUEST_RECEIVE_FAILED); - // todo send a request failure message + try { + final Marshaller marshaller = marshallerFactory.createMarshaller(marshallingConfiguration); + try { + List<ByteBuffer> buffers = new ArrayList<ByteBuffer>(); + marshaller.start(createByteOutput(allocator, buffers)); + marshaller.write(MessageType.REQUEST_RECEIVE_FAILED.getId()); + final IOException ioe = new IOException("Request receive failed"); + ex.setStackTrace(emptyStackTraceElements); + ioe.initCause(ex); + ioe.setStackTrace(emptyStackTraceElements); + marshaller.writeObject(ioe); + marshaller.finish(); + registerWriter(channel, new SimpleWriteHandler(allocator, buffers)); + } catch (InterruptedException e1) { + Thread.currentThread().interrupt(); + log.debug("Remoting channel handler thread interrupted; closing channel"); + IoUtils.safeClose(channel); + } finally { + IoUtils.safeClose(marshaller); + } + } catch (IOException ioe) { + log.warn("Failed to send notification of failure to unmarshal a request: %s", ioe); + } break; } + // request received OK final RequestHandler requestHandler = handle.getResource(); requestHandler.receiveRequest(payload, new ReplyHandlerImpl(channel, requestId, allocator)); break; @@ -207,7 +253,7 @@ } catch (IOException ex) { log.trace("Failed to unmarshal a reply (%s), sending a ReplyException", ex); // todo - SpiUtils.safeHandleException(replyHandler, ex); + SpiUtils.safeHandleException(replyHandler, new ReplyException("Unmarshal failed", ex)); break; } SpiUtils.safeHandleReply(replyHandler, payload); @@ -238,9 +284,24 @@ log.trace("Got reply to unknown request %d", Integer.valueOf(requestId)); break; } - final String reason = readUTFZ(buffer); - - // todo - throw a new ReplyException + final IOException cause; + try { + final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); + try { + unmarshaller.start(Marshalling.createByteInput(buffer)); + cause = (IOException) unmarshaller.readObject(); + unmarshaller.finish(); + } finally { + IoUtils.safeClose(unmarshaller); + } + } catch (IOException e) { + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote operation failed; the remote exception could not be read", e)); + break; + } catch (ClassNotFoundException e) { + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote operation failed; the remote exception could not be read", e)); + break; + } + SpiUtils.safeHandleException(replyHandler, cause); break; } case REQUEST_FAILED: { @@ -250,13 +311,13 @@ log.trace("Got reply to unknown request %d", Integer.valueOf(requestId)); break; } - final Throwable cause; + final IOException cause; try { final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); try { unmarshaller.start(Marshalling.createByteInput(buffer)); try { - cause = (Throwable) unmarshaller.readObject(); + cause = (IOException) unmarshaller.readObject(); } catch (ClassNotFoundException e) { replyHandler.handleException(new InvalidClassException("Class not found: " + e.toString())); log.trace("Class not found in exception reply to request ID %d", Integer.valueOf(requestId)); @@ -339,7 +400,11 @@ final int calcMetric = baseMetric + linkMetric; if (calcMetric > 0) { try { - final SimpleCloseable closeable = endpoint.registerRemoteService(serviceType, groupName, endpointName, handlerSource, calcMetric); + final RemoteServiceConfiguration config = new RemoteServiceConfiguration(); + config.setServiceType(serviceType); + config.setGroupName(groupName); + config.setEndpointName(endpointName); + final SimpleCloseable closeable = endpoint.registerRemoteService(config); // todo - something with that closeable } catch (IOException e) { log.error(e, "Unable to register remote service"); @@ -436,15 +501,14 @@ } public void handleException(final IOException exception) throws IOException { - ByteBuffer buffer = allocator.allocate(); - buffer.put((byte) MessageType.REQUEST_FAILED.getId()); - buffer.putInt(requestId); try { final Marshaller marshaller = marshallerFactory.createMarshaller(marshallingConfiguration); try { final List<ByteBuffer> bufferList = new ArrayList<ByteBuffer>(); final ByteOutput output = createByteOutput(allocator, bufferList); try { + marshaller.write(MessageType.REQUEST_FAILED.getId()); + marshaller.writeInt(requestId); marshaller.start(output); marshaller.writeObject(exception); marshaller.close(); @@ -561,7 +625,7 @@ this.allocator = allocator; addCloseHandler(new CloseHandler<RequestHandler>() { public void handleClose(final RequestHandler closed) { - remoteClients.remove(identifier, this); + remoteClients.remove(identifier); ByteBuffer buffer = allocator.allocate(); buffer.put((byte) MessageType.CLIENT_CLOSE.getId()); buffer.putInt(identifier); @@ -587,10 +651,8 @@ marshaller.write(MessageType.REQUEST.getId()); marshaller.writeInt(identifier); - int id; - do { - id = requestSequence.getAndIncrement(); - } while (remoteRequests.putIfAbsent(id, handler) != null); + final int id = nextRequest(); + remoteRequests.put(id, handler); marshaller.writeInt(id); marshaller.writeObject(request); marshaller.close(); @@ -683,15 +745,13 @@ } public Handle<RequestHandler> createRequestHandler() throws IOException { - int id; - do { - id = remoteClientSequence.getAndIncrement() << 1; - } while (remoteClients.putIfAbsent(id, new RequestHandlerImpl(id, MultiplexHandler.this.allocator)) != null); - final int clientId = id; + final int id = nextRemoteClient(); + final RequestHandler requestHandler = new RequestHandlerImpl(id, MultiplexHandler.this.allocator); + remoteClients.put(id, requestHandler); final ByteBuffer buffer = allocator.allocate(); buffer.put((byte) MessageType.CLIENT_OPEN.getId()); buffer.putInt(identifier); - buffer.putInt(clientId); + buffer.putInt(id); buffer.flip(); // todo - probably should bail out if we're interrupted? boolean intr = false; @@ -699,7 +759,7 @@ try { registerWriter(channel, new SimpleWriteHandler(allocator, buffer)); try { - return new RequestHandlerImpl(clientId, allocator).getHandle(); + return new RequestHandlerImpl(id, allocator).getHandle(); } finally { if (intr) { Thread.currentThread().interrupt(); @@ -769,25 +829,32 @@ }; } + private final ProtocolObjectTableWriter protocolObjectTableWriter = new ProtocolObjectTableWriter(); + public class ProtocolObjectTableWriter implements ObjectTable.Writer { public void writeObject(final Marshaller marshaller, final Object o) throws IOException { - + final RequestHandler requestHandler = (RequestHandler) o; + final int existingId = forwardedClients.get(requestHandler, -1); + marshaller.write(1); + if (existingId == -1) { + final int newId = nextForwardedClient(); + forwardedClients.put(newId, requestHandler.getHandle()); + marshaller.writeInt(newId); + } else { + marshaller.writeInt(existingId); + } } } public class ProtocolObjectTable implements ObjectTable { - public Writer getObjectWriter(final Object o) throws IOException /* fixed in beta2 */ { + public Writer getObjectWriter(final Object o) throws IOException { if (o instanceof RequestHandler) { - final RequestHandler requestHandler = (RequestHandler) o; - - } else if (o instanceof RequestHandlerSource) { - final RequestHandlerSource requestHandlerSource = (RequestHandlerSource) o; - + return protocolObjectTableWriter; } else { + return null; } - return null; } public Object readObject(final Unmarshaller unmarshaller) throws IOException, ClassNotFoundException { @@ -795,18 +862,18 @@ case 1: { // remote client final int id = unmarshaller.readInt(); + return remoteClients.get(id); } case 2: { // remote client source + final int id = unmarshaller.readInt(); + return remoteServices.get(id); } - case 3: { - // stream - } default: { // invalid + throw new InvalidObjectException("Invalid ID sent for protocol object table"); } } - return null; } } } Modified: remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalBasicExampleMain.java =================================================================== --- remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalBasicExampleMain.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalBasicExampleMain.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -18,7 +18,7 @@ final StringRot13RequestListener listener = new StringRot13RequestListener(); final Endpoint endpoint = Remoting.createEndpoint("simple"); try { - final Client<String,String> client = Remoting.createLocalClient(endpoint, listener); + final Client<String,String> client = Remoting.createLocalClient(endpoint, listener, null, null); try { final String original = "The Secret Message\n"; final String result = client.invoke(original); Modified: remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalStreamExampleMain.java =================================================================== --- remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalStreamExampleMain.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/samples/src/main/java/org/jboss/remoting/samples/simple/LocalStreamExampleMain.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -21,7 +21,7 @@ final StreamingRot13RequestListener listener = new StreamingRot13RequestListener(); final Endpoint endpoint = Remoting.createEndpoint("simple"); try { - final Client<Reader,Reader> client = Remoting.createLocalClient(endpoint, listener); + final Client<Reader,Reader> client = Remoting.createLocalClient(endpoint, listener, null, null); try { final String original = "The Secret Message\n"; final StringReader originalReader = new StringReader(original); Modified: remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java =================================================================== --- remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/standalone/src/main/java/org/jboss/remoting/Remoting.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -32,19 +32,19 @@ } } - public static <I, O> Client<I, O> createLocalClient(Endpoint endpoint, RequestListener<I, O> requestListener) throws IOException { - final Handle<RequestHandler> handle = endpoint.createRequestHandler(requestListener); + public static <I, O> Client<I, O> createLocalClient(final Endpoint endpoint, final RequestListener<I, O> requestListener, final Class<I> requestClass, final Class<O> replyClass) throws IOException { + final Handle<RequestHandler> handle = endpoint.createRequestHandler(requestListener, requestClass, replyClass); try { - return endpoint.createClient(handle.getResource()); + return endpoint.createClient(handle.getResource(), requestClass, replyClass); } finally { IoUtils.safeClose(handle); } } - public static <I, O> ClientSource<I, O> createLocalClientSource(Endpoint endpoint, RequestListener<I, O> requestListener, final String serviceType, final String groupName) throws IOException { - final Handle<RequestHandlerSource> handle = endpoint.createRequestHandlerSource(requestListener, serviceType, groupName); + public static <I, O> ClientSource<I, O> createLocalClientSource(final Endpoint endpoint, final LocalServiceConfiguration<I, O> config) throws IOException { + final Handle<RequestHandlerSource> handle = endpoint.registerService(config); try { - return endpoint.createClientSource(handle.getResource()); + return endpoint.createClientSource(handle.getResource(), config.getRequestClass(), config.getReplyClass()); } finally { IoUtils.safeClose(handle); } Modified: remoting3/trunk/transporter/src/main/java/org/jboss/remoting/transporter/Transporter.java =================================================================== --- remoting3/trunk/transporter/src/main/java/org/jboss/remoting/transporter/Transporter.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/transporter/src/main/java/org/jboss/remoting/transporter/Transporter.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -50,9 +50,9 @@ */ public static <T> T createTransporter(Endpoint endpoint, Class<T> interfaceType, T instance) throws IOException { boolean ok = false; - final Handle<RequestHandler> requestHandlerHandle = endpoint.createRequestHandler(new TransporterRequestListener<T>(instance)); + final Handle<RequestHandler> requestHandlerHandle = endpoint.createRequestHandler(new TransporterRequestListener<T>(instance), TransporterInvocation.class, Object.class); try { - final Client<TransporterInvocation,Object> client = endpoint.createClient(requestHandlerHandle.getResource()); + final Client<TransporterInvocation,Object> client = endpoint.createClient(requestHandlerHandle.getResource(), TransporterInvocation.class, Object.class); try { requestHandlerHandle.close(); final T proxy = createProxy(interfaceType, client); @@ -68,8 +68,7 @@ } } - @SuppressWarnings({ "unchecked" }) private static <T> T createProxy(final Class<T> interfaceType, final Client<TransporterInvocation, Object> client) { - return (T) Proxy.newProxyInstance(interfaceType.getClassLoader(), new Class<?>[] { interfaceType }, new TransporterInvocationHandler(client)); + return interfaceType.cast(Proxy.newProxyInstance(interfaceType.getClassLoader(), new Class<?>[] { interfaceType }, new TransporterInvocationHandler(client))); } } Modified: remoting3/trunk/util/src/main/java/org/jboss/remoting/util/CollectionUtil.java =================================================================== --- remoting3/trunk/util/src/main/java/org/jboss/remoting/util/CollectionUtil.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/util/src/main/java/org/jboss/remoting/util/CollectionUtil.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -69,15 +69,6 @@ } /** - * Create a concurrent integer-keyed map. - * - * @return a concurrent integer-keyed map - */ - public static <V> ConcurrentIntegerMap<V> concurrentIntegerMap() { - return new EmulatedConcurrentIntegerHashMap<V>(CollectionUtil.<Integer, V>concurrentMap()); - } - - /** * Create a synchronized map that obeys the contract for {@code ConcurrentMap}. * * @param original the map to be wrapped Deleted: remoting3/trunk/util/src/main/java/org/jboss/remoting/util/ConcurrentIntegerMap.java =================================================================== --- remoting3/trunk/util/src/main/java/org/jboss/remoting/util/ConcurrentIntegerMap.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/util/src/main/java/org/jboss/remoting/util/ConcurrentIntegerMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,51 +0,0 @@ -/* - * 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.util; - -/** - * A concurrent map that is keyed by integer. - */ -public interface ConcurrentIntegerMap<V> { - - V get(int key); - - V put(int key, V value); - - V putIfAbsent(int key, V value); - - V remove(int key); - - boolean remove(int key, Object oldValue); - - V replace(int key, V value); - - boolean replace(int key, V oldValue, V newValue); - - void clear(); - - boolean isEmpty(); - - boolean equals(Object other); - - int hashCode(); -} Deleted: remoting3/trunk/util/src/main/java/org/jboss/remoting/util/EmulatedConcurrentIntegerHashMap.java =================================================================== --- remoting3/trunk/util/src/main/java/org/jboss/remoting/util/EmulatedConcurrentIntegerHashMap.java 2008-10-30 05:00:22 UTC (rev 4641) +++ remoting3/trunk/util/src/main/java/org/jboss/remoting/util/EmulatedConcurrentIntegerHashMap.java 2008-11-04 02:34:36 UTC (rev 4642) @@ -1,81 +0,0 @@ -/* - * 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.util; - -import java.util.concurrent.ConcurrentMap; - -/** - * - */ -public final class EmulatedConcurrentIntegerHashMap<V> implements ConcurrentIntegerMap<V> { - - private final ConcurrentMap<Integer, V> delegate; - - public EmulatedConcurrentIntegerHashMap(final ConcurrentMap<Integer, V> delegate) { - this.delegate = delegate; - } - - public V get(final int key) { - return delegate.get(Integer.valueOf(key)); - } - - public V put(final int key, final V value) { - return delegate.put(Integer.valueOf(key), value); - } - - public V putIfAbsent(final int key, final V value) { - return delegate.putIfAbsent(Integer.valueOf(key), value); - } - - public V remove(final int key) { - return delegate.remove(Integer.valueOf(key)); - } - - public boolean remove(final int key, final Object oldValue) { - return delegate.remove(Integer.valueOf(key), oldValue); - } - - public V replace(final int key, final V value) { - return delegate.replace(Integer.valueOf(key), value); - } - - public boolean replace(final int key, final V oldValue, final V newValue) { - return delegate.replace(Integer.valueOf(key), oldValue, newValue); - } - - public void clear() { - delegate.clear(); - } - - public boolean isEmpty() { - return delegate.isEmpty(); - } - - public boolean equals(final Object obj) { - return super.equals(obj); - } - - public int hashCode() { - return super.hashCode(); - } -}