Author: david.lloyd(a)jboss.com Date: 2008-02-26 15:45:54 -0500 (Tue, 26 Feb 2008) New Revision: 3517 Added: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceReply.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceRequest.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/RemoteResource.java remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/RemoteClassLoader.java Removed: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassReply.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassRequest.java Modified: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/Context.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/FutureReply.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ServiceReply.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/protocol/ProtocolHandler.java remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/wrapper/ContextWrapper.java remoting3/trunk/build.xml remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundContext.java remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundRequest.java remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/LocalProtocol.java remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/ServiceLocatorListener.java remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/stream/InputStreamStreamSerializerFactory.java remoting3/trunk/http/src/main/java/org/jboss/cx/remoting/http/RemotingHttpSessionImpl.java remoting3/trunk/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java remoting3/trunk/util/src/main/java/org/jboss/cx/remoting/util/AtomicStateMachine.java Log: Progress towards classloading, spelling cleanups, interruptible/non-interruptible variants of invocation methods, removal of classloading module since it's so trivial after all... Modified: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/Context.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/Context.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/Context.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -26,9 +26,26 @@ * @throws RemoteExecutionException if the remote handler threw an exception * @throws InterruptedException if the request was interrupted (and thereby cancelled) */ - O invoke(I request) throws RemotingException, RemoteExecutionException, InterruptedException; + O invokeInterruptibly(I request) throws RemotingException, RemoteExecutionException, InterruptedException; /** + * Send a request and block until a reply is received. + * <p/> + * Uses the default invocation policy for handling remote invocations. If the remote side manipulates a stream, the + * current thread MAY be used to handle it. + * <p/> + * If the remote session cannot handle the request, a {@code RemotingException} will be thrown. + * + * @param request the request to send + * + * @return the result of the request + * + * @throws RemotingException if the request could not be sent + * @throws RemoteExecutionException if the remote handler threw an exception + */ + O invoke(I request) throws RemotingException, RemoteExecutionException; + + /** * Send a request asynchronously. * <p/> * Uses the default invocation policy for handling remote invocations. If the remote side manipulates a stream, it Modified: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/FutureReply.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/FutureReply.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/FutureReply.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -51,9 +51,19 @@ * * @throws CancellationException if the computation was cancelled * @throws RemoteExecutionException if the computation threw an exception + */ + T get() throws CancellationException, RemoteExecutionException; + + /** + * Waits if necessary for the request to complete, and then retrieves its reply. + * + * @return the reply + * + * @throws CancellationException if the computation was cancelled + * @throws RemoteExecutionException if the computation threw an exception * @throws InterruptedException if the current thread was interrupted while waiting */ - T get() throws InterruptedException, CancellationException, RemoteExecutionException; + T getInterruptibly() throws InterruptedException, CancellationException, RemoteExecutionException; /** * Waits if necessary for at most the given time for the request to complete, and then retrieves the reply, if @@ -66,9 +76,23 @@ * * @throws CancellationException if the computation was cancelled * @throws RemoteExecutionException if the computation threw an exception + */ + T get(long timeout, TimeUnit unit) throws CancellationException, RemoteExecutionException; + + /** + * Waits if necessary for at most the given time for the request to complete, and then retrieves the reply, if + * available. If no reply was available, {@code null} is returned. + * + * @param timeout the maximum time to wait + * @param unit the time unit of the timeout argument + * + * @return the reply, or {@code null} if the operation timed out + * + * @throws CancellationException if the computation was cancelled + * @throws RemoteExecutionException if the computation threw an exception * @throws InterruptedException if the current thread was interrupted while waiting */ - T get(long timeout, TimeUnit unit) throws InterruptedException, CancellationException, RemoteExecutionException; + T getInterruptibly(long timeout, TimeUnit unit) throws InterruptedException, CancellationException, RemoteExecutionException; /** * Add a notifier to be called when the request has completed. The notifier may be called from the current thread Copied: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceReply.java (from rev 3502, remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassReply.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceReply.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceReply.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -0,0 +1,24 @@ +package org.jboss.cx.remoting.service; + +import java.io.Serializable; +import org.jboss.cx.remoting.stream.ObjectSource; + +/** + * + */ +public final class ClassLoaderResourceReply implements Serializable { + private static final long serialVersionUID = 1L; + + private ObjectSource<RemoteResource> resources; + + public ClassLoaderResourceReply() { + } + + public ObjectSource<RemoteResource> getResources() { + return resources; + } + + public void setResources(final ObjectSource<RemoteResource> resources) { + this.resources = resources; + } +} Copied: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceRequest.java (from rev 3502, remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassRequest.java) =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceRequest.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassLoaderResourceRequest.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -0,0 +1,27 @@ +package org.jboss.cx.remoting.service; + +import java.io.Serializable; + +/** + * + */ +public final class ClassLoaderResourceRequest implements Serializable { + private static final long serialVersionUID = 1L; + + private String name; + + public ClassLoaderResourceRequest() { + } + + public ClassLoaderResourceRequest(final String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(final String name) { + this.name = name; + } +} Deleted: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassReply.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassReply.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassReply.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -1,23 +0,0 @@ -package org.jboss.cx.remoting.service; - -import java.io.Serializable; - -/** - * - */ -public final class ClassReply implements Serializable { - private static final long serialVersionUID = 1L; - - private byte[] classBytes; - - public ClassReply() { - } - - public byte[] getClassBytes() { - return classBytes; - } - - public void setClassBytes(final byte[] classBytes) { - this.classBytes = classBytes; - } -} Deleted: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassRequest.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassRequest.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ClassRequest.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -1,23 +0,0 @@ -package org.jboss.cx.remoting.service; - -import java.io.Serializable; - -/** - * - */ -public final class ClassRequest implements Serializable { - private static final long serialVersionUID = 1L; - - private String name; - - public ClassRequest() { - } - - public String getName() { - return name; - } - - public void setName(final String name) { - this.name = name; - } -} Added: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/RemoteResource.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/RemoteResource.java (rev 0) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/RemoteResource.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -0,0 +1,30 @@ +package org.jboss.cx.remoting.service; + +import java.io.Serializable; +import java.io.InputStream; + +/** + * + */ +public final class RemoteResource implements Serializable { + private static final long serialVersionUID = 1L; + + private InputStream inputStream; + private int size; + + public InputStream getInputStream() { + return inputStream; + } + + public void setInputStream(final InputStream inputStream) { + this.inputStream = inputStream; + } + + public int getSize() { + return size; + } + + public void setSize(final int size) { + this.size = size; + } +} Modified: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ServiceReply.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ServiceReply.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/service/ServiceReply.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -11,7 +11,7 @@ private static final long serialVersionUID = 1L; private ContextSource<I, O> serviceContextSource; - private Context<ClassRequest, ClassReply> classLoadingContext; + private Context<ClassLoaderResourceRequest, ClassLoaderResourceReply> classLoadingContext; public ServiceReply() { } @@ -24,11 +24,11 @@ this.serviceContextSource = serviceContextSource; } - public Context<ClassRequest, ClassReply> getClassLoadingContext() { + public Context<ClassLoaderResourceRequest, ClassLoaderResourceReply> getClassLoadingContext() { return classLoadingContext; } - public void setClassLoadingContext(final Context<ClassRequest, ClassReply> classLoadingContext) { + public void setClassLoadingContext(final Context<ClassLoaderResourceRequest, ClassLoaderResourceReply> classLoadingContext) { this.classLoadingContext = classLoadingContext; } } Modified: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/protocol/ProtocolHandler.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/protocol/ProtocolHandler.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/protocol/ProtocolHandler.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -88,10 +88,10 @@ ContextIdentifier getRemoteRootContextIdentifier(); /** - * Get a new context identifier that will be used to send requests to the remote side. The service identifier - * was received from the remote side. Should send a message to the remote side such that the + * Get a new context identifier. The service identifier was received from the remote side. Should send a message + * to the remote side such that the * {@link ProtocolContext#receiveOpenedContext(ServiceIdentifier, ContextIdentifier)} method is called with - * the new service and context identifiers. + * the service and context identifiers. * * @param serviceIdentifier the service identifier * @return a context identifier associated with the given service identifier @@ -107,7 +107,7 @@ * @param contextIdentifier * @throws IOException if an I/O error occurs */ - void closeContext(ContextIdentifier contextIdentifier) throws IOException; + void sendContextClose(ContextIdentifier contextIdentifier) throws IOException; /** * Acquire a new request identifier that will be used to send a request. @@ -119,14 +119,6 @@ RequestIdentifier openRequest(ContextIdentifier contextIdentifier) throws IOException; /** - * Get a new service identifier that may be transmitted to the remote side. - * - * @return the new service identifier - * @throws IOException if an I/O error occurs - */ - ServiceIdentifier openService() throws IOException; - - /** * Send a notification that the client is no longer using the given service. * * @param serviceIdentifier the service identifier @@ -166,6 +158,14 @@ ContextIdentifier openContext() throws IOException; /** + * Get a new service identifier that may be transmitted to the remote side. + * + * @return the new service identifier + * @throws IOException if an I/O error occurs + */ + ServiceIdentifier openService() throws IOException; + + /** * Open a stream on this session. Since either side may open a stream, it is important that the client and * server side take precautions to ensure that both the client and server will not initiate the same stream at * the same time. Modified: remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/wrapper/ContextWrapper.java =================================================================== --- remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/wrapper/ContextWrapper.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/api/src/main/java/org/jboss/cx/remoting/spi/wrapper/ContextWrapper.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -29,7 +29,11 @@ }); } - public O invoke(final I request) throws RemotingException, RemoteExecutionException, InterruptedException { + public O invokeInterruptibly(final I request) throws RemotingException, RemoteExecutionException, InterruptedException { + return delegate.invokeInterruptibly(request); + } + + public O invoke(final I request) throws RemotingException, RemoteExecutionException { return delegate.invoke(request); } Modified: remoting3/trunk/build.xml =================================================================== --- remoting3/trunk/build.xml 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/build.xml 2008-02-26 20:45:54 UTC (rev 3517) @@ -722,45 +722,6 @@ </path> </target> - <!-- remote-classloader module --> - - <target name="remote-classloader.compile.depcheck"> - <mkdir dir="remote-classloader/target/main"/> - <uptodate property="remote-classloader.compile.uptodate" targetfile="remote-classloader/target/main/.lastcompile"> - <srcfiles dir="remote-classloader/src/main/java"> - <include name="**/"/> - <include name="**/*.java"/> - <exclude name="**/.*"/> - </srcfiles> - </uptodate> - </target> - - <target name="remote-classloader.compile" depends="remote-classloader.compile.depcheck" unless="remote-classloader.compile.uptodate"> - <mkdir dir="remote-classloader/target/main/classes"/> - <javac - source="${javac.source}" - target="${javac.target}" - srcdir="remote-classloader/src/main/java" - destdir="remote-classloader/target/main/classes" - debug="true"> - <compilerarg value="-Xlint:unchecked"/> - <classpath> - <path refid="api.classpath"/> - </classpath> - </javac> - <touch file="remote-classloader/target/main/.lastcompile" verbose="false"/> - </target> - - <target name="remote-classloader.clean"> - <delete dir="remote-classloader/target"/> - </target> - - <target name="remote-classloader" description="Build the remote-classloader module" depends="api,remote-classloader.compile"> - <path id="remote-classloader.classpath"> - <pathelement location="remote-classloader/target/main/classes"/> - </path> - </target> - <!-- samples module --> <target name="samples.compile.depcheck"> @@ -1074,6 +1035,7 @@ <pathelement location="version/target/main/classes"/> </path> <java classpathref="version.classpath" classname="org.jboss.cx.remoting.version.Version" outputproperty="version"/> + <property name="version" value="UNKNOWN"/> </target> <!-- ============================================== --> @@ -1229,12 +1191,6 @@ <target name="clean-http" description="Clean all HTTP targets" depends="http.clean,http-mina-client.clean,http-mina-server.clean,http-se6.clean,http-servlet.clean,http-urlconnection.clean"/> - <!-- interceptors --> - - <target name="all-interceptors" description="Build all interceptor targets" depends="remote-classloader,transaction"/> - - <target name="clean-interceptors" description="Clean all interceptor targets" depends="remote-classloader.clean,transaction.clean"/> - <!-- jrpp --> <target name="all-jrpp" description="Build all JRPP targets" depends="jrpp,mina-sasl,sasl-null,srp"/> @@ -1257,8 +1213,8 @@ <!-- all: These should be the last targets in the file --> - <target name="all" description="Build everything" depends="all-core,all-http,all-interceptors,all-jrpp,all-log,all-jars"/> + <target name="all" description="Build everything" depends="all-core,all-http,all-jrpp,all-log,all-jars"/> - <target name="clean" description="Clean out all build files" depends="clean-core,clean-http,clean-interceptors,clean-jrpp,clean-log"/> + <target name="clean" description="Clean out all build files" depends="clean-core,clean-http,clean-jrpp,clean-log"/> </project> Modified: remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundContext.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundContext.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundContext.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -137,23 +137,41 @@ // todo ... } - public O invoke(final I request) throws RemotingException, RemoteExecutionException, InterruptedException { + private FutureReply<O> doSend(final I request, final QueueExecutor queueExecutor) throws RemotingException { + final RequestIdentifier requestIdentifier; + requestIdentifier = openRequest(); + final CoreOutboundRequest<I, O> outboundRequest = new CoreOutboundRequest<I, O>(CoreOutboundContext.this, requestIdentifier); + requests.put(requestIdentifier, outboundRequest); + // Request must be sent *after* the identifier is registered in the map + sendRequest(requestIdentifier, request, queueExecutor); + final FutureReply<O> futureReply = outboundRequest.getFutureReply(); + futureReply.addCompletionNotifier(new RequestCompletionHandler<O>() { + public void notifyComplete(final FutureReply<O> futureReply) { + queueExecutor.shutdown(); + } + }); + return futureReply; + } + + public O invokeInterruptibly(final I request) throws RemotingException, RemoteExecutionException, InterruptedException { state.requireHold(State.UP); try { - final RequestIdentifier requestIdentifier; - requestIdentifier = openRequest(); - final CoreOutboundRequest<I, O> outboundRequest = new CoreOutboundRequest<I, O>(CoreOutboundContext.this, requestIdentifier); - requests.put(requestIdentifier, outboundRequest); - // Request must be sent *after* the identifier is registered in the map final QueueExecutor queueExecutor = new QueueExecutor(); - sendRequest(requestIdentifier, request, queueExecutor); - final FutureReply<O> futureReply = outboundRequest.getFutureReply(); - futureReply.addCompletionNotifier(new RequestCompletionHandler<O>() { - public void notifyComplete(final FutureReply<O> futureReply) { - queueExecutor.shutdown(); - } - }); + final FutureReply<O> futureReply = doSend(request, queueExecutor); + // todo - find a safe way to make this interruptable queueExecutor.runQueue(); + return futureReply.getInterruptibly(); + } finally { + state.release(); + } + } + + public O invoke(final I request) throws RemotingException, RemoteExecutionException { + state.requireHold(State.UP); + try { + final QueueExecutor queueExecutor = new QueueExecutor(); + final FutureReply<O> futureReply = doSend(request, queueExecutor); + queueExecutor.runQueue(); return futureReply.get(); } finally { state.release(); Modified: remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundRequest.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundRequest.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/CoreOutboundRequest.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -158,8 +158,8 @@ return state.in(State.DONE); } - public O get() throws InterruptedException, CancellationException, RemoteExecutionException { - final State newState = state.waitInterruptablyForNotHold(State.WAITING); + public O get() throws CancellationException, RemoteExecutionException { + final State newState = state.waitForNotHold(State.WAITING); try { switch(newState) { case CANCELLED: @@ -178,9 +178,29 @@ } } - public O get(long timeout, TimeUnit unit) throws InterruptedException, CancellationException, RemoteExecutionException { - final State newState = state.waitInterruptablyForNotHold(State.WAITING, timeout, unit); + public O getInterruptibly() throws InterruptedException, CancellationException, RemoteExecutionException { + final State newState = state.waitInterruptiblyForNotHold(State.WAITING); try { + switch(newState) { + case CANCELLED: + throw new CancellationException("Request was cancelled"); + case EXCEPTION: + throw exception; + case DONE: + return reply; + case TERMINATED: + throw new IndeterminateOutcomeException("Request terminated abruptly; outcome unknown"); + default: + throw new IllegalStateException("Wrong state"); + } + } finally { + state.release(); + } + } + + public O get(long timeout, TimeUnit unit) throws CancellationException, RemoteExecutionException { + final State newState = state.waitForNotHold(State.WAITING, timeout, unit); + try { switch (newState) { case CANCELLED: throw new CancellationException("Request was cancelled"); @@ -199,6 +219,10 @@ } } + public O getInterruptibly(final long timeout, final TimeUnit unit) throws InterruptedException, CancellationException, RemoteExecutionException { + return null; + } + public FutureReply<O> addCompletionNotifier(RequestCompletionHandler<O> handler) { final State currentState = state.getStateHold(); try { Modified: remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/LocalProtocol.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/LocalProtocol.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/LocalProtocol.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -1,8 +1,6 @@ package org.jboss.cx.remoting.core; import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; import java.net.URI; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; @@ -12,7 +10,6 @@ import org.jboss.cx.remoting.util.MessageOutput; import org.jboss.cx.remoting.util.AttributeMap; import org.jboss.cx.remoting.util.CollectionUtil; -import org.jboss.cx.remoting.util.MessageInput; import org.jboss.cx.remoting.log.Logger; import org.jboss.cx.remoting.spi.protocol.ContextIdentifier; import org.jboss.cx.remoting.spi.protocol.ProtocolContext; @@ -111,7 +108,7 @@ public void closeService(ServiceIdentifier serviceIdentifier) throws IOException { } - public void closeContext(ContextIdentifier contextIdentifier) throws IOException { + public void sendContextClose(ContextIdentifier contextIdentifier) throws IOException { log.trace("Closing context for local protocol"); remoteContext.closeContext(contextIdentifier); } Added: remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/RemoteClassLoader.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/RemoteClassLoader.java (rev 0) +++ remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/RemoteClassLoader.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -0,0 +1,64 @@ +package org.jboss.cx.remoting.core; + +import java.io.IOException; +import java.io.InputStream; +import org.jboss.cx.remoting.Context; +import org.jboss.cx.remoting.RemoteExecutionException; +import org.jboss.cx.remoting.RemotingException; +import org.jboss.cx.remoting.log.Logger; +import org.jboss.cx.remoting.service.ClassLoaderResourceRequest; +import org.jboss.cx.remoting.service.ClassLoaderResourceReply; +import org.jboss.cx.remoting.service.RemoteResource; +import org.jboss.cx.remoting.stream.ObjectSource; + +/** + * + */ +public final class RemoteClassLoader extends ClassLoader { + private static final Logger log = Logger.getLogger(RemoteClassLoader.class); + + private final Context<ClassLoaderResourceRequest, ClassLoaderResourceReply> loaderContext; + + public RemoteClassLoader(ClassLoader parent, final Context<ClassLoaderResourceRequest, ClassLoaderResourceReply> loaderContext) { + super(parent); + this.loaderContext = loaderContext; + } + + protected Class<?> findClass(String name) throws ClassNotFoundException { + try { + final ClassLoaderResourceReply reply = loaderContext.invoke(new ClassLoaderResourceRequest(name + ".class")); + final ObjectSource<RemoteResource> source = reply.getResources(); + try { + if (! source.hasNext()) { + throw new ClassNotFoundException("No resources matched"); + } + final RemoteResource resource = source.next(); + final InputStream stream = resource.getInputStream(); + try { + final int size = resource.getSize(); + final byte[] bytes = new byte[size]; + for (int t = 0; t < size; t += stream.read(bytes, t, size - t)); + return defineClass(name, bytes, 0, size); + } finally { + try { + stream.close(); + } catch (IOException e) { + log.trace("Stream close failed", e); + } + } + } finally { + try { + source.close(); + } catch (IOException e) { + log.trace("Resource ObjectSource close failed", e); + } + } + } catch (RemotingException e) { + throw new ClassNotFoundException("Cannot load class " + name + " due to an invocation failure", e); + } catch (RemoteExecutionException e) { + throw new ClassNotFoundException("Cannot load class " + name + " due to a remote invocation failure", e.getCause()); + } catch (IOException e) { + throw new ClassNotFoundException("Cannot load class " + name + " due to an input/output error", e); + } + } +} Modified: remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/ServiceLocatorListener.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/ServiceLocatorListener.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/ServiceLocatorListener.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -11,18 +11,32 @@ import org.jboss.cx.remoting.service.ServiceReply; import java.net.URI; import java.util.concurrent.ConcurrentMap; +import java.util.SortedMap; /** * */ public final class ServiceLocatorListener<I, O> implements RequestListener<ServiceRequest<I, O>, ServiceReply<I, O>> { + private interface Service { + String getGroupName(); + + String getType(); + + // todo - add in whatever negotation to the request object (security?) + <X, Y> Context<Void, ServiceReply<X, Y>> getServiceChannel(); + } + private interface Peer { String getName(); int getCost(); <X, Y> Context<ServiceRequest<X, Y>, ServiceReply<X, Y>> getLocatorContext(); + + SortedMap<String, Service> getServicesByGroupName(); + + SortedMap<String, Service> getServicesByType(); } private static <K, V> ConcurrentMap<K, V> syncMap() { @@ -40,4 +54,6 @@ } + + } Modified: remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/stream/InputStreamStreamSerializerFactory.java =================================================================== --- remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/stream/InputStreamStreamSerializerFactory.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/core/src/main/java/org/jboss/cx/remoting/core/stream/InputStreamStreamSerializerFactory.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -22,6 +22,7 @@ } public StreamSerializer getLocalSide(StreamContext context, Object local) throws IOException { + return new StreamSerializerImpl(context, (InputStream)local); } Modified: remoting3/trunk/http/src/main/java/org/jboss/cx/remoting/http/RemotingHttpSessionImpl.java =================================================================== --- remoting3/trunk/http/src/main/java/org/jboss/cx/remoting/http/RemotingHttpSessionImpl.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/http/src/main/java/org/jboss/cx/remoting/http/RemotingHttpSessionImpl.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -206,7 +206,7 @@ return contextIdentifier; } - public void closeContext(final ContextIdentifier contextIdentifier) throws IOException { + public void sendContextClose(final ContextIdentifier contextIdentifier) throws IOException { outgoingQueue.add(new OutputAction() { public void run(ByteOutput target) throws IOException { final MessageOutput msgOutput = protocolContext.getMessageOutput(target); Modified: remoting3/trunk/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java =================================================================== --- remoting3/trunk/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/jrpp/src/main/java/org/jboss/cx/remoting/jrpp/JrppConnection.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -459,7 +459,7 @@ output.commit(); } - public void closeContext(ContextIdentifier contextIdentifier) throws IOException { + public void sendContextClose(ContextIdentifier contextIdentifier) throws IOException { if (! state.in(State.UP)) { return; } Modified: remoting3/trunk/util/src/main/java/org/jboss/cx/remoting/util/AtomicStateMachine.java =================================================================== --- remoting3/trunk/util/src/main/java/org/jboss/cx/remoting/util/AtomicStateMachine.java 2008-02-26 14:49:26 UTC (rev 3516) +++ remoting3/trunk/util/src/main/java/org/jboss/cx/remoting/util/AtomicStateMachine.java 2008-02-26 20:45:54 UTC (rev 3517) @@ -207,7 +207,7 @@ } - public void waitInterruptablyFor(final T state) throws InterruptedException { + public void waitInterruptiblyFor(final T state) throws InterruptedException { writeLock.lockInterruptibly(); try { while (this.state != state) { @@ -250,7 +250,7 @@ } } - public boolean waitInterruptablyFor(final T state, final long timeout, final TimeUnit timeUnit) throws InterruptedException { + public boolean waitInterruptiblyFor(final T state, final long timeout, final TimeUnit timeUnit) throws InterruptedException { final long timeoutMillis = timeUnit.toMillis(timeout); final long startTime = System.currentTimeMillis(); final long endTime = startTime + timeoutMillis < 0 ? Long.MAX_VALUE : startTime + timeoutMillis; @@ -268,7 +268,7 @@ } } - public T waitInterruptablyForNot(final T state) throws InterruptedException { + public T waitInterruptiblyForNot(final T state) throws InterruptedException { writeLock.lockInterruptibly(); try { while (this.state == state) { @@ -280,7 +280,7 @@ } } - public T waitInterruptablyForNotHold(final T state) throws InterruptedException { + public T waitInterruptiblyForNotHold(final T state) throws InterruptedException { writeLock.lockInterruptibly(); try { while (this.state == state) { @@ -326,7 +326,7 @@ return this.state; } - public T waitInterruptablyForNot(final T state, final long timeout, final TimeUnit timeUnit) throws InterruptedException { + public T waitInterruptiblyForNot(final T state, final long timeout, final TimeUnit timeUnit) throws InterruptedException { final long timeoutMillis = timeUnit.toMillis(timeout); final long startTime = System.currentTimeMillis(); final long endTime = startTime + timeoutMillis < 0 ? Long.MAX_VALUE : startTime + timeoutMillis; @@ -343,15 +343,20 @@ } - public T waitInterruptablyForNotHold(final T state, final long timeout, final TimeUnit timeUnit) throws InterruptedException { + public T waitInterruptiblyForNotHold(final T state, final long timeout, final TimeUnit timeUnit) throws InterruptedException { final long timeoutMillis = timeUnit.toMillis(timeout); final long startTime = System.currentTimeMillis(); final long endTime = startTime + timeoutMillis < 0 ? Long.MAX_VALUE : startTime + timeoutMillis; final Date deadLine = new Date(endTime); + boolean waiting = true; writeLock.lockInterruptibly(); try { while (this.state == state) { - cond.awaitUntil(deadLine); + if (waiting) { + waiting = cond.awaitUntil(deadLine); + } else { + break; + } } readLock.lockInterruptibly(); return this.state; @@ -360,6 +365,37 @@ } } + public T waitForNotHold(final T state, final long timeout, final TimeUnit timeUnit) { + final long timeoutMillis = timeUnit.toMillis(timeout); + final long startTime = System.currentTimeMillis(); + final long endTime = startTime + timeoutMillis < 0 ? Long.MAX_VALUE : startTime + timeoutMillis; + final Date deadLine = new Date(endTime); + boolean intr = false; + try { + boolean waiting = true; + writeLock.lock(); + try { + while (this.state == state) { + if (waiting) { + try { + waiting = cond.awaitUntil(deadLine); + } catch (InterruptedException e) { + intr = Thread.currentThread().isInterrupted(); + } + } else { + break; + } + } + readLock.lock(); + return this.state; + } finally { + writeLock.unlock(); + } + } finally { + if (intr) Thread.currentThread().interrupt(); + } + } + public T waitForNot(final T state, final long timeout, final TimeUnit timeUnit) { final long timeoutMillis = timeUnit.toMillis(timeout); final long startTime = System.currentTimeMillis();