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

Author: david.lloyd(a)jboss.com Date: 2009-10-01 16:56:19 -0400 (Thu, 01 Oct 2009) New Revision: 5553 Modified: remoting3/trunk/jboss-remoting/pom.xml remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Client.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ClientImpl.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/InputStreamHandlerFactory.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/OutputStreamHandlerFactory.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/StreamHandler.java Log: Bring API up to XNIO 2 compatibility Modified: remoting3/trunk/jboss-remoting/pom.xml =================================================================== --- remoting3/trunk/jboss-remoting/pom.xml 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/pom.xml 2009-10-01 20:56:19 UTC (rev 5553) @@ -34,7 +34,7 @@ <dependency> <groupId>org.jboss.xnio</groupId> <artifactId>xnio-api</artifactId> - <version>1.2.0.GA</version> + <version>2.0.0.CR3-SNAPSHOT</version> <scope>compile</scope> </dependency> <dependency> Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Client.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Client.java 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Client.java 2009-10-01 20:56:19 UTC (rev 5553) @@ -108,7 +108,7 @@ * * @return a future representing the result of the request * - * @throws ObjectStreamException if marshalling or unmarshalling some part of the request failed + * @throws ObjectStreamException if marshalling some part of the request failed * @throws IOException if some other I/O error occurred while sending the request */ IoFuture<? extends O> send(I request) throws IOException; Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ClientImpl.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ClientImpl.java 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ClientImpl.java 2009-10-01 20:56:19 UTC (rev 5553) @@ -75,7 +75,7 @@ final ReplyHandler replyHandler = futureReply.getReplyHandler(); final RemoteRequestContext requestContext = handler.receiveRequest(actualRequest, replyHandler); futureReply.setRemoteRequestContext(requestContext); - futureReply.addNotifier(IoUtils.<O>attachmentClosingNotifier(), executor); + futureReply.addNotifier(IoUtils.attachmentClosingNotifier(), executor); executor.runQueue(); try { final O reply = futureReply.getInterruptibly(); Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java 2009-10-01 20:56:19 UTC (rev 5553) @@ -115,7 +115,7 @@ log.trace("Closed %s", this); if (closeHandlers != null) { for (final CloseHandler<? super T> handler : closeHandlers.values()) { - runCloseTask(executor, new CloseHandlerTask<T>(handler)); + runCloseTask(executor, new CloseHandlerTask(handler)); } } closeAction(); @@ -143,11 +143,11 @@ return key; } } - runCloseTask(executor, new CloseHandlerTask<T>(handler)); + runCloseTask(executor, new CloseHandlerTask(handler)); return new NullKey(); } - private static <T extends HandleableCloseable<T>> void runCloseTask(final Executor executor, final CloseHandlerTask<T> task) { + private static void runCloseTask(final Executor executor, final Runnable task) { try { executor.execute(task); } catch (RejectedExecutionException ree) { @@ -232,7 +232,7 @@ } } - private final class CloseHandlerTask<T extends HandleableCloseable<T>> implements Runnable { + private final class CloseHandlerTask implements Runnable { private final CloseHandler<? super T> handler; Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/InputStreamHandlerFactory.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/InputStreamHandlerFactory.java 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/InputStreamHandlerFactory.java 2009-10-01 20:56:19 UTC (rev 5553) @@ -27,8 +27,8 @@ import java.nio.ByteBuffer; import java.nio.channels.Channel; import org.jboss.xnio.IoFuture; -import org.jboss.xnio.IoHandler; import org.jboss.xnio.IoUtils; +import org.jboss.xnio.ChannelListener; import org.jboss.xnio.channels.StreamChannel; import org.jboss.xnio.log.Logger; @@ -56,26 +56,21 @@ streamContext = context; } - public IoHandler<StreamChannel> getLocalHandler() { - return new IoHandler<StreamChannel>() { - public void handleOpened(final StreamChannel channel) { + public ChannelListener<StreamChannel> getLocalHandler() { + return new ChannelListener<StreamChannel>() { + public void handleEvent(final StreamChannel channel) { + channel.getCloseSetter().set(new ChannelListener<StreamChannel>() { + public void handleEvent(final StreamChannel channel) { + IoUtils.safeClose(localInstance); + } + }); streamContext.execute(new LocalRunnable(channel, localInstance)); } - - public void handleClosed(final StreamChannel channel) { - IoUtils.safeClose(localInstance); - } - - public void handleReadable(final StreamChannel channel) { - } - - public void handleWritable(final StreamChannel channel) { - } }; } - public IoHandler<Channel> getRemoteHandler() { - return IoUtils.nullHandler(); + public ChannelListener<Channel> getRemoteHandler() { + return IoUtils.nullChannelListener(); } public InputStream getRemoteProxy(final IoFuture<? extends StreamChannel> futureChannel) { Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/OutputStreamHandlerFactory.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/OutputStreamHandlerFactory.java 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/OutputStreamHandlerFactory.java 2009-10-01 20:56:19 UTC (rev 5553) @@ -33,8 +33,8 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.jboss.xnio.IoFuture; -import org.jboss.xnio.IoHandler; -import static org.jboss.xnio.IoUtils.nullHandler; +import org.jboss.xnio.ChannelListener; +import org.jboss.xnio.IoUtils; import static org.jboss.xnio.IoUtils.safeClose; import org.jboss.xnio.channels.ChannelInputStream; import org.jboss.xnio.channels.StreamChannel; @@ -59,12 +59,12 @@ localInstance = instance; } - public IoHandler<StreamChannel> getLocalHandler() { + public ChannelListener<StreamChannel> getLocalHandler() { return new LocalHandler(localInstance); } - public IoHandler<Channel> getRemoteHandler() { - return nullHandler(); + public ChannelListener<Channel> getRemoteHandler() { + return IoUtils.nullChannelListener(); } public OutputStream getRemoteProxy(final IoFuture<? extends StreamChannel> futureChannel) { @@ -72,7 +72,7 @@ } } - private static final class LocalHandler implements IoHandler<StreamChannel> { + private static final class LocalHandler implements ChannelListener<StreamChannel> { private final OutputStream localInstance; private final byte[] bytes = new byte[1024]; @@ -80,33 +80,33 @@ localInstance = instance; } - public void handleOpened(final StreamChannel channel) { - } - - public void handleClosed(final StreamChannel channel) { - safeClose(localInstance); - } - - public void handleReadable(final StreamChannel channel) { - final byte[] bytes = this.bytes; - final ByteBuffer buffer = ByteBuffer.wrap(bytes); - try { - for (;;) { - final int res = channel.read(buffer); - if (res == 0) { - channel.resumeReads(); - return; + public void handleEvent(final StreamChannel channel) { + channel.getCloseSetter().set(new ChannelListener<StreamChannel>() { + public void handleEvent(final StreamChannel channel) { + safeClose(localInstance); + } + }); + channel.getReadSetter().set(new ChannelListener<StreamChannel>() { + public void handleEvent(final StreamChannel channel) { + final byte[] bytes = LocalHandler.this.bytes; + final ByteBuffer buffer = ByteBuffer.wrap(bytes); + try { + for (;;) { + final int res = channel.read(buffer); + if (res == 0) { + channel.resumeReads(); + return; + } + localInstance.write(bytes, 0, buffer.position()); + buffer.clear(); + } + } catch (IOException e) { + safeClose(channel); } - localInstance.write(bytes, 0, buffer.position()); - buffer.clear(); } - } catch (IOException e) { - safeClose(channel); - } + }); + channel.resumeReads(); } - - public void handleWritable(final StreamChannel channel) { - } } private static final class ProxyOutputStream extends OutputStream { Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/StreamHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/StreamHandler.java 2009-10-01 20:18:15 UTC (rev 5552) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/stream/StreamHandler.java 2009-10-01 20:56:19 UTC (rev 5553) @@ -22,8 +22,8 @@ package org.jboss.remoting3.stream; -import org.jboss.xnio.IoHandler; import org.jboss.xnio.IoFuture; +import org.jboss.xnio.ChannelListener; import java.nio.channels.Channel; import java.io.Serializable; @@ -39,20 +39,20 @@ public interface StreamHandler<T, C extends Channel> extends Serializable { /** - * Get the local XNIO handler for this stream. If this handler is cached on the object, it should be + * Get the local XNIO open handler for this stream. If this handler is cached on the object, it should be * done in a {@code transient} fashion to prevent the local handler from being sent to the remote side. * * @return the local XNIO handler */ - IoHandler<? super C> getLocalHandler(); + ChannelListener<? super C> getLocalHandler(); /** - * Get the remote XNIO handler for this stream. The remote handler should not be instantiated until the + * Get the remote XNIO open handler for this stream. The remote handler should not be instantiated until the * {@code StreamHandler} instance is on the remote side to avoid copying the handler across the wire. * * @return the remote XNIO handler */ - IoHandler<? super C> getRemoteHandler(); + ChannelListener<? super C> getRemoteHandler(); /** * Get the remote proxy instance for this stream. The remote proxy should not be instantiated until the @@ -62,5 +62,5 @@ * @param futureChannel the future channel * @return the remote proxy instance */ - T getRemoteProxy(final IoFuture<? extends C> futureChannel); + T getRemoteProxy(IoFuture<? extends C> futureChannel); }