Author: david.lloyd(a)jboss.com Date: 2009-02-26 16:52:06 -0500 (Thu, 26 Feb 2009) New Revision: 4859 Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Connection.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionClient.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionListener.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServer.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceListener.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceProvider.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Multiplex.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ServiceConfiguration.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SynchronizedSet.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/impl/ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/impl/AbstractConnectionServiceProvider.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/SAClassTable.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisement.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementReply.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementRequestListener.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/protocol/ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/protocol/ Removed: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/protocol/ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/protocol/multiplex/ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting/ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/protocol/ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/protocol/multiplex/ Log: Incremental - still more work to go before this will compile Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3 (from rev 4848, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting) Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex (from rev 4848, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex) Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/BufferByteOutput.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,90 +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 org.jboss.marshalling.ByteOutput; -import org.jboss.xnio.BufferAllocator; -import java.nio.ByteBuffer; -import java.util.Collection; - -/** - * - */ -public class BufferByteOutput implements ByteOutput { - - private ByteBuffer current; - private final BufferAllocator<ByteBuffer> allocator; - private final Collection<ByteBuffer> target; - - public BufferByteOutput(final BufferAllocator<ByteBuffer> allocator, final Collection<ByteBuffer> target) { - this.allocator = allocator; - this.target = target; - } - - private ByteBuffer getCurrent() { - final ByteBuffer buffer = current; - return buffer == null ? (current = allocator.allocate()) : buffer; - } - - public void write(final int i) { - final ByteBuffer buffer = getCurrent(); - buffer.put((byte) i); - if (! buffer.hasRemaining()) { - buffer.flip(); - target.add(buffer); - current = null; - } - } - - public void write(final byte[] bytes) { - write(bytes, 0, bytes.length); - } - - public void write(final byte[] bytes, int offs, int len) { - while (len > 0) { - final ByteBuffer buffer = getCurrent(); - final int c = Math.min(len, buffer.remaining()); - buffer.put(bytes, offs, c); - offs += c; - len -= c; - if (! buffer.hasRemaining()) { - buffer.flip(); - target.add(buffer); - current = null; - } - } - } - - public void close() { - flush(); - } - - public void flush() { - final ByteBuffer buffer = current; - if (buffer != null) { - buffer.flip(); - target.add(buffer); - current = null; - } - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/BufferByteOutput.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/BufferByteOutput.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,90 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.marshalling.ByteOutput; +import org.jboss.xnio.BufferAllocator; +import java.nio.ByteBuffer; +import java.util.Collection; + +/** + * + */ +public class BufferByteOutput implements ByteOutput { + + private ByteBuffer current; + private final BufferAllocator<ByteBuffer> allocator; + private final Collection<ByteBuffer> target; + + public BufferByteOutput(final BufferAllocator<ByteBuffer> allocator, final Collection<ByteBuffer> target) { + this.allocator = allocator; + this.target = target; + } + + private ByteBuffer getCurrent() { + final ByteBuffer buffer = current; + return buffer == null ? (current = allocator.allocate()) : buffer; + } + + public void write(final int i) { + final ByteBuffer buffer = getCurrent(); + buffer.put((byte) i); + if (! buffer.hasRemaining()) { + buffer.flip(); + target.add(buffer); + current = null; + } + } + + public void write(final byte[] bytes) { + write(bytes, 0, bytes.length); + } + + public void write(final byte[] bytes, int offs, int len) { + while (len > 0) { + final ByteBuffer buffer = getCurrent(); + final int c = Math.min(len, buffer.remaining()); + buffer.put(bytes, offs, c); + offs += c; + len -= c; + if (! buffer.hasRemaining()) { + buffer.flip(); + target.add(buffer); + current = null; + } + } + } + + public void close() { + flush(); + } + + public void flush() { + final ByteBuffer buffer = current; + if (buffer != null) { + buffer.flip(); + target.add(buffer); + current = null; + } + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Connection.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Connection.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Connection.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,44 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.spi.Handle; +import org.jboss.remoting3.spi.AutoCloseable; +import org.jboss.remoting3.QualifiedName; +import org.jboss.xnio.IoFuture; + +/** + * A connection with a remote system. + */ +public interface Connection extends AutoCloseable<Connection>, ConnectionServiceProvider { + + /** + * Directly access a remote service on this connection. + * + * @param path the path of the service to access + * @param config the configuration to use + * @return a future handle to a request handler source that accesses the remote service + */ + IoFuture<Handle<RequestHandlerSource>> accessRemoteService(QualifiedName path, ServiceConfiguration config); +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionClient.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionClient.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionClient.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,34 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.ChannelSource; +import org.jboss.xnio.channels.AllocatedMessageChannel; + +/** + * + */ +public interface ConnectionClient extends ConnectionServiceProvider { + IoFuture<Connection> connect(ChannelSource<AllocatedMessageChannel> channelSource); +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionListener.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionListener.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionListener.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,30 @@ +/* + * 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.remoting3.multiplex; + +/** + * + */ +public interface ConnectionListener { + void handleConnection(Connection connection); +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServer.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServer.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServer.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,34 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.spi.AutoCloseable; +import org.jboss.xnio.IoHandlerFactory; +import org.jboss.xnio.channels.AllocatedMessageChannel; + +/** + * + */ +public interface ConnectionServer extends IoHandlerFactory<AllocatedMessageChannel>, ConnectionServiceProvider, AutoCloseable<ConnectionServer> { + void addConnectionListener(ConnectionListener connectionListener); +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceListener.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceListener.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceListener.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,96 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.QualifiedName; +import org.jboss.remoting3.SimpleCloseable; +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.spi.Handle; + +/** + * + */ +public interface ConnectionServiceListener { + void serviceRegistered(SimpleCloseable listenerHandler, ServiceInfo info); + + final class ServiceInfo { + + private String endpointName; + private String serviceType; + private String groupName; + private QualifiedName path; + private int metric; + private Handle<RequestHandlerSource> requestHandlerSource; + + public ServiceInfo() { + } + + public String getEndpointName() { + return endpointName; + } + + public void setEndpointName(final String endpointName) { + this.endpointName = endpointName; + } + + 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 QualifiedName getPath() { + return path; + } + + public void setPath(final QualifiedName path) { + this.path = path; + } + + public int getMetric() { + return metric; + } + + public void setMetric(final int metric) { + this.metric = metric; + } + + public Handle<RequestHandlerSource> getRequestHandlerSource() { + return requestHandlerSource; + } + + public void setRequestHandlerSource(final Handle<RequestHandlerSource> requestHandlerSource) { + this.requestHandlerSource = requestHandlerSource; + } + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceProvider.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceProvider.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ConnectionServiceProvider.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,65 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.QualifiedName; +import org.jboss.remoting3.RemoteServiceConfiguration; +import org.jboss.remoting3.SimpleCloseable; +import org.jboss.remoting3.spi.Handle; +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.marshalling.MarshallerFactory; +import org.jboss.marshalling.MarshallingConfiguration; +import java.io.IOException; +import java.io.Closeable; + +/** + * A connection service provider. Implementations of this interface allow services to be registered at + * named locations for remote access. + */ +public interface ConnectionServiceProvider extends Closeable { + + void registerMarshaller(String name, MarshallerFactory marshallerFactory, MarshallingConfiguration configuration); + + void unregisterMarshaller(String name); + + /** + * Register a service on this connection. When the given handle is closed, the service is unregistered. A + * strong reference is maintained to the given handle. The handle will be closed automatically if this provider + * is closed. + * + * @param path the path at which the service should be registered + * @param remoteServiceConfiguration the remote service configuration of the service + * @param serviceConfiguration the marshalling setup to use + * @throws java.io.IOException if the service could not be registered + */ + Handle<RequestHandlerSource> registerService(QualifiedName path, RemoteServiceConfiguration remoteServiceConfiguration, ServiceConfiguration serviceConfiguration) throws IOException; + + /** + * Add a connection service listener which is notified whenever a service is registered to this provider. The + * returned handle may be used to unregister the listener. + * + * @param listener the listener + * @return a handle which can be used to unregister the listener + */ + SimpleCloseable addServiceListener(ConnectionServiceListener listener); +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/FutureRemoteRequestHandlerSource.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,50 +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 org.jboss.xnio.AbstractIoFuture; -import org.jboss.xnio.IoFuture; -import org.jboss.remoting.spi.RequestHandlerSource; -import java.io.IOException; - -/** - * - */ -public final class FutureRemoteRequestHandlerSource extends AbstractIoFuture<RequestHandlerSource> { - - public IoFuture<RequestHandlerSource> cancel() { - return this; - } - - protected boolean setException(final IOException exception) { - return super.setException(exception); - } - - protected boolean setResult(final RequestHandlerSource result) { - return super.setResult(result); - } - - public String toString() { - return "future remote request handler source <" + Integer.toHexString(hashCode()) + ">"; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/FutureRemoteRequestHandlerSource.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/FutureRemoteRequestHandlerSource.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,50 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.xnio.AbstractIoFuture; +import org.jboss.xnio.IoFuture; +import org.jboss.remoting3.spi.RequestHandlerSource; +import java.io.IOException; + +/** + * + */ +public final class FutureRemoteRequestHandlerSource extends AbstractIoFuture<RequestHandlerSource> { + + public IoFuture<RequestHandlerSource> cancel() { + return this; + } + + protected boolean setException(final IOException exception) { + return super.setException(exception); + } + + protected boolean setResult(final RequestHandlerSource result) { + return super.setResult(result); + } + + public String toString() { + return "future remote request handler source <" + Integer.toHexString(hashCode()) + ">"; + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerBiMap.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,96 +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.HashMap; -import java.util.IdentityHashMap; -import java.util.Set; -import java.util.Collections; - -/** - * - */ -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 boolean putIfAbsent(final int key1, final T key2) { - final Integer key1Obj = Integer.valueOf(key1); - if (leftMap.containsKey(key1Obj)) { - return false; - } - final T oldKey2 = leftMap.put(key1Obj, key2); - rightMap.put(key2, key1Obj); - rightMap.remove(oldKey2); - return true; - } - - 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 Set<T> getKeys() { - return Collections.unmodifiableSet(rightMap.keySet()); - } - - public static <T> IntegerBiMap<T> create() { - return new IdentityHashIntegerBiMap<T>(); - } - - public static <T> IntegerBiMap<T> createSynchronizing() { - return IntegerBiMap.Util.synchronizing(new IdentityHashIntegerBiMap<T>()); - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerBiMap.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,96 @@ +/* + * 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.remoting3.multiplex; + +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.Set; +import java.util.Collections; + +/** + * + */ +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 boolean putIfAbsent(final int key1, final T key2) { + final Integer key1Obj = Integer.valueOf(key1); + if (leftMap.containsKey(key1Obj)) { + return false; + } + final T oldKey2 = leftMap.put(key1Obj, key2); + rightMap.put(key2, key1Obj); + rightMap.remove(oldKey2); + return true; + } + + 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 Set<T> getKeys() { + return Collections.unmodifiableSet(rightMap.keySet()); + } + + public static <T> IntegerBiMap<T> create() { + return new IdentityHashIntegerBiMap<T>(); + } + + public static <T> IntegerBiMap<T> createSynchronizing() { + return IntegerBiMap.Util.synchronizing(new IdentityHashIntegerBiMap<T>()); + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerResourceBiMap.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,110 +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.HashMap; -import java.util.IdentityHashMap; -import java.util.Iterator; -import java.util.Map; -import java.util.Collections; -import java.util.Collection; -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 Collection<Handle<T>> getKeys() { - return Collections.unmodifiableCollection(leftMap.values()); - } - - 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 Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IdentityHashIntegerResourceBiMap.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IdentityHashIntegerResourceBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,110 @@ +/* + * 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.remoting3.multiplex; + +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Collections; +import java.util.Collection; +import org.jboss.remoting3.spi.AutoCloseable; +import org.jboss.remoting3.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 Collection<Handle<T>> getKeys() { + return Collections.unmodifiableCollection(leftMap.values()); + } + + 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-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerBiMap.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,106 +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 org.jboss.remoting.core.util.SynchronizedSet; -import java.util.Set; - -/** - * - */ -interface IntegerBiMap<T> { - int get(T key, int defValue); - - T get(int key); - - void put(int key1, T key2); - - boolean putIfAbsent(int key1, T key2); - - T remove(int key); - - void remove(T key); - - Set<T> getKeys(); - - 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 boolean putIfAbsent(final int key1, final T key2) { - synchronized (lock) { - return orig.putIfAbsent(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 Set<T> getKeys() { - return new SynchronizedSet<T>(orig.getKeys(), lock); - } - } - - public static <T> IntegerBiMap<T> synchronizing(IntegerBiMap<T> orig) { - return new SyncWrapper<T>(orig, new Object()); - } - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerBiMap.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,105 @@ +/* + * 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.remoting3.multiplex; + +import java.util.Set; + +/** + * + */ +interface IntegerBiMap<T> { + int get(T key, int defValue); + + T get(int key); + + void put(int key1, T key2); + + boolean putIfAbsent(int key1, T key2); + + T remove(int key); + + void remove(T key); + + Set<T> getKeys(); + + 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 boolean putIfAbsent(final int key1, final T key2) { + synchronized (lock) { + return orig.putIfAbsent(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 Set<T> getKeys() { + return new SynchronizedSet<T>(orig.getKeys(), lock); + } + } + + public static <T> IntegerBiMap<T> synchronizing(IntegerBiMap<T> orig) { + return new SyncWrapper<T>(orig, new Object()); + } + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerResourceBiMap.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,105 +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 org.jboss.remoting.spi.AutoCloseable; -import org.jboss.remoting.spi.Handle; -import org.jboss.remoting.core.util.SynchronizedCollection; -import java.util.Iterator; -import java.util.Collection; - -/** - * - */ -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); - - Collection<Handle<T>> getKeys(); - - 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 Collection<Handle<T>> getKeys() { - return new SynchronizedCollection<Handle<T>>(orig.getKeys(), lock); - } - - 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()); - } - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/IntegerResourceBiMap.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/IntegerResourceBiMap.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,105 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.spi.AutoCloseable; +import org.jboss.remoting3.spi.Handle; +import org.jboss.remoting3.core.util.SynchronizedCollection; +import java.util.Iterator; +import java.util.Collection; + +/** + * + */ +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); + + Collection<Handle<T>> getKeys(); + + 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 Collection<Handle<T>> getKeys() { + return new SynchronizedCollection<Handle<T>>(orig.getKeys(), lock); + } + + 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()); + } + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MessageType.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,113 +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; - -/** - * The type of a protocol message. - */ -enum MessageType { - - /** - * Signals that the connection should be closed in an orderly fashion. After this message is sent, no further - * requests or service advertisements may be sent. - */ - CONNECTION_CLOSE(0x00), - /** - * The request part of a request-response sequence, sent from the Client to the RequestListener. - */ - REQUEST(0x10), - /** - * The reply part of a request-response sequence, sent from the RequestListener to the Client. - */ - REPLY(0x11), - /** - * A cancellation request for an outstanding request, sent from the Client to the RequestListener. - */ - CANCEL_REQUEST(0x12), - /** - * Acknowlegement that a request was cancelled, sent from the RequestListener to the Client. - */ - CANCEL_ACK(0x13), - /** - * Message that the request could not be received on the remote end, sent from to the Client from the - * protocol handler. - */ - REQUEST_RECEIVE_FAILED(0x14), - // Request failed due to exception - REQUEST_FAILED(0x15), - // Remote side called .close() on a forwarded RequestHandler - CLIENT_CLOSE(0x20), - // Remote side pulled a new RequestHandler off of a forwarded RequestHandlerSource - CLIENT_OPEN(0x21), - // Request to open a service at a path - SERVICE_OPEN_REQUEST(0x30), - // Reply for a successful service open - SERVICE_OPEN_REPLY(0x31), - // Reply for a generally failed service open - SERVICE_OPEN_FAILED(0x32), - SERVICE_OPEN_NOT_FOUND(0x33), - SERVICE_OPEN_FORBIDDEN(0x34), - - // Notify the remote side that the service will no longer be used - SERVICE_CLOSE_REQUEST(0x3e), - // The service channel is closed; no further clients may be opened - SERVICE_CLOSE_NOTIFY(0x3f), - ; - private final int id; - - private MessageType(int id) { - this.id = id; - } - - public int getId() { - return id; - } - - /** - * Get the message type for an integer ID. - * - * @param id the integer ID - * @return the message type instance - */ - public static MessageType getMessageType(final int id) { - switch (id) { - case 0x00: return CONNECTION_CLOSE; - case 0x10: return REQUEST; - case 0x11: return REPLY; - case 0x12: return CANCEL_REQUEST; - case 0x13: return CANCEL_ACK; - case 0x14: return REQUEST_RECEIVE_FAILED; - case 0x15: return REQUEST_FAILED; - case 0x20: return CLIENT_CLOSE; - case 0x21: return CLIENT_OPEN; - case 0x30: return SERVICE_OPEN_REQUEST; - case 0x31: return SERVICE_OPEN_REPLY; - case 0x32: return SERVICE_OPEN_FAILED; - case 0x33: return SERVICE_OPEN_NOT_FOUND; - case 0x34: return SERVICE_OPEN_FORBIDDEN; - case 0x3e: return SERVICE_CLOSE_REQUEST; - case 0x3f: return SERVICE_CLOSE_NOTIFY; - default: throw new IllegalArgumentException("Invalid message type ID"); - } - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MessageType.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MessageType.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,113 @@ +/* + * 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.remoting3.multiplex; + +/** + * The type of a protocol message. + */ +enum MessageType { + + /** + * Signals that the connection should be closed in an orderly fashion. After this message is sent, no further + * requests or service advertisements may be sent. + */ + CONNECTION_CLOSE(0x00), + /** + * The request part of a request-response sequence, sent from the Client to the RequestListener. + */ + REQUEST(0x10), + /** + * The reply part of a request-response sequence, sent from the RequestListener to the Client. + */ + REPLY(0x11), + /** + * A cancellation request for an outstanding request, sent from the Client to the RequestListener. + */ + CANCEL_REQUEST(0x12), + /** + * Acknowlegement that a request was cancelled, sent from the RequestListener to the Client. + */ + CANCEL_ACK(0x13), + /** + * Message that the request could not be received on the remote end, sent from to the Client from the + * protocol handler. + */ + REQUEST_RECEIVE_FAILED(0x14), + // Request failed due to exception + REQUEST_FAILED(0x15), + // Remote side called .close() on a forwarded RequestHandler + CLIENT_CLOSE(0x20), + // Remote side pulled a new RequestHandler off of a forwarded RequestHandlerSource + CLIENT_OPEN(0x21), + // Request to open a service at a path + SERVICE_OPEN_REQUEST(0x30), + // Reply for a successful service open + SERVICE_OPEN_REPLY(0x31), + // Reply for a generally failed service open + SERVICE_OPEN_FAILED(0x32), + SERVICE_OPEN_NOT_FOUND(0x33), + SERVICE_OPEN_FORBIDDEN(0x34), + + // Notify the remote side that the service will no longer be used + SERVICE_CLOSE_REQUEST(0x3e), + // The service channel is closed; no further clients may be opened + SERVICE_CLOSE_NOTIFY(0x3f), + ; + private final int id; + + private MessageType(int id) { + this.id = id; + } + + public int getId() { + return id; + } + + /** + * Get the message type for an integer ID. + * + * @param id the integer ID + * @return the message type instance + */ + public static MessageType getMessageType(final int id) { + switch (id) { + case 0x00: return CONNECTION_CLOSE; + case 0x10: return REQUEST; + case 0x11: return REPLY; + case 0x12: return CANCEL_REQUEST; + case 0x13: return CANCEL_ACK; + case 0x14: return REQUEST_RECEIVE_FAILED; + case 0x15: return REQUEST_FAILED; + case 0x20: return CLIENT_CLOSE; + case 0x21: return CLIENT_OPEN; + case 0x30: return SERVICE_OPEN_REQUEST; + case 0x31: return SERVICE_OPEN_REPLY; + case 0x32: return SERVICE_OPEN_FAILED; + case 0x33: return SERVICE_OPEN_NOT_FOUND; + case 0x34: return SERVICE_OPEN_FORBIDDEN; + case 0x3e: return SERVICE_CLOSE_REQUEST; + case 0x3f: return SERVICE_CLOSE_NOTIFY; + default: throw new IllegalArgumentException("Invalid message type ID"); + } + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Multiplex.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Multiplex.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/Multiplex.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,39 @@ +/* + * 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.remoting3.multiplex; + +/** + * + */ +public final class Multiplex { + private Multiplex() {} + + /** + * Create a new multiplex connector. + * + * @return the connector + */ + public static ConnectionClient createConnector() { + + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConfiguration.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,156 +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.concurrent.Executor; -import java.nio.ByteBuffer; -import org.jboss.xnio.BufferAllocator; -import org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.MarshallingConfiguration; -import org.jboss.remoting.spi.NamedServiceRegistry; - -/** - * A configuration object for the multiplex protocol. - */ -public final class MultiplexConfiguration { - private MarshallerFactory marshallerFactory; - private MarshallingConfiguration marshallingConfiguration; - private int linkMetric; - private Executor executor; - private BufferAllocator<ByteBuffer> allocator; - private NamedServiceRegistry namedServiceRegistry; - - /** - * Construct a new instance. - */ - public MultiplexConfiguration() { - } - - /** - * Get the marshaller factory to use for messages transmitted and received by this multiplex connection. - * - * @return the marshaller factory - */ - public MarshallerFactory getMarshallerFactory() { - return marshallerFactory; - } - - /** - * Set the marshaller factory to use for messages transmitted and received by this multiplex connection. - * - * @param marshallerFactory the marshaller factory - */ - public void setMarshallerFactory(final MarshallerFactory marshallerFactory) { - this.marshallerFactory = marshallerFactory; - } - - /** - * Get the marshaller configuration to pass into the marshaller factory. - * - * @return the configuration - */ - public MarshallingConfiguration getMarshallingConfiguration() { - return marshallingConfiguration; - } - - /** - * Set the marshaller configuration to pass into the marshaller factory. - * - * @param marshallingConfiguration the configuration - */ - public void setMarshallingConfiguration(final MarshallingConfiguration marshallingConfiguration) { - this.marshallingConfiguration = marshallingConfiguration; - } - - /** - * Get the link metric to assign to this multiplex connection. - * - * @return the link metric - */ - public int getLinkMetric() { - return linkMetric; - } - - /** - * Set the link metric to assign to this multiplex connection. - * - * @param linkMetric the link metric - */ - public void setLinkMetric(final int linkMetric) { - this.linkMetric = linkMetric; - } - - /** - * Get the executor to use to execute handlers. - * - * @return the executor - */ - public Executor getExecutor() { - return executor; - } - - /** - * Set the executor to use to execute handlers. - * - * @param executor the executor - */ - public void setExecutor(final Executor executor) { - this.executor = executor; - } - - /** - * Get the buffer allocator to use for outbound messages on this connection. - * - * @return the allocator - */ - public BufferAllocator<ByteBuffer> getAllocator() { - return allocator; - } - - /** - * Set the buffer allocator to use for outbound messages on this connection. - * - * @param allocator the allocator - */ - public void setAllocator(final BufferAllocator<ByteBuffer> allocator) { - this.allocator = allocator; - } - - /** - * Get the named service registry for this connection. - * - * @return the registry - */ - public NamedServiceRegistry getNamedServiceRegistry() { - return namedServiceRegistry; - } - - /** - * Set the named service registry for this connection. - * - * @param namedServiceRegistry the registry - */ - public void setNamedServiceRegistry(final NamedServiceRegistry namedServiceRegistry) { - this.namedServiceRegistry = namedServiceRegistry; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConfiguration.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConfiguration.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,156 @@ +/* + * 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.remoting3.multiplex; + +import java.util.concurrent.Executor; +import java.nio.ByteBuffer; +import org.jboss.xnio.BufferAllocator; +import org.jboss.marshalling.MarshallerFactory; +import org.jboss.marshalling.MarshallingConfiguration; +import org.jboss.remoting3.spi.NamedServiceRegistry; + +/** + * A configuration object for the multiplex protocol. + */ +public final class MultiplexConfiguration { + private MarshallerFactory marshallerFactory; + private MarshallingConfiguration marshallingConfiguration; + private int linkMetric; + private Executor executor; + private BufferAllocator<ByteBuffer> allocator; + private NamedServiceRegistry namedServiceRegistry; + + /** + * Construct a new instance. + */ + public MultiplexConfiguration() { + } + + /** + * Get the marshaller factory to use for messages transmitted and received by this multiplex connection. + * + * @return the marshaller factory + */ + public MarshallerFactory getMarshallerFactory() { + return marshallerFactory; + } + + /** + * Set the marshaller factory to use for messages transmitted and received by this multiplex connection. + * + * @param marshallerFactory the marshaller factory + */ + public void setMarshallerFactory(final MarshallerFactory marshallerFactory) { + this.marshallerFactory = marshallerFactory; + } + + /** + * Get the marshaller configuration to pass into the marshaller factory. + * + * @return the configuration + */ + public MarshallingConfiguration getMarshallingConfiguration() { + return marshallingConfiguration; + } + + /** + * Set the marshaller configuration to pass into the marshaller factory. + * + * @param marshallingConfiguration the configuration + */ + public void setMarshallingConfiguration(final MarshallingConfiguration marshallingConfiguration) { + this.marshallingConfiguration = marshallingConfiguration; + } + + /** + * Get the link metric to assign to this multiplex connection. + * + * @return the link metric + */ + public int getLinkMetric() { + return linkMetric; + } + + /** + * Set the link metric to assign to this multiplex connection. + * + * @param linkMetric the link metric + */ + public void setLinkMetric(final int linkMetric) { + this.linkMetric = linkMetric; + } + + /** + * Get the executor to use to execute handlers. + * + * @return the executor + */ + public Executor getExecutor() { + return executor; + } + + /** + * Set the executor to use to execute handlers. + * + * @param executor the executor + */ + public void setExecutor(final Executor executor) { + this.executor = executor; + } + + /** + * Get the buffer allocator to use for outbound messages on this connection. + * + * @return the allocator + */ + public BufferAllocator<ByteBuffer> getAllocator() { + return allocator; + } + + /** + * Set the buffer allocator to use for outbound messages on this connection. + * + * @param allocator the allocator + */ + public void setAllocator(final BufferAllocator<ByteBuffer> allocator) { + this.allocator = allocator; + } + + /** + * Get the named service registry for this connection. + * + * @return the registry + */ + public NamedServiceRegistry getNamedServiceRegistry() { + return namedServiceRegistry; + } + + /** + * Set the named service registry for this connection. + * + * @param namedServiceRegistry the registry + */ + public void setNamedServiceRegistry(final NamedServiceRegistry namedServiceRegistry) { + this.namedServiceRegistry = namedServiceRegistry; + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConnection.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,455 +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 org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.MarshallingConfiguration; -import org.jboss.xnio.BufferAllocator; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.Buffers; -import org.jboss.xnio.log.Logger; -import org.jboss.xnio.channels.AllocatedMessageChannel; -import org.jboss.remoting.spi.ReplyHandler; -import org.jboss.remoting.spi.RemoteRequestContext; -import org.jboss.remoting.spi.RequestHandler; -import org.jboss.remoting.spi.RequestHandlerSource; -import org.jboss.remoting.spi.Handle; -import org.jboss.remoting.spi.NamedServiceRegistry; -import org.jboss.remoting.spi.SpiUtils; -import org.jboss.remoting.spi.AbstractHandleableCloseable; -import org.jboss.remoting.QualifiedName; -import org.jboss.remoting.Endpoint; -import org.jboss.remoting.IndeterminateOutcomeException; -import java.util.concurrent.Executor; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.List; -import java.nio.ByteBuffer; -import java.io.IOException; -import java.io.InterruptedIOException; - -/** - * - */ -public final class MultiplexConnection extends AbstractHandleableCloseable<MultiplexConnection> { - private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex"); - - //--== Connection configuration items ==-- - private final MarshallerFactory marshallerFactory; - 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 IntegerBiMap<ReplyHandler> remoteRequests = IdentityHashIntegerBiMap.createSynchronizing(); - // running on local node - 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 IntegerBiMap<RequestHandler> remoteClients = IdentityHashIntegerBiMap.createSynchronizing(); - // forwarded to remote side (handled on this side) - 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 (shift left one bit, limit is 2^30) - private final AtomicInteger remoteClientSequence = new AtomicInteger(); - - // services on the remote side - private final IntegerBiMap<FutureRemoteRequestHandlerSource> remoteServices = IdentityHashIntegerBiMap.createSynchronizing(); - // forwarded to remote side (handled on this side) - private final IntegerResourceBiMap<RequestHandlerSource> forwardedServices = IdentityHashIntegerResourceBiMap.createSynchronizing(); - // sequence for remote services - private final AtomicInteger remoteServiceSequence = new AtomicInteger(); - - // registered services by path - private final NamedServiceRegistry namedServiceRegistry; - - private final Endpoint endpoint; - - private final AllocatedMessageChannel channel; - - public MultiplexConnection(final Endpoint endpoint, final AllocatedMessageChannel channel, final MultiplexConfiguration configuration) { - super(configuration.getExecutor()); - this.endpoint = endpoint; - this.channel = channel; - marshallerFactory = configuration.getMarshallerFactory(); - if (marshallerFactory == null) { - throw new NullPointerException("marshallerFactory is null"); - } - marshallingConfiguration = configuration.getMarshallingConfiguration(); - if (marshallingConfiguration == null) { - throw new NullPointerException("marshallingConfiguration is null"); - } - linkMetric = configuration.getLinkMetric(); - executor = configuration.getExecutor(); - if (executor == null) { - throw new NullPointerException("executor is null"); - } - allocator = configuration.getAllocator(); - if (allocator == null) { - throw new NullPointerException("allocator is null"); - } - namedServiceRegistry = configuration.getNamedServiceRegistry(); - if (namedServiceRegistry == null) { - throw new NullPointerException("namedServiceRegistry is null"); - } - } - - // sequence methods - - int nextRequest() { - return requestSequence.getAndIncrement() & 0x7fffffff; - } - - int nextForwardedClient() { - return (forwardedClientSequence.getAndIncrement() << 1 | 1) & 0x7fffffff; - } - - int nextRemoteClient() { - return remoteClientSequence.getAndIncrement() << 1 & 0x7fffffff; - } - - int nextRemoteService() { - return remoteServiceSequence.getAndIncrement() & 0x7fffffff; - } - - void doBlockingWrite(ByteBuffer... buffers) throws IOException { - log.trace("Sending message:\n%s", new MultiDumper(buffers)); - if (buffers.length == 1) doBlockingWrite(buffers[0]); else for (;;) { - if (channel.send(buffers)) { - return; - } - channel.awaitWritable(); - } - } - - private static final class MultiDumper { - private final ByteBuffer[] buffers; - - public MultiDumper(final ByteBuffer[] buffers) { - this.buffers = buffers; - } - - public String toString() { - StringBuilder builder = new StringBuilder(); - for (int i = 0; i < buffers.length; i++) { - ByteBuffer buffer = buffers[i]; - builder.append("Buffer "); - builder.append(i); - builder.append(":\n"); - try { - Buffers.dump(buffer, builder, 8, 1); - } catch (IOException e) { - // ignore - } - } - return builder.toString(); - } - } - - void doBlockingWrite(ByteBuffer buffer) throws IOException { - log.trace("Sending message:\n%s", Buffers.createDumper(buffer, 8, 1)); - for (;;) { - if (channel.send(buffer)) { - return; - } - channel.awaitWritable(); - } - } - - void doBlockingWrite(List<ByteBuffer> buffers) throws IOException { - doBlockingWrite(buffers.toArray(new ByteBuffer[buffers.size()])); - } - - MarshallerFactory getMarshallerFactory() { - return marshallerFactory; - } - - MarshallingConfiguration getMarshallingConfiguration() { - return marshallingConfiguration; - } - - int getLinkMetric() { - return linkMetric; - } - - protected Executor getExecutor() { - return executor; - } - - BufferAllocator<ByteBuffer> getAllocator() { - return allocator; - } - - Endpoint getEndpoint() { - return endpoint; - } - - AllocatedMessageChannel getChannel() { - return channel; - } - - void removeRemoteClient(final int identifier) { - remoteClients.remove(identifier); - } - - void addRemoteRequest(final int id, final ReplyHandler handler) { - remoteRequests.put(id, handler); - } - - void addRemoteClient(final int id, final RequestHandler handler) { - remoteClients.put(id, handler); - } - - Handle<RequestHandler> getForwardedClient(final int id) { - return forwardedClients.get(id); - } - - ReplyHandler removeRemoteRequest(final int id) { - return remoteRequests.remove(id); - } - - RemoteRequestContext getLocalRequest(final int id) { - return localRequests.get(id); - } - - ReplyHandler getRemoteRequest(final int id) { - return remoteRequests.get(id); - } - - Handle<RequestHandler> removeForwardedClient(final int id) { - return forwardedClients.remove(id); - } - - Handle<RequestHandlerSource> getForwardedService(final int id) { - return forwardedServices.get(id); - } - - void addForwardedClient(final int id, final Handle<RequestHandler> handle) { - forwardedClients.put(id, handle); - } - - void addForwadedService(final int id, final Handle<RequestHandlerSource> service) { - forwardedServices.put(id, service); - } - - Handle<RequestHandlerSource> removeForwardedService(final int id) { - return forwardedServices.remove(id); - } - - Handle<RequestHandlerSource> getServiceByPath(String path) { - return getService(QualifiedName.parse(path)); - } - - Handle<RequestHandlerSource> getService(final QualifiedName name) { - return namedServiceRegistry.lookupService(name); - } - - FutureRemoteRequestHandlerSource getFutureRemoteService(final int id) { - return remoteServices.get(id); - } - - FutureRemoteRequestHandlerSource removeFutureRemoteService(final int id) { - return remoteServices.remove(id); - } - - public Handle<RequestHandlerSource> openRemoteService(final QualifiedName name) throws IOException { - log.trace("Sending request to open remote service \"%s\"", name); - final FutureRemoteRequestHandlerSource future = new FutureRemoteRequestHandlerSource(); - int id; - for (;;) { - id = nextRemoteService(); - if (remoteServices.putIfAbsent(id, future)) { - break; - } - } - ByteBuffer buffer = ByteBuffer.allocate(5 + getByteLength(name)); - buffer.put((byte) MessageType.SERVICE_OPEN_REQUEST.getId()); - buffer.putInt(id); - putQualifiedName(buffer, name); - buffer.flip(); - doBlockingWrite(buffer); - try { - final Handle<RequestHandlerSource> handle = future.getInterruptibly().getHandle(); - log.trace("Opened %s", handle); - return handle; - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new InterruptedIOException("Interrupted while waiting for remote service"); - } - } - - static int getByteLength(QualifiedName name) { - int cnt = 2; // short header - for (String s : name) { - cnt += getByteLength(s); - } - return cnt; - } - - static int getByteLength(String s) { - final int len = s.length(); - int cnt = 0; - for (int i = 0; i < len; i++) { - char ch = s.charAt(i); - if (ch > 0 && ch <= 0x7f) { - cnt ++; - } else if (ch <= 0x07ff) { - cnt += 2; - } else { - cnt += 3; - } - } - // null terminator... - cnt ++; - return cnt; - } - - static String getString(final ByteBuffer buffer) { - StringBuilder builder = new StringBuilder(); - int state = 0, a = 0; - while (buffer.hasRemaining()) { - final int v = buffer.get() & 0xff; - switch (state) { - case 0: { - if (v == 0) { - return builder.toString(); - } else if (v < 128) { - builder.append((char) v); - } else if (192 <= v && v < 224) { - a = v << 6; - state = 1; - } else if (224 <= v && v < 232) { - a = v << 12; - state = 2; - } else { - builder.append('?'); - } - break; - } - case 1: { - if (v == 0) { - builder.append('?'); - return builder.toString(); - } else if (128 <= v && v < 192) { - a |= v & 0x3f; - builder.append((char) a); - } else { - builder.append('?'); - } - state = 0; - break; - } - case 2: { - if (v == 0) { - builder.append('?'); - return builder.toString(); - } else if (128 <= v && v < 192) { - a |= (v & 0x3f) << 6; - state = 1; - } else { - builder.append('?'); - state = 0; - } - break; - } - default: - throw new IllegalStateException("wrong state"); - } - } - return builder.toString(); - } - - static void putString(final ByteBuffer buffer, final String string) { - final int len = string.length(); - for (int i = 0; i < len; i ++) { - char ch = string.charAt(i); - if (ch > 0 && ch <= 0x7f) { - buffer.put((byte) ch); - } else if (ch <= 0x07ff) { - buffer.put((byte) (0xc0 | 0x1f & ch >> 6)); - buffer.put((byte) (0x80 | 0x3f & ch)); - } else { - buffer.put((byte) (0xe0 | 0x0f & ch >> 12)); - buffer.put((byte) (0x80 | 0x3f & ch >> 6)); - buffer.put((byte) (0x80 | 0x3f & ch)); - } - } - buffer.put((byte) 0); - } - - static QualifiedName getQualifiedName(final ByteBuffer buffer) { - final int len = buffer.getShort() & 0xffff; - final String[] segs = new String[len]; - for (int i = 0; i < len; i++) { - segs[i] = getString(buffer); - } - return new QualifiedName(segs); - } - - static void putQualifiedName(final ByteBuffer buffer, final QualifiedName qualifiedName) { - final int len = qualifiedName.length(); - if (len > 0xffff) { - throw new IllegalArgumentException("Qualified name is too long"); - } - buffer.putShort((short) len); - for (String seg : qualifiedName) { - putString(buffer, seg); - } - } - - protected void closeAction() { - // just to make sure... - IoUtils.safeClose(channel); - final IndeterminateOutcomeException ioe = new IndeterminateOutcomeException("The connection was closed"); - // Things running remotely - for (ReplyHandler x : remoteRequests.getKeys()) { - SpiUtils.safeHandleException(x, ioe); - } - for (RequestHandler x : remoteClients.getKeys()) { - IoUtils.safeClose(x); - } - for (FutureRemoteRequestHandlerSource future : remoteServices.getKeys()) { - future.addNotifier(IoUtils.<RequestHandlerSource>closingNotifier(), null); - } - // Things running locally - for (RemoteRequestContext localRequest : localRequests.getKeys()) { - localRequest.cancel(); - } - for (Handle<RequestHandler> client : forwardedClients.getKeys()) { - IoUtils.safeClose(client); - } - for (Handle<RequestHandlerSource> service : forwardedServices.getKeys()) { - IoUtils.safeClose(service); - } - } - - public String toString() { - return "multiplex connection <" + Integer.toHexString(hashCode()) + "> via " + channel; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexConnection.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexConnection.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,447 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.marshalling.MarshallerFactory; +import org.jboss.marshalling.MarshallingConfiguration; +import org.jboss.xnio.BufferAllocator; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.Buffers; +import org.jboss.xnio.log.Logger; +import org.jboss.xnio.channels.AllocatedMessageChannel; +import org.jboss.remoting3.spi.ReplyHandler; +import org.jboss.remoting3.spi.RemoteRequestContext; +import org.jboss.remoting3.spi.RequestHandler; +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.spi.Handle; +import org.jboss.remoting3.spi.NamedServiceRegistry; +import org.jboss.remoting3.spi.SpiUtils; +import org.jboss.remoting3.spi.AbstractAutoCloseable; +import org.jboss.remoting3.QualifiedName; +import org.jboss.remoting3.IndeterminateOutcomeException; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.List; +import java.nio.ByteBuffer; +import java.io.IOException; +import java.io.InterruptedIOException; + +/** + * + */ +public final class MultiplexConnection extends AbstractAutoCloseable<Connection> implements Connection { + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex"); + + //--== Connection configuration items ==-- + private final MarshallerFactory marshallerFactory; + 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 IntegerBiMap<ReplyHandler> remoteRequests = IdentityHashIntegerBiMap.createSynchronizing(); + // running on local node + 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 IntegerBiMap<RequestHandler> remoteClients = IdentityHashIntegerBiMap.createSynchronizing(); + // forwarded to remote side (handled on this side) + 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 (shift left one bit, limit is 2^30) + private final AtomicInteger remoteClientSequence = new AtomicInteger(); + + // services on the remote side + private final IntegerBiMap<FutureRemoteRequestHandlerSource> remoteServices = IdentityHashIntegerBiMap.createSynchronizing(); + // forwarded to remote side (handled on this side) + private final IntegerResourceBiMap<RequestHandlerSource> forwardedServices = IdentityHashIntegerResourceBiMap.createSynchronizing(); + // sequence for remote services + private final AtomicInteger remoteServiceSequence = new AtomicInteger(); + + // registered services by path + private final NamedServiceRegistry namedServiceRegistry; + + private final AllocatedMessageChannel channel; + + public MultiplexConnection(final AllocatedMessageChannel channel, final MultiplexConfiguration configuration) { + super(configuration.getExecutor()); + this.channel = channel; + marshallerFactory = configuration.getMarshallerFactory(); + if (marshallerFactory == null) { + throw new NullPointerException("marshallerFactory is null"); + } + marshallingConfiguration = configuration.getMarshallingConfiguration(); + if (marshallingConfiguration == null) { + throw new NullPointerException("marshallingConfiguration is null"); + } + linkMetric = configuration.getLinkMetric(); + executor = configuration.getExecutor(); + if (executor == null) { + throw new NullPointerException("executor is null"); + } + allocator = configuration.getAllocator(); + if (allocator == null) { + throw new NullPointerException("allocator is null"); + } + namedServiceRegistry = configuration.getNamedServiceRegistry(); + if (namedServiceRegistry == null) { + throw new NullPointerException("namedServiceRegistry is null"); + } + } + + // sequence methods + + int nextRequest() { + return requestSequence.getAndIncrement() & 0x7fffffff; + } + + int nextForwardedClient() { + return (forwardedClientSequence.getAndIncrement() << 1 | 1) & 0x7fffffff; + } + + int nextRemoteClient() { + return remoteClientSequence.getAndIncrement() << 1 & 0x7fffffff; + } + + int nextRemoteService() { + return remoteServiceSequence.getAndIncrement() & 0x7fffffff; + } + + void doBlockingWrite(ByteBuffer... buffers) throws IOException { + log.trace("Sending message:\n%s", new MultiDumper(buffers)); + if (buffers.length == 1) doBlockingWrite(buffers[0]); else for (;;) { + if (channel.send(buffers)) { + return; + } + channel.awaitWritable(); + } + } + + private static final class MultiDumper { + private final ByteBuffer[] buffers; + + public MultiDumper(final ByteBuffer[] buffers) { + this.buffers = buffers; + } + + public String toString() { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < buffers.length; i++) { + ByteBuffer buffer = buffers[i]; + builder.append("Buffer "); + builder.append(i); + builder.append(":\n"); + try { + Buffers.dump(buffer, builder, 8, 1); + } catch (IOException e) { + // ignore + } + } + return builder.toString(); + } + } + + void doBlockingWrite(ByteBuffer buffer) throws IOException { + log.trace("Sending message:\n%s", Buffers.createDumper(buffer, 8, 1)); + for (;;) { + if (channel.send(buffer)) { + return; + } + channel.awaitWritable(); + } + } + + protected Executor getExecutor() { + return executor; + } + + void doBlockingWrite(List<ByteBuffer> buffers) throws IOException { + doBlockingWrite(buffers.toArray(new ByteBuffer[buffers.size()])); + } + + MarshallerFactory getMarshallerFactory() { + return marshallerFactory; + } + + MarshallingConfiguration getMarshallingConfiguration() { + return marshallingConfiguration; + } + + int getLinkMetric() { + return linkMetric; + } + + BufferAllocator<ByteBuffer> getAllocator() { + return allocator; + } + + AllocatedMessageChannel getChannel() { + return channel; + } + + void removeRemoteClient(final int identifier) { + remoteClients.remove(identifier); + } + + void addRemoteRequest(final int id, final ReplyHandler handler) { + remoteRequests.put(id, handler); + } + + void addRemoteClient(final int id, final RequestHandler handler) { + remoteClients.put(id, handler); + } + + Handle<RequestHandler> getForwardedClient(final int id) { + return forwardedClients.get(id); + } + + ReplyHandler removeRemoteRequest(final int id) { + return remoteRequests.remove(id); + } + + RemoteRequestContext getLocalRequest(final int id) { + return localRequests.get(id); + } + + ReplyHandler getRemoteRequest(final int id) { + return remoteRequests.get(id); + } + + Handle<RequestHandler> removeForwardedClient(final int id) { + return forwardedClients.remove(id); + } + + Handle<RequestHandlerSource> getForwardedService(final int id) { + return forwardedServices.get(id); + } + + void addForwardedClient(final int id, final Handle<RequestHandler> handle) { + forwardedClients.put(id, handle); + } + + void addForwadedService(final int id, final Handle<RequestHandlerSource> service) { + forwardedServices.put(id, service); + } + + Handle<RequestHandlerSource> removeForwardedService(final int id) { + return forwardedServices.remove(id); + } + + Handle<RequestHandlerSource> getServiceByPath(String path) { + return getService(QualifiedName.parse(path)); + } + + Handle<RequestHandlerSource> getService(final QualifiedName name) { + return namedServiceRegistry.lookupService(name); + } + + FutureRemoteRequestHandlerSource getFutureRemoteService(final int id) { + return remoteServices.get(id); + } + + FutureRemoteRequestHandlerSource removeFutureRemoteService(final int id) { + return remoteServices.remove(id); + } + + public Handle<RequestHandlerSource> openRemoteService(final QualifiedName name) throws IOException { + log.trace("Sending request to open remote service \"%s\"", name); + final FutureRemoteRequestHandlerSource future = new FutureRemoteRequestHandlerSource(); + int id; + for (;;) { + id = nextRemoteService(); + if (remoteServices.putIfAbsent(id, future)) { + break; + } + } + ByteBuffer buffer = ByteBuffer.allocate(5 + getByteLength(name)); + buffer.put((byte) MessageType.SERVICE_OPEN_REQUEST.getId()); + buffer.putInt(id); + putQualifiedName(buffer, name); + buffer.flip(); + doBlockingWrite(buffer); + try { + final Handle<RequestHandlerSource> handle = future.getInterruptibly().getHandle(); + log.trace("Opened %s", handle); + return handle; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while waiting for remote service"); + } + } + + static int getByteLength(QualifiedName name) { + int cnt = 2; // short header + for (String s : name) { + cnt += getByteLength(s); + } + return cnt; + } + + static int getByteLength(String s) { + final int len = s.length(); + int cnt = 0; + for (int i = 0; i < len; i++) { + char ch = s.charAt(i); + if (ch > 0 && ch <= 0x7f) { + cnt ++; + } else if (ch <= 0x07ff) { + cnt += 2; + } else { + cnt += 3; + } + } + // null terminator... + cnt ++; + return cnt; + } + + static String getString(final ByteBuffer buffer) { + StringBuilder builder = new StringBuilder(); + int state = 0, a = 0; + while (buffer.hasRemaining()) { + final int v = buffer.get() & 0xff; + switch (state) { + case 0: { + if (v == 0) { + return builder.toString(); + } else if (v < 128) { + builder.append((char) v); + } else if (192 <= v && v < 224) { + a = v << 6; + state = 1; + } else if (224 <= v && v < 232) { + a = v << 12; + state = 2; + } else { + builder.append('?'); + } + break; + } + case 1: { + if (v == 0) { + builder.append('?'); + return builder.toString(); + } else if (128 <= v && v < 192) { + a |= v & 0x3f; + builder.append((char) a); + } else { + builder.append('?'); + } + state = 0; + break; + } + case 2: { + if (v == 0) { + builder.append('?'); + return builder.toString(); + } else if (128 <= v && v < 192) { + a |= (v & 0x3f) << 6; + state = 1; + } else { + builder.append('?'); + state = 0; + } + break; + } + default: + throw new IllegalStateException("wrong state"); + } + } + return builder.toString(); + } + + static void putString(final ByteBuffer buffer, final String string) { + final int len = string.length(); + for (int i = 0; i < len; i ++) { + char ch = string.charAt(i); + if (ch > 0 && ch <= 0x7f) { + buffer.put((byte) ch); + } else if (ch <= 0x07ff) { + buffer.put((byte) (0xc0 | 0x1f & ch >> 6)); + buffer.put((byte) (0x80 | 0x3f & ch)); + } else { + buffer.put((byte) (0xe0 | 0x0f & ch >> 12)); + buffer.put((byte) (0x80 | 0x3f & ch >> 6)); + buffer.put((byte) (0x80 | 0x3f & ch)); + } + } + buffer.put((byte) 0); + } + + static QualifiedName getQualifiedName(final ByteBuffer buffer) { + final int len = buffer.getShort() & 0xffff; + final String[] segs = new String[len]; + for (int i = 0; i < len; i++) { + segs[i] = getString(buffer); + } + return new QualifiedName(segs); + } + + static void putQualifiedName(final ByteBuffer buffer, final QualifiedName qualifiedName) { + final int len = qualifiedName.length(); + if (len > 0xffff) { + throw new IllegalArgumentException("Qualified name is too long"); + } + buffer.putShort((short) len); + for (String seg : qualifiedName) { + putString(buffer, seg); + } + } + + protected void closeAction() { + // just to make sure... + IoUtils.safeClose(channel); + final IndeterminateOutcomeException ioe = new IndeterminateOutcomeException("The connection was closed"); + // Things running remotely + for (ReplyHandler x : remoteRequests.getKeys()) { + SpiUtils.safeHandleException(x, ioe); + } + for (RequestHandler x : remoteClients.getKeys()) { + IoUtils.safeClose(x); + } + for (FutureRemoteRequestHandlerSource future : remoteServices.getKeys()) { + future.addNotifier(IoUtils.<RequestHandlerSource>closingNotifier(), null); + } + // Things running locally + for (RemoteRequestContext localRequest : localRequests.getKeys()) { + localRequest.cancel(); + } + for (Handle<RequestHandler> client : forwardedClients.getKeys()) { + IoUtils.safeClose(client); + } + for (Handle<RequestHandlerSource> service : forwardedServices.getKeys()) { + IoUtils.safeClose(service); + } + } + + public String toString() { + return "multiplex connection <" + Integer.toHexString(hashCode()) + "> via " + channel; + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexProtocol.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,75 +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 org.jboss.remoting.Endpoint; -import org.jboss.xnio.IoHandlerFactory; -import org.jboss.xnio.ChannelSource; -import org.jboss.xnio.IoFuture; -import org.jboss.xnio.AbstractConvertingIoFuture; -import org.jboss.xnio.IoHandler; -import org.jboss.xnio.channels.AllocatedMessageChannel; -import java.io.IOException; - -/** - * - */ -public final class MultiplexProtocol { - - private MultiplexProtocol() { - } - - /** - * Create a request server for the multiplex protocol. - * - * @param endpoint the endpoint - * @param configuration the configuration - * @return a handler factory for passing to an XNIO server - */ - public static IoHandlerFactory<AllocatedMessageChannel> createServer(final Endpoint endpoint, final MultiplexConfiguration configuration) { - return new IoHandlerFactory<AllocatedMessageChannel>() { - public IoHandler<? super AllocatedMessageChannel> createHandler() { - return new SimpleMultiplexHandler(endpoint, configuration); - } - }; - } - - /** - * Create a request client for the multiplex protocol. - * - * @param endpoint the endpoint - * @param configuration the configuration - * @param channelSource the XNIO channel source to use to establish the connection - * @return a handle which may be used to close the connection - * @throws IOException if an error occurs - */ - public static IoFuture<MultiplexConnection> connect(final Endpoint endpoint, final MultiplexConfiguration configuration, final ChannelSource<AllocatedMessageChannel> channelSource) throws IOException { - final SimpleMultiplexHandler handler = new SimpleMultiplexHandler(endpoint, configuration); - final IoFuture<AllocatedMessageChannel> futureChannel = channelSource.open(handler); - return new AbstractConvertingIoFuture<MultiplexConnection, AllocatedMessageChannel>(futureChannel) { - protected MultiplexConnection convert(final AllocatedMessageChannel channel) throws IOException { - return handler.getConnection().get(); - } - }; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexProtocol.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexProtocol.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,72 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.xnio.IoHandlerFactory; +import org.jboss.xnio.ChannelSource; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.AbstractConvertingIoFuture; +import org.jboss.xnio.IoHandler; +import org.jboss.xnio.channels.AllocatedMessageChannel; +import java.io.IOException; + +/** + * + */ +public final class MultiplexProtocol { + + private MultiplexProtocol() { + } + + /** + * Create a request server for the multiplex protocol. + * + * @param configuration the configuration + * @return a handler factory for passing to an XNIO server + */ + public static IoHandlerFactory<AllocatedMessageChannel> createServer(final MultiplexConfiguration configuration) { + return new IoHandlerFactory<AllocatedMessageChannel>() { + public IoHandler<? super AllocatedMessageChannel> createHandler() { + return new SimpleMultiplexHandler(configuration); + } + }; + } + + /** + * Create a request client for the multiplex protocol. + * + * @param configuration the configuration + * @param channelSource the XNIO channel source to use to establish the connection + * @return a handle which may be used to close the connection + * @throws IOException if an error occurs + */ + public static IoFuture<MultiplexConnection> connect(final MultiplexConfiguration configuration, final ChannelSource<AllocatedMessageChannel> channelSource) throws IOException { + final SimpleMultiplexHandler handler = new SimpleMultiplexHandler(configuration); + final IoFuture<AllocatedMessageChannel> futureChannel = channelSource.open(handler); + return new AbstractConvertingIoFuture<MultiplexConnection, AllocatedMessageChannel>(futureChannel) { + protected MultiplexConnection convert(final AllocatedMessageChannel channel) throws IOException { + return handler.getConnection().get(); + } + }; + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexReadHandler.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,391 +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 org.jboss.xnio.IoReadHandler; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.Buffers; -import org.jboss.xnio.log.Logger; -import org.jboss.xnio.channels.AllocatedMessageChannel; -import org.jboss.remoting.spi.RequestHandler; -import org.jboss.remoting.spi.Handle; -import org.jboss.remoting.spi.ReplyHandler; -import org.jboss.remoting.spi.SpiUtils; -import org.jboss.remoting.spi.RemoteRequestContext; -import org.jboss.remoting.spi.RequestHandlerSource; -import org.jboss.remoting.ReplyException; -import org.jboss.remoting.RemoteExecutionException; -import org.jboss.remoting.ServiceRegistrationException; -import org.jboss.remoting.QualifiedName; -import org.jboss.marshalling.Unmarshaller; -import org.jboss.marshalling.Marshalling; -import org.jboss.marshalling.Marshaller; -import org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.MarshallingConfiguration; -import java.nio.ByteBuffer; -import java.nio.BufferUnderflowException; -import java.io.IOException; -import java.util.List; -import java.util.ArrayList; - -/** - * - */ -public final class MultiplexReadHandler implements IoReadHandler<AllocatedMessageChannel> { - - private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex"); - private static final StackTraceElement[] emptyStackTraceElements = new StackTraceElement[0]; - private final MultiplexConnection connection; - - public MultiplexReadHandler(final MultiplexConnection connection) { - this.connection = connection; - } - - public void handleReadable(final AllocatedMessageChannel channel) { - final MultiplexConnection connection = this.connection; - final MarshallerFactory marshallerFactory = connection.getMarshallerFactory(); - final MarshallingConfiguration marshallingConfiguration = connection.getMarshallingConfiguration(); - for (;;) try { - final ByteBuffer buffer; - try { - buffer = channel.receive(); - } catch (IOException e) { - log.error(e, "I/O error in protocol channel; closing channel"); - IoUtils.safeClose(channel); - return; - } - if (buffer == null) { - IoUtils.safeClose(channel); - return; - } - if (! buffer.hasRemaining()) { - // would block - channel.resumeReads(); - return; - } - log.trace("Received raw message:\n%s", Buffers.createDumper(buffer, 8, 1)); - final MessageType msgType; - try { - msgType = MessageType.getMessageType(buffer.get() & 0xff); - } catch (IllegalArgumentException ex) { - log.trace("Received invalid message type"); - return; - } - switch (msgType) { - case REQUEST: { - final int clientId = buffer.getInt(); - final Handle<RequestHandler> handle = connection.getForwardedClient(clientId); - if (handle == null) { - log.trace("Request on invalid client ID %d", Integer.valueOf(clientId)); - break; - } - final int requestId = buffer.getInt(); - final Object payload; - try { - final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); - try { - unmarshaller.start(Marshalling.createByteInput(buffer)); - payload = unmarshaller.readObject(); - unmarshaller.finish(); - log.trace("Received inbound request (client id = %d, request id = %d) (type is %s)", Integer.valueOf(clientId), Integer.valueOf(requestId), payload == null ? "null" : payload.getClass()); - } finally { - IoUtils.safeClose(unmarshaller); - } - } catch (Exception ex) { - // IOException | ClassNotFoundException - log.trace("Failed to unmarshal a request (%s), sending %s", ex, MessageType.REQUEST_RECEIVE_FAILED); - try { - final Marshaller marshaller = marshallerFactory.createMarshaller(marshallingConfiguration); - try { - List<ByteBuffer> buffers = new ArrayList<ByteBuffer>(); - marshaller.start(new BufferByteOutput(connection.getAllocator(), buffers)); - marshaller.write(MessageType.REQUEST_RECEIVE_FAILED.getId()); - ex.setStackTrace(emptyStackTraceElements); - final IOException ioe = new IOException("Request receive failed"); - ioe.initCause(ex); - ioe.setStackTrace(emptyStackTraceElements); - marshaller.writeObject(ioe); - marshaller.finish(); - connection.doBlockingWrite(buffers); - } 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 MultiplexReplyHandler(requestId, connection)); - break; - } - case REPLY: { - final int requestId = buffer.getInt(); - final ReplyHandler replyHandler = connection.removeRemoteRequest(requestId); - if (replyHandler == null) { - log.trace("Got reply to unknown request %d", Integer.valueOf(requestId)); - break; - } - final Object payload; - try { - final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); - try { - unmarshaller.start(Marshalling.createByteInput(buffer)); - payload = unmarshaller.readObject(); - unmarshaller.finish(); - log.trace("Received inbound reply (id = %d) (type is %s)", Integer.valueOf(requestId), payload == null ? "null" : payload.getClass()); - } finally { - IoUtils.safeClose(unmarshaller); - } - } catch (Exception ex) { - // IOException | ClassNotFoundException - log.trace("Failed to unmarshal a reply (%s), sending a ReplyException", ex); - SpiUtils.safeHandleException(replyHandler, new ReplyException("Unmarshal failed", ex)); - break; - } - SpiUtils.safeHandleReply(replyHandler, payload); - break; - } - case CANCEL_REQUEST: { - final int requestId = buffer.getInt(); - final RemoteRequestContext context = connection.getLocalRequest(requestId); - if (context != null) { - log.trace("Received inbound cancel request (request id = %d) to %s", Integer.valueOf(requestId), context); - context.cancel(); - } - break; - } - case CANCEL_ACK: { - final int requestId = buffer.getInt(); - final ReplyHandler replyHandler = connection.getRemoteRequest(requestId); - if (replyHandler != null) { - log.trace("Received inbound cancel acknowledge (request id = %d) to ", Integer.valueOf(requestId), replyHandler); - SpiUtils.safeHandleCancellation(replyHandler); - } - break; - } - case REQUEST_RECEIVE_FAILED: { - final int requestId = buffer.getInt(); - final ReplyHandler replyHandler = connection.removeRemoteRequest(requestId); - if (replyHandler == null) { - log.trace("Got a failure reply to unknown request %d to %s", Integer.valueOf(requestId), replyHandler); - break; - } - final IOException cause; - try { - final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); - try { - unmarshaller.start(Marshalling.createByteInput(buffer)); - cause = (IOException) unmarshaller.readObject(); - unmarshaller.finish(); - log.trace("Received inbound request receive failure notification to %s: %s", replyHandler, cause); - } finally { - IoUtils.safeClose(unmarshaller); - } - } catch (IOException e) { - log.trace("Received inbound request receive failure notification; the remote exception could not be read: %s", e); - SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote operation failed; the remote exception could not be read", e)); - break; - } catch (ClassNotFoundException e) { - log.trace("Received inbound request receive failure notification; the remote exception could not be read: %s", 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: { - final int requestId = buffer.getInt(); - final ReplyHandler replyHandler = connection.removeRemoteRequest(requestId); - if (replyHandler == null) { - log.trace("Got reply to unknown request %d", Integer.valueOf(requestId)); - break; - } - final IOException cause; - try { - final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); - try { - unmarshaller.start(Marshalling.createByteInput(buffer)); - try { - cause = (IOException) unmarshaller.readObject(); - log.trace("Received inbound request failure notification: %s", cause); - } catch (ClassNotFoundException e) { - log.trace("Received inbound request failure notification; the remote exception could not be read: %s", e); - SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote request failed (and an ClassNotFoundException occurred when attempting to unmarshal the cause)")); - log.trace(e, "Class not found in exception reply to request ID %d", Integer.valueOf(requestId)); - break; - } catch (ClassCastException e) { - log.trace("Received inbound request failure notification; the remote exception could not be read: %s", e); - SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote request failed (and an ClassCastException occurred when attempting to unmarshal the cause)")); - log.trace(e, "Class cast exception in exception reply to request ID %d", Integer.valueOf(requestId)); - break; - } - } finally { - IoUtils.safeClose(unmarshaller); - } - } catch (IOException ex) { - log.trace("Received inbound request failure notification; the remote exception could not be read: %s", ex); - SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote request failed (and an unexpected I/O error occurred when attempting to read the cause)")); - break; - } - SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote execution failed", cause)); - break; - } - case CLIENT_CLOSE: { - final int clientId = buffer.getInt(); - final Handle<RequestHandler> handle = connection.removeForwardedClient(clientId); - if (handle == null) { - log.warn("Got client close message for unknown client %d", Integer.valueOf(clientId)); - break; - } - log.trace("Received inbound client close for %s", handle); - IoUtils.safeClose(handle); - break; - } - case CLIENT_OPEN: { - final int serviceId = buffer.getInt(); - final int clientId = buffer.getInt(); - final Handle<RequestHandlerSource> handle = connection.getForwardedService(serviceId); - if (handle == null) { - log.warn("Received client open message for unknown service %d", Integer.valueOf(serviceId)); - break; - } - try { - final RequestHandlerSource requestHandlerSource = handle.getResource(); - final Handle<RequestHandler> clientHandle = requestHandlerSource.createRequestHandler(); - log.trace("Opening client %d from service %d", Integer.valueOf(clientId), Integer.valueOf(serviceId)); - connection.addForwardedClient(clientId, clientHandle); - } catch (IOException ex) { - log.error(ex, "Failed to create a request handler for client ID %d", Integer.valueOf(clientId)); - break; - } finally { - IoUtils.safeClose(handle); - } - break; - } - case SERVICE_OPEN_REQUEST: { - final int serviceId = buffer.getInt(); - final QualifiedName qualifiedName = MultiplexConnection.getQualifiedName(buffer); - log.trace("Received a service open request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); - final Handle<RequestHandlerSource> service = connection.getService(qualifiedName); - if (service == null) { - ByteBuffer replyBuffer = ByteBuffer.allocate(5); - replyBuffer.put((byte) MessageType.SERVICE_OPEN_NOT_FOUND.getId()); - replyBuffer.putInt(serviceId); - replyBuffer.flip(); - try { - log.trace("Sending a service-open-not-found message for request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); - connection.doBlockingWrite(replyBuffer); - } catch (IOException e) { - log.error(e, "Failed to send an error reply to an invalid service open request"); - } - break; - } - final Handle<RequestHandlerSource> ourHandle; - try { - ourHandle = service.getResource().getHandle(); - } catch (IOException e) { - log.error("Failed to acquire a handle to registered service: %s", e); - ByteBuffer replyBuffer = ByteBuffer.allocate(5); - replyBuffer.put((byte) MessageType.SERVICE_OPEN_FAILED.getId()); - replyBuffer.putInt(serviceId); - replyBuffer.flip(); - try { - log.trace("Sending a service-open-failed message for request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); - connection.doBlockingWrite(replyBuffer); - } catch (IOException e2) { - log.trace(e, "Failed to send an exception reply to a service open request"); - } - break; - } - connection.addForwadedService(serviceId, ourHandle); - ByteBuffer replyBuffer = ByteBuffer.allocate(5); - replyBuffer.put((byte) MessageType.SERVICE_OPEN_REPLY.getId()); - replyBuffer.putInt(serviceId); - replyBuffer.flip(); - try { - log.trace("Sending a service open reply message for request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); - connection.doBlockingWrite(replyBuffer); - } catch (IOException e) { - log.trace(e, "Failed to send a reply to a service open request"); - } - break; - } - case SERVICE_OPEN_FAILED: - case SERVICE_OPEN_NOT_FOUND: - case SERVICE_OPEN_FORBIDDEN: { - final int serviceId = buffer.getInt(); - log.trace("Received a service open failure (%s) message for service %d", msgType, Integer.valueOf(serviceId)); - final FutureRemoteRequestHandlerSource future = connection.removeFutureRemoteService(serviceId); - if (future == null) { - log.trace("Service open failure reply received for unknown service ID %d", Integer.valueOf(serviceId)); - break; - } - future.setException( - msgType == MessageType.SERVICE_OPEN_NOT_FOUND ? new ServiceRegistrationException("Service not found") : - msgType == MessageType.SERVICE_OPEN_FORBIDDEN ? new ServiceRegistrationException("Service open forbidden") : - new ServiceRegistrationException("Service open failed") - ); - break; - } - case SERVICE_OPEN_REPLY: { - final int serviceId = buffer.getInt(); - final FutureRemoteRequestHandlerSource future = connection.getFutureRemoteService(serviceId); - if (future == null) { - log.trace("Service open reply received for unknown service ID %d", Integer.valueOf(serviceId)); - break; - } - log.trace("Received a service open reply message for service %d for %s", Integer.valueOf(serviceId), future); - final MultiplexRequestHandlerSource requestHandlerSource = new MultiplexRequestHandlerSource(serviceId, connection); - future.setResult(requestHandlerSource); - break; - } - case SERVICE_CLOSE_NOTIFY: { - final int serviceId = buffer.getInt(); - final FutureRemoteRequestHandlerSource future = connection.removeFutureRemoteService(serviceId); - log.trace("Received a service close notify message for service %d for %s", Integer.valueOf(serviceId), future); - future.addNotifier(IoUtils.<RequestHandlerSource>closingNotifier(), null); - break; - } - case SERVICE_CLOSE_REQUEST: { - final int serviceId = buffer.getInt(); - final Handle<RequestHandlerSource> handle = connection.removeForwardedService(serviceId); - if (handle == null) { - log.trace("Received service close request on unknown ID %d", Integer.valueOf(serviceId)); - break; - } - IoUtils.safeClose(handle); - break; - } - default: { - log.error("Malformed packet received (invalid message type %s)", msgType); - } - case CONNECTION_CLOSE: - break; - } - } catch (BufferUnderflowException e) { - log.error(e, "Malformed packet received (buffer underflow)"); - } - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexReadHandler.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReadHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,391 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.xnio.IoReadHandler; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.Buffers; +import org.jboss.xnio.log.Logger; +import org.jboss.xnio.channels.AllocatedMessageChannel; +import org.jboss.remoting3.spi.RequestHandler; +import org.jboss.remoting3.spi.Handle; +import org.jboss.remoting3.spi.ReplyHandler; +import org.jboss.remoting3.spi.SpiUtils; +import org.jboss.remoting3.spi.RemoteRequestContext; +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.ReplyException; +import org.jboss.remoting3.RemoteExecutionException; +import org.jboss.remoting3.ServiceRegistrationException; +import org.jboss.remoting3.QualifiedName; +import org.jboss.marshalling.Unmarshaller; +import org.jboss.marshalling.Marshalling; +import org.jboss.marshalling.Marshaller; +import org.jboss.marshalling.MarshallerFactory; +import org.jboss.marshalling.MarshallingConfiguration; +import java.nio.ByteBuffer; +import java.nio.BufferUnderflowException; +import java.io.IOException; +import java.util.List; +import java.util.ArrayList; + +/** + * + */ +public final class MultiplexReadHandler implements IoReadHandler<AllocatedMessageChannel> { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex"); + private static final StackTraceElement[] emptyStackTraceElements = new StackTraceElement[0]; + private final MultiplexConnection connection; + + public MultiplexReadHandler(final MultiplexConnection connection) { + this.connection = connection; + } + + public void handleReadable(final AllocatedMessageChannel channel) { + final MultiplexConnection connection = this.connection; + final MarshallerFactory marshallerFactory = connection.getMarshallerFactory(); + final MarshallingConfiguration marshallingConfiguration = connection.getMarshallingConfiguration(); + for (;;) try { + final ByteBuffer buffer; + try { + buffer = channel.receive(); + } catch (IOException e) { + log.error(e, "I/O error in protocol channel; closing channel"); + IoUtils.safeClose(channel); + return; + } + if (buffer == null) { + IoUtils.safeClose(channel); + return; + } + if (! buffer.hasRemaining()) { + // would block + channel.resumeReads(); + return; + } + log.trace("Received raw message:\n%s", Buffers.createDumper(buffer, 8, 1)); + final MessageType msgType; + try { + msgType = MessageType.getMessageType(buffer.get() & 0xff); + } catch (IllegalArgumentException ex) { + log.trace("Received invalid message type"); + return; + } + switch (msgType) { + case REQUEST: { + final int clientId = buffer.getInt(); + final Handle<RequestHandler> handle = connection.getForwardedClient(clientId); + if (handle == null) { + log.trace("Request on invalid client ID %d", Integer.valueOf(clientId)); + break; + } + final int requestId = buffer.getInt(); + final Object payload; + try { + final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); + try { + unmarshaller.start(Marshalling.createByteInput(buffer)); + payload = unmarshaller.readObject(); + unmarshaller.finish(); + log.trace("Received inbound request (client id = %d, request id = %d) (type is %s)", Integer.valueOf(clientId), Integer.valueOf(requestId), payload == null ? "null" : payload.getClass()); + } finally { + IoUtils.safeClose(unmarshaller); + } + } catch (Exception ex) { + // IOException | ClassNotFoundException + log.trace("Failed to unmarshal a request (%s), sending %s", ex, MessageType.REQUEST_RECEIVE_FAILED); + try { + final Marshaller marshaller = marshallerFactory.createMarshaller(marshallingConfiguration); + try { + List<ByteBuffer> buffers = new ArrayList<ByteBuffer>(); + marshaller.start(new BufferByteOutput(connection.getAllocator(), buffers)); + marshaller.write(MessageType.REQUEST_RECEIVE_FAILED.getId()); + ex.setStackTrace(emptyStackTraceElements); + final IOException ioe = new IOException("Request receive failed"); + ioe.initCause(ex); + ioe.setStackTrace(emptyStackTraceElements); + marshaller.writeObject(ioe); + marshaller.finish(); + connection.doBlockingWrite(buffers); + } 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 MultiplexReplyHandler(requestId, connection)); + break; + } + case REPLY: { + final int requestId = buffer.getInt(); + final ReplyHandler replyHandler = connection.removeRemoteRequest(requestId); + if (replyHandler == null) { + log.trace("Got reply to unknown request %d", Integer.valueOf(requestId)); + break; + } + final Object payload; + try { + final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); + try { + unmarshaller.start(Marshalling.createByteInput(buffer)); + payload = unmarshaller.readObject(); + unmarshaller.finish(); + log.trace("Received inbound reply (id = %d) (type is %s)", Integer.valueOf(requestId), payload == null ? "null" : payload.getClass()); + } finally { + IoUtils.safeClose(unmarshaller); + } + } catch (Exception ex) { + // IOException | ClassNotFoundException + log.trace("Failed to unmarshal a reply (%s), sending a ReplyException", ex); + SpiUtils.safeHandleException(replyHandler, new ReplyException("Unmarshal failed", ex)); + break; + } + SpiUtils.safeHandleReply(replyHandler, payload); + break; + } + case CANCEL_REQUEST: { + final int requestId = buffer.getInt(); + final RemoteRequestContext context = connection.getLocalRequest(requestId); + if (context != null) { + log.trace("Received inbound cancel request (request id = %d) to %s", Integer.valueOf(requestId), context); + context.cancel(); + } + break; + } + case CANCEL_ACK: { + final int requestId = buffer.getInt(); + final ReplyHandler replyHandler = connection.getRemoteRequest(requestId); + if (replyHandler != null) { + log.trace("Received inbound cancel acknowledge (request id = %d) to ", Integer.valueOf(requestId), replyHandler); + SpiUtils.safeHandleCancellation(replyHandler); + } + break; + } + case REQUEST_RECEIVE_FAILED: { + final int requestId = buffer.getInt(); + final ReplyHandler replyHandler = connection.removeRemoteRequest(requestId); + if (replyHandler == null) { + log.trace("Got a failure reply to unknown request %d to %s", Integer.valueOf(requestId), replyHandler); + break; + } + final IOException cause; + try { + final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); + try { + unmarshaller.start(Marshalling.createByteInput(buffer)); + cause = (IOException) unmarshaller.readObject(); + unmarshaller.finish(); + log.trace("Received inbound request receive failure notification to %s: %s", replyHandler, cause); + } finally { + IoUtils.safeClose(unmarshaller); + } + } catch (IOException e) { + log.trace("Received inbound request receive failure notification; the remote exception could not be read: %s", e); + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote operation failed; the remote exception could not be read", e)); + break; + } catch (ClassNotFoundException e) { + log.trace("Received inbound request receive failure notification; the remote exception could not be read: %s", 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: { + final int requestId = buffer.getInt(); + final ReplyHandler replyHandler = connection.removeRemoteRequest(requestId); + if (replyHandler == null) { + log.trace("Got reply to unknown request %d", Integer.valueOf(requestId)); + break; + } + final IOException cause; + try { + final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallingConfiguration); + try { + unmarshaller.start(Marshalling.createByteInput(buffer)); + try { + cause = (IOException) unmarshaller.readObject(); + log.trace("Received inbound request failure notification: %s", cause); + } catch (ClassNotFoundException e) { + log.trace("Received inbound request failure notification; the remote exception could not be read: %s", e); + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote request failed (and an ClassNotFoundException occurred when attempting to unmarshal the cause)")); + log.trace(e, "Class not found in exception reply to request ID %d", Integer.valueOf(requestId)); + break; + } catch (ClassCastException e) { + log.trace("Received inbound request failure notification; the remote exception could not be read: %s", e); + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote request failed (and an ClassCastException occurred when attempting to unmarshal the cause)")); + log.trace(e, "Class cast exception in exception reply to request ID %d", Integer.valueOf(requestId)); + break; + } + } finally { + IoUtils.safeClose(unmarshaller); + } + } catch (IOException ex) { + log.trace("Received inbound request failure notification; the remote exception could not be read: %s", ex); + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote request failed (and an unexpected I/O error occurred when attempting to read the cause)")); + break; + } + SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote execution failed", cause)); + break; + } + case CLIENT_CLOSE: { + final int clientId = buffer.getInt(); + final Handle<RequestHandler> handle = connection.removeForwardedClient(clientId); + if (handle == null) { + log.warn("Got client close message for unknown client %d", Integer.valueOf(clientId)); + break; + } + log.trace("Received inbound client close for %s", handle); + IoUtils.safeClose(handle); + break; + } + case CLIENT_OPEN: { + final int serviceId = buffer.getInt(); + final int clientId = buffer.getInt(); + final Handle<RequestHandlerSource> handle = connection.getForwardedService(serviceId); + if (handle == null) { + log.warn("Received client open message for unknown service %d", Integer.valueOf(serviceId)); + break; + } + try { + final RequestHandlerSource requestHandlerSource = handle.getResource(); + final Handle<RequestHandler> clientHandle = requestHandlerSource.createRequestHandler(); + log.trace("Opening client %d from service %d", Integer.valueOf(clientId), Integer.valueOf(serviceId)); + connection.addForwardedClient(clientId, clientHandle); + } catch (IOException ex) { + log.error(ex, "Failed to create a request handler for client ID %d", Integer.valueOf(clientId)); + break; + } finally { + IoUtils.safeClose(handle); + } + break; + } + case SERVICE_OPEN_REQUEST: { + final int serviceId = buffer.getInt(); + final QualifiedName qualifiedName = MultiplexConnection.getQualifiedName(buffer); + log.trace("Received a service open request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); + final Handle<RequestHandlerSource> service = connection.getService(qualifiedName); + if (service == null) { + ByteBuffer replyBuffer = ByteBuffer.allocate(5); + replyBuffer.put((byte) MessageType.SERVICE_OPEN_NOT_FOUND.getId()); + replyBuffer.putInt(serviceId); + replyBuffer.flip(); + try { + log.trace("Sending a service-open-not-found message for request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); + connection.doBlockingWrite(replyBuffer); + } catch (IOException e) { + log.error(e, "Failed to send an error reply to an invalid service open request"); + } + break; + } + final Handle<RequestHandlerSource> ourHandle; + try { + ourHandle = service.getResource().getHandle(); + } catch (IOException e) { + log.error("Failed to acquire a handle to registered service: %s", e); + ByteBuffer replyBuffer = ByteBuffer.allocate(5); + replyBuffer.put((byte) MessageType.SERVICE_OPEN_FAILED.getId()); + replyBuffer.putInt(serviceId); + replyBuffer.flip(); + try { + log.trace("Sending a service-open-failed message for request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); + connection.doBlockingWrite(replyBuffer); + } catch (IOException e2) { + log.trace(e, "Failed to send an exception reply to a service open request"); + } + break; + } + connection.addForwadedService(serviceId, ourHandle); + ByteBuffer replyBuffer = ByteBuffer.allocate(5); + replyBuffer.put((byte) MessageType.SERVICE_OPEN_REPLY.getId()); + replyBuffer.putInt(serviceId); + replyBuffer.flip(); + try { + log.trace("Sending a service open reply message for request for service %d on path \"%s\"", Integer.valueOf(serviceId), qualifiedName); + connection.doBlockingWrite(replyBuffer); + } catch (IOException e) { + log.trace(e, "Failed to send a reply to a service open request"); + } + break; + } + case SERVICE_OPEN_FAILED: + case SERVICE_OPEN_NOT_FOUND: + case SERVICE_OPEN_FORBIDDEN: { + final int serviceId = buffer.getInt(); + log.trace("Received a service open failure (%s) message for service %d", msgType, Integer.valueOf(serviceId)); + final FutureRemoteRequestHandlerSource future = connection.removeFutureRemoteService(serviceId); + if (future == null) { + log.trace("Service open failure reply received for unknown service ID %d", Integer.valueOf(serviceId)); + break; + } + future.setException( + msgType == MessageType.SERVICE_OPEN_NOT_FOUND ? new ServiceRegistrationException("Service not found") : + msgType == MessageType.SERVICE_OPEN_FORBIDDEN ? new ServiceRegistrationException("Service open forbidden") : + new ServiceRegistrationException("Service open failed") + ); + break; + } + case SERVICE_OPEN_REPLY: { + final int serviceId = buffer.getInt(); + final FutureRemoteRequestHandlerSource future = connection.getFutureRemoteService(serviceId); + if (future == null) { + log.trace("Service open reply received for unknown service ID %d", Integer.valueOf(serviceId)); + break; + } + log.trace("Received a service open reply message for service %d for %s", Integer.valueOf(serviceId), future); + final MultiplexRequestHandlerSource requestHandlerSource = new MultiplexRequestHandlerSource(serviceId, connection); + future.setResult(requestHandlerSource); + break; + } + case SERVICE_CLOSE_NOTIFY: { + final int serviceId = buffer.getInt(); + final FutureRemoteRequestHandlerSource future = connection.removeFutureRemoteService(serviceId); + log.trace("Received a service close notify message for service %d for %s", Integer.valueOf(serviceId), future); + future.addNotifier(IoUtils.<RequestHandlerSource>closingNotifier(), null); + break; + } + case SERVICE_CLOSE_REQUEST: { + final int serviceId = buffer.getInt(); + final Handle<RequestHandlerSource> handle = connection.removeForwardedService(serviceId); + if (handle == null) { + log.trace("Received service close request on unknown ID %d", Integer.valueOf(serviceId)); + break; + } + IoUtils.safeClose(handle); + break; + } + default: { + log.error("Malformed packet received (invalid message type %s)", msgType); + } + case CONNECTION_CLOSE: + break; + } + } catch (BufferUnderflowException e) { + log.error(e, "Malformed packet received (buffer underflow)"); + } + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexReplyHandler.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,106 +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 org.jboss.remoting.spi.ReplyHandler; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.log.Logger; -import org.jboss.marshalling.Marshaller; -import org.jboss.marshalling.ByteOutput; -import java.nio.ByteBuffer; -import java.io.IOException; -import java.util.List; -import java.util.ArrayList; - -/** - * - */ -final class MultiplexReplyHandler implements ReplyHandler { - private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.reply-handler"); - - private final int requestId; - private final MultiplexConnection connection; - - MultiplexReplyHandler(final int requestId, final MultiplexConnection connection) { - this.requestId = requestId; - this.connection = connection; - } - - public void handleReply(final Object reply) throws IOException { - log.trace("Sending outbound reply (request id = %d) (type is %s)", Integer.valueOf(requestId), reply == null ? "null" : reply.getClass()); - final MultiplexConnection connection = this.connection; - final Marshaller marshaller = connection.getMarshallerFactory().createMarshaller(connection.getMarshallingConfiguration()); - try { - final List<ByteBuffer> bufferList = new ArrayList<ByteBuffer>(); - final ByteOutput output = new BufferByteOutput(connection.getAllocator(), bufferList); - try { - marshaller.start(output); - marshaller.write(MessageType.REPLY.getId()); - marshaller.writeInt(requestId); - marshaller.writeObject(reply); - marshaller.close(); - output.close(); - connection.doBlockingWrite(bufferList); - log.trace("Sent reply %s", reply); - } finally { - IoUtils.safeClose(output); - } - } finally { - IoUtils.safeClose(marshaller); - } - } - - public void handleException(final IOException exception) throws IOException { - final MultiplexConnection connection = this.connection; - final Marshaller marshaller = connection.getMarshallerFactory().createMarshaller(connection.getMarshallingConfiguration()); - try { - final List<ByteBuffer> bufferList = new ArrayList<ByteBuffer>(); - final ByteOutput output = new BufferByteOutput(connection.getAllocator(), bufferList); - try { - marshaller.start(output); - marshaller.write(MessageType.REQUEST_FAILED.getId()); - marshaller.writeInt(requestId); - marshaller.writeObject(exception); - marshaller.close(); - output.close(); - connection.doBlockingWrite(bufferList); - } finally { - IoUtils.safeClose(output); - } - } finally { - IoUtils.safeClose(marshaller); - } - } - - public void handleCancellation() throws IOException { - final ByteBuffer buffer = ByteBuffer.allocate(5); - buffer.put((byte) MessageType.CANCEL_ACK.getId()); - buffer.putInt(requestId); - buffer.flip(); - connection.doBlockingWrite(buffer); - } - - public String toString() { - return "forwarding reply handler <" + Integer.toHexString(hashCode()) + "> (request id = " + requestId + ") for " + connection; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexReplyHandler.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexReplyHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,106 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.spi.ReplyHandler; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.log.Logger; +import org.jboss.marshalling.Marshaller; +import org.jboss.marshalling.ByteOutput; +import java.nio.ByteBuffer; +import java.io.IOException; +import java.util.List; +import java.util.ArrayList; + +/** + * + */ +final class MultiplexReplyHandler implements ReplyHandler { + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.reply-handler"); + + private final int requestId; + private final MultiplexConnection connection; + + MultiplexReplyHandler(final int requestId, final MultiplexConnection connection) { + this.requestId = requestId; + this.connection = connection; + } + + public void handleReply(final Object reply) throws IOException { + log.trace("Sending outbound reply (request id = %d) (type is %s)", Integer.valueOf(requestId), reply == null ? "null" : reply.getClass()); + final MultiplexConnection connection = this.connection; + final Marshaller marshaller = connection.getMarshallerFactory().createMarshaller(connection.getMarshallingConfiguration()); + try { + final List<ByteBuffer> bufferList = new ArrayList<ByteBuffer>(); + final ByteOutput output = new BufferByteOutput(connection.getAllocator(), bufferList); + try { + marshaller.start(output); + marshaller.write(MessageType.REPLY.getId()); + marshaller.writeInt(requestId); + marshaller.writeObject(reply); + marshaller.close(); + output.close(); + connection.doBlockingWrite(bufferList); + log.trace("Sent reply %s", reply); + } finally { + IoUtils.safeClose(output); + } + } finally { + IoUtils.safeClose(marshaller); + } + } + + public void handleException(final IOException exception) throws IOException { + final MultiplexConnection connection = this.connection; + final Marshaller marshaller = connection.getMarshallerFactory().createMarshaller(connection.getMarshallingConfiguration()); + try { + final List<ByteBuffer> bufferList = new ArrayList<ByteBuffer>(); + final ByteOutput output = new BufferByteOutput(connection.getAllocator(), bufferList); + try { + marshaller.start(output); + marshaller.write(MessageType.REQUEST_FAILED.getId()); + marshaller.writeInt(requestId); + marshaller.writeObject(exception); + marshaller.close(); + output.close(); + connection.doBlockingWrite(bufferList); + } finally { + IoUtils.safeClose(output); + } + } finally { + IoUtils.safeClose(marshaller); + } + } + + public void handleCancellation() throws IOException { + final ByteBuffer buffer = ByteBuffer.allocate(5); + buffer.put((byte) MessageType.CANCEL_ACK.getId()); + buffer.putInt(requestId); + buffer.flip(); + connection.doBlockingWrite(buffer); + } + + public String toString() { + return "forwarding reply handler <" + Integer.toHexString(hashCode()) + "> (request id = " + requestId + ") for " + connection; + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexRequestHandler.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,136 +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 org.jboss.remoting.spi.RequestHandler; -import org.jboss.remoting.spi.AbstractAutoCloseable; -import org.jboss.remoting.spi.RemoteRequestContext; -import org.jboss.remoting.spi.ReplyHandler; -import org.jboss.remoting.spi.SpiUtils; -import org.jboss.xnio.BufferAllocator; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.log.Logger; -import org.jboss.marshalling.Marshaller; -import org.jboss.marshalling.ByteOutput; -import java.nio.ByteBuffer; -import java.util.List; -import java.util.ArrayList; -import java.util.concurrent.atomic.AtomicBoolean; -import java.io.IOException; - -/** - * - */ -final class MultiplexRequestHandler extends AbstractAutoCloseable<RequestHandler> implements RequestHandler { - private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.request-handler"); - - private final int identifier; - private final BufferAllocator<ByteBuffer> allocator; - private final MultiplexConnection connection; - - public MultiplexRequestHandler(final int identifier, final MultiplexConnection connection) { - super(connection.getExecutor()); - this.connection = connection; - this.identifier = identifier; - allocator = connection.getAllocator(); - } - - @Override - protected void closeAction() throws IOException { - connection.removeRemoteClient(identifier); - ByteBuffer buffer = allocator.allocate(); - buffer.put((byte) MessageType.CLIENT_CLOSE.getId()); - buffer.putInt(identifier); - buffer.flip(); - connection.doBlockingWrite(buffer); - } - - public RemoteRequestContext receiveRequest(final Object request, final ReplyHandler handler) { - log.trace("Sending outbound request (request id = %d) (type is %s)", request == null ? "null" : request.getClass()); - final List<ByteBuffer> bufferList; - final MultiplexConnection connection = this.connection; - try { - final Marshaller marshaller = connection.getMarshallerFactory().createMarshaller(connection.getMarshallingConfiguration()); - try { - bufferList = new ArrayList<ByteBuffer>(); - final ByteOutput output = new BufferByteOutput(allocator, bufferList); - try { - marshaller.start(output); - marshaller.write(MessageType.REQUEST.getId()); - marshaller.writeInt(identifier); - final int id = connection.nextRequest(); - connection.addRemoteRequest(id, handler); - marshaller.writeInt(id); - marshaller.writeObject(request); - marshaller.close(); - output.close(); - connection.doBlockingWrite(bufferList); - return new RemoteRequestContextImpl(id, connection); - } finally { - IoUtils.safeClose(output); - } - } finally { - IoUtils.safeClose(marshaller); - } - } catch (final IOException t) { - log.trace(t, "receiveRequest failed with an exception"); - SpiUtils.safeHandleException(handler, t); - return SpiUtils.getBlankRemoteRequestContext(); - } - } - - public String toString() { - return "forwarding request handler <" + Integer.toHexString(hashCode()) + "> (id = " + identifier + ") for " + connection; - } -} - -final class RemoteRequestContextImpl implements RemoteRequestContext { - - private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.requesthandler.context"); - - private final int id; - private final MultiplexConnection connection; - private final AtomicBoolean cancelSent = new AtomicBoolean(); - - public RemoteRequestContextImpl(final int id, final MultiplexConnection connection) { - this.id = id; - this.connection = connection; - } - - public void cancel() { - if (! cancelSent.getAndSet(true)) try { - log.trace("Sending cancel request from %s", this); - final ByteBuffer buffer = ByteBuffer.allocate(5); - buffer.put((byte) MessageType.CANCEL_REQUEST.getId()); - buffer.putInt(id); - buffer.flip(); - connection.doBlockingWrite(buffer); - } catch (Throwable t) { - log.warn("Sending cancel request failed: %s", t); - } - } - - public String toString() { - return "remote request context <" + Integer.toHexString(hashCode()) + "> (request id = " + id + ") for " + connection; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexRequestHandler.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,136 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.spi.RequestHandler; +import org.jboss.remoting3.spi.AbstractAutoCloseable; +import org.jboss.remoting3.spi.RemoteRequestContext; +import org.jboss.remoting3.spi.ReplyHandler; +import org.jboss.remoting3.spi.SpiUtils; +import org.jboss.xnio.BufferAllocator; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.log.Logger; +import org.jboss.marshalling.Marshaller; +import org.jboss.marshalling.ByteOutput; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.ArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import java.io.IOException; + +/** + * + */ +final class MultiplexRequestHandler extends AbstractAutoCloseable<RequestHandler> implements RequestHandler { + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.request-handler"); + + private final int identifier; + private final BufferAllocator<ByteBuffer> allocator; + private final MultiplexConnection connection; + + public MultiplexRequestHandler(final int identifier, final MultiplexConnection connection) { + super(connection.getExecutor()); + this.connection = connection; + this.identifier = identifier; + allocator = connection.getAllocator(); + } + + @Override + protected void closeAction() throws IOException { + connection.removeRemoteClient(identifier); + ByteBuffer buffer = allocator.allocate(); + buffer.put((byte) MessageType.CLIENT_CLOSE.getId()); + buffer.putInt(identifier); + buffer.flip(); + connection.doBlockingWrite(buffer); + } + + public RemoteRequestContext receiveRequest(final Object request, final ReplyHandler handler) { + log.trace("Sending outbound request (request id = %d) (type is %s)", request == null ? "null" : request.getClass()); + final List<ByteBuffer> bufferList; + final MultiplexConnection connection = this.connection; + try { + final Marshaller marshaller = connection.getMarshallerFactory().createMarshaller(connection.getMarshallingConfiguration()); + try { + bufferList = new ArrayList<ByteBuffer>(); + final ByteOutput output = new BufferByteOutput(allocator, bufferList); + try { + marshaller.start(output); + marshaller.write(MessageType.REQUEST.getId()); + marshaller.writeInt(identifier); + final int id = connection.nextRequest(); + connection.addRemoteRequest(id, handler); + marshaller.writeInt(id); + marshaller.writeObject(request); + marshaller.close(); + output.close(); + connection.doBlockingWrite(bufferList); + return new RemoteRequestContextImpl(id, connection); + } finally { + IoUtils.safeClose(output); + } + } finally { + IoUtils.safeClose(marshaller); + } + } catch (final IOException t) { + log.trace(t, "receiveRequest failed with an exception"); + SpiUtils.safeHandleException(handler, t); + return SpiUtils.getBlankRemoteRequestContext(); + } + } + + public String toString() { + return "forwarding request handler <" + Integer.toHexString(hashCode()) + "> (id = " + identifier + ") for " + connection; + } +} + +final class RemoteRequestContextImpl implements RemoteRequestContext { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.requesthandler.context"); + + private final int id; + private final MultiplexConnection connection; + private final AtomicBoolean cancelSent = new AtomicBoolean(); + + public RemoteRequestContextImpl(final int id, final MultiplexConnection connection) { + this.id = id; + this.connection = connection; + } + + public void cancel() { + if (! cancelSent.getAndSet(true)) try { + log.trace("Sending cancel request from %s", this); + final ByteBuffer buffer = ByteBuffer.allocate(5); + buffer.put((byte) MessageType.CANCEL_REQUEST.getId()); + buffer.putInt(id); + buffer.flip(); + connection.doBlockingWrite(buffer); + } catch (Throwable t) { + log.warn("Sending cancel request failed: %s", t); + } + } + + public String toString() { + return "remote request context <" + Integer.toHexString(hashCode()) + "> (request id = " + id + ") for " + connection; + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexRequestHandlerSource.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,92 +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 org.jboss.remoting.spi.RequestHandlerSource; -import org.jboss.remoting.spi.AbstractAutoCloseable; -import org.jboss.remoting.spi.RequestHandler; -import org.jboss.remoting.spi.Handle; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.log.Logger; -import java.nio.ByteBuffer; -import java.io.IOException; - -/** - * - */ -final class MultiplexRequestHandlerSource extends AbstractAutoCloseable<RequestHandlerSource> implements RequestHandlerSource { - - private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.request-handler-source"); - - private final int identifier; - private final MultiplexConnection connection; - - MultiplexRequestHandlerSource(final int identifier, final MultiplexConnection connection) { - super(connection.getExecutor()); - this.connection = connection; - this.identifier = identifier; - } - - @Override - protected void closeAction() throws IOException { - ByteBuffer buffer = ByteBuffer.allocate(5); - buffer.put((byte) MessageType.SERVICE_CLOSE_REQUEST.getId()); - buffer.putInt(identifier); - buffer.flip(); - connection.doBlockingWrite(buffer); - } - - public Handle<RequestHandler> createRequestHandler() throws IOException { - log.trace("Creating new request handler from %s", this); - final int id = connection.nextRemoteClient(); - final RequestHandler requestHandler = new MultiplexRequestHandler(id, connection); - boolean ok = false; - try { - connection.addRemoteClient(id, requestHandler); - try { - final ByteBuffer buffer = ByteBuffer.allocate(9); - buffer.put((byte) MessageType.CLIENT_OPEN.getId()); - buffer.putInt(identifier); - buffer.putInt(id); - buffer.flip(); - connection.doBlockingWrite(buffer); - final Handle<RequestHandler> handlerHandle = new MultiplexRequestHandler(id, connection).getHandle(); - log.trace("Created new request handler with a handle of %s", handlerHandle); - ok = true; - return handlerHandle; - } finally { - if (! ok) { - connection.removeRemoteClient(id); - } - } - } finally { - if (! ok) { - IoUtils.safeClose(requestHandler); - } - } - } - - public String toString() { - return "forwarding request handler source <" + Integer.toHexString(hashCode()) + "> (id = " + identifier + ") for " + connection; - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/MultiplexRequestHandlerSource.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/MultiplexRequestHandlerSource.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,92 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.spi.AbstractAutoCloseable; +import org.jboss.remoting3.spi.RequestHandler; +import org.jboss.remoting3.spi.Handle; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.log.Logger; +import java.nio.ByteBuffer; +import java.io.IOException; + +/** + * + */ +final class MultiplexRequestHandlerSource extends AbstractAutoCloseable<RequestHandlerSource> implements RequestHandlerSource { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex.request-handler-source"); + + private final int identifier; + private final MultiplexConnection connection; + + MultiplexRequestHandlerSource(final int identifier, final MultiplexConnection connection) { + super(connection.getExecutor()); + this.connection = connection; + this.identifier = identifier; + } + + @Override + protected void closeAction() throws IOException { + ByteBuffer buffer = ByteBuffer.allocate(5); + buffer.put((byte) MessageType.SERVICE_CLOSE_REQUEST.getId()); + buffer.putInt(identifier); + buffer.flip(); + connection.doBlockingWrite(buffer); + } + + public Handle<RequestHandler> createRequestHandler() throws IOException { + log.trace("Creating new request handler from %s", this); + final int id = connection.nextRemoteClient(); + final RequestHandler requestHandler = new MultiplexRequestHandler(id, connection); + boolean ok = false; + try { + connection.addRemoteClient(id, requestHandler); + try { + final ByteBuffer buffer = ByteBuffer.allocate(9); + buffer.put((byte) MessageType.CLIENT_OPEN.getId()); + buffer.putInt(identifier); + buffer.putInt(id); + buffer.flip(); + connection.doBlockingWrite(buffer); + final Handle<RequestHandler> handlerHandle = new MultiplexRequestHandler(id, connection).getHandle(); + log.trace("Created new request handler with a handle of %s", handlerHandle); + ok = true; + return handlerHandle; + } finally { + if (! ok) { + connection.removeRemoteClient(id); + } + } + } finally { + if (! ok) { + IoUtils.safeClose(requestHandler); + } + } + } + + public String toString() { + return "forwarding request handler source <" + Integer.toHexString(hashCode()) + "> (id = " + identifier + ") for " + connection; + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ServiceConfiguration.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ServiceConfiguration.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/ServiceConfiguration.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,47 @@ +/* + * 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.remoting3.multiplex; + +/** + * A service configuration for a service registered on one side of a connection. + */ +public final class ServiceConfiguration implements Cloneable { + private String marshaller; + + public String getMarshaller() { + return marshaller; + } + + public void setMarshaller(final String marshaller) { + this.marshaller = marshaller; + } + + public ServiceConfiguration clone() { + try { + return (ServiceConfiguration) super.clone(); + } catch (CloneNotSupportedException e) { + // not possible + throw new IllegalStateException(); + } + } +} Deleted: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/SimpleMultiplexHandler.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,71 +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 org.jboss.xnio.DelegatingIoHandler; -import org.jboss.xnio.AbstractIoFuture; -import org.jboss.xnio.IoFuture; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.channels.AllocatedMessageChannel; -import org.jboss.remoting.Endpoint; - -/** - * - */ -public final class SimpleMultiplexHandler extends DelegatingIoHandler<AllocatedMessageChannel> { - - private volatile MultiplexConnection connection; - private final Endpoint endpoint; - private final MultiplexConfiguration configuration; - private final FutureConnection futureConnection = new FutureConnection(); - - public SimpleMultiplexHandler(final Endpoint endpoint, final MultiplexConfiguration configuration) { - this.endpoint = endpoint; - this.configuration = configuration; - } - - public void handleOpened(final AllocatedMessageChannel channel) { - connection = new MultiplexConnection(endpoint, channel, configuration); - futureConnection.setResult(connection); - setReadHandler(new MultiplexReadHandler(connection)); - channel.resumeReads(); - } - - public void handleClosed(final AllocatedMessageChannel channel) { - IoUtils.safeClose(connection); - } - - public IoFuture<MultiplexConnection> getConnection() { - return futureConnection; - } - - public static final class FutureConnection extends AbstractIoFuture<MultiplexConnection> { - public IoFuture<MultiplexConnection> cancel() { - return this; - } - - protected boolean setResult(final MultiplexConnection result) { - return super.setResult(result); - } - } -} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol/multiplex/SimpleMultiplexHandler.java) =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SimpleMultiplexHandler.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,68 @@ +/* + * 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.remoting3.multiplex; + +import org.jboss.xnio.DelegatingIoHandler; +import org.jboss.xnio.AbstractIoFuture; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.channels.AllocatedMessageChannel; + +/** + * + */ +public final class SimpleMultiplexHandler extends DelegatingIoHandler<AllocatedMessageChannel> { + + private volatile MultiplexConnection connection; + private final MultiplexConfiguration configuration; + private final FutureConnection futureConnection = new FutureConnection(); + + public SimpleMultiplexHandler(final MultiplexConfiguration configuration) { + this.configuration = configuration; + } + + public void handleOpened(final AllocatedMessageChannel channel) { + connection = new MultiplexConnection(channel, configuration); + futureConnection.setResult(connection); + setReadHandler(new MultiplexReadHandler(connection)); + channel.resumeReads(); + } + + public void handleClosed(final AllocatedMessageChannel channel) { + IoUtils.safeClose(connection); + } + + public IoFuture<MultiplexConnection> getConnection() { + return futureConnection; + } + + public static final class FutureConnection extends AbstractIoFuture<MultiplexConnection> { + public IoFuture<MultiplexConnection> cancel() { + return this; + } + + protected boolean setResult(final MultiplexConnection result) { + return super.setResult(result); + } + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SynchronizedSet.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SynchronizedSet.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/SynchronizedSet.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,150 @@ +/* + * 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.remoting3.multiplex; + +import java.util.Set; +import java.util.Iterator; +import java.util.Collection; + +/** + * + */ +final class SynchronizedSet<E> implements Set<E> { + private final Set<E> delegate; + private final Object lock; + + public SynchronizedSet(final Set<E> delegate, final Object lock) { + this.delegate = delegate; + this.lock = lock; + } + + public int size() { + synchronized (lock) { + return delegate.size(); + } + } + + public boolean isEmpty() { + synchronized (lock) { + return delegate.isEmpty(); + } + } + + public boolean contains(final Object o) { + synchronized (lock) { + return delegate.contains(o); + } + } + + public Iterator<E> iterator() { + synchronized (lock) { + final Iterator<E> i = delegate.iterator(); + return new Iterator<E>() { + public boolean hasNext() { + synchronized (lock) { + return i.hasNext(); + } + } + + public E next() { + synchronized (lock) { + return i.next(); + } + } + + public void remove() { + synchronized (lock) { + i.remove(); + } + } + }; + } + } + + public Object[] toArray() { + synchronized (lock) { + return delegate.toArray(); + } + } + + @SuppressWarnings({ "SuspiciousToArrayCall" }) + public <T> T[] toArray(final T[] a) { + synchronized (lock) { + return delegate.<T>toArray(a); + } + } + + public boolean add(final E e) { + synchronized (lock) { + return delegate.add(e); + } + } + + public boolean remove(final Object o) { + synchronized (lock) { + return delegate.remove(o); + } + } + + public boolean containsAll(final Collection<?> c) { + synchronized (lock) { + return delegate.containsAll(c); + } + } + + public boolean addAll(final Collection<? extends E> c) { + synchronized (lock) { + return delegate.addAll(c); + } + } + + public boolean retainAll(final Collection<?> c) { + synchronized (lock) { + return delegate.retainAll(c); + } + } + + public boolean removeAll(final Collection<?> c) { + synchronized (lock) { + return delegate.removeAll(c); + } + } + + public void clear() { + synchronized (lock) { + delegate.clear(); + } + } + + public boolean equals(final Object o) { + synchronized (lock) { + return delegate.equals(o); + } + } + + public int hashCode() { + synchronized (lock) { + return delegate.hashCode(); + } + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/impl/AbstractConnectionServiceProvider.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/impl/AbstractConnectionServiceProvider.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/impl/AbstractConnectionServiceProvider.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,143 @@ +/* + * 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.remoting3.multiplex.impl; + +import org.jboss.remoting3.multiplex.ConnectionServiceProvider; +import org.jboss.remoting3.multiplex.ServiceConfiguration; +import org.jboss.remoting3.multiplex.ConnectionServiceListener; +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.spi.Handle; +import org.jboss.remoting3.spi.AbstractAutoCloseable; +import org.jboss.remoting3.spi.AbstractSimpleCloseable; +import org.jboss.remoting3.QualifiedName; +import org.jboss.remoting3.RemoteServiceConfiguration; +import org.jboss.remoting3.SimpleCloseable; +import org.jboss.marshalling.MarshallerFactory; +import org.jboss.marshalling.MarshallingConfiguration; +import org.jboss.xnio.log.Logger; +import java.io.IOException; +import java.util.concurrent.Executor; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.Map; +import java.util.HashMap; +import java.util.Set; +import java.util.HashSet; + +/** + * + */ +public abstract class AbstractConnectionServiceProvider<T> extends AbstractAutoCloseable<T> implements ConnectionServiceProvider { + + private static final Logger log = Logger.getLogger("org.jboss.remoting.multiplex"); + + private final Executor executor; + private final ConnectionServiceProvider parent; + private final Lock registrationLock = new ReentrantLock(); + private final Map<Object, ListenerRegistration> listenerRegistrations = new HashMap<Object, ListenerRegistration>(); + private final Set<ServiceRegistration> serviceRegistrations = new HashSet<ServiceRegistration>(); + + protected AbstractConnectionServiceProvider(final Executor executor) { + super(executor); + this.executor = executor; + parent = null; + } + + protected AbstractConnectionServiceProvider(final Executor executor, final ConnectionServiceProvider parent) { + super(executor); + this.executor = executor; + this.parent = parent; + } + + public Handle<RequestHandlerSource> registerService(final QualifiedName path, final RemoteServiceConfiguration remoteServiceConfiguration, final ServiceConfiguration config) throws IOException { + return null; + } + + // TODO - need a way to remotely convey the marshalling impl and configuration + public SimpleCloseable addServiceListener(final ConnectionServiceListener listener) { + final Object key = new Object(); + final ListenerRegistration registration = new ListenerRegistration(executor, key, listener); + registrationLock.lock(); + try { + listenerRegistrations.put(key, registration); + for (ServiceRegistration serviceRegistration : serviceRegistrations) { + handleRegistration(listener, registration, serviceRegistration); + } + } finally { + registrationLock.unlock(); + } + return registration; + } + + private void handleRegistration(ConnectionServiceListener listener, SimpleCloseable handle, ServiceRegistration registration) { + final ConnectionServiceListener.ServiceInfo info = new ConnectionServiceListener.ServiceInfo(); + info.setEndpointName(registration.endpointName); + info.setGroupName(registration.groupName); + info.setMetric(registration.metric); + info.setPath(registration.path); + info.setRequestHandlerSource(registration.requestHandlerSource); + info.setServiceType(registration.serviceType); + try { + listener.serviceRegistered(handle, info); + } catch (Throwable t) { + log.error(t, "Connection service listener threw an exception"); + } + } + + private final class ListenerRegistration extends AbstractSimpleCloseable { + private final Object key; + private final ConnectionServiceListener listener; + + private ListenerRegistration(final Executor executor, final Object key, final ConnectionServiceListener listener) { + super(executor); + this.key = key; + this.listener = listener; + } + + protected void closeAction() throws IOException { + listenerRegistrations.remove(key); + } + } + + private final class ServiceRegistration { + private final String serviceType; + private final String groupName; + private final String endpointName; + private final Handle<RequestHandlerSource> requestHandlerSource; + private final int metric; + private final MarshallerFactory marshallerFactory; + private final MarshallingConfiguration marshallingConfiguration; + private final QualifiedName path; + + private ServiceRegistration(RemoteServiceConfiguration remoteServiceConfiguration, ServiceConfiguration serviceConfiguration, final QualifiedName path) throws IOException { + this.path = path; + serviceType = remoteServiceConfiguration.getServiceType(); + groupName = remoteServiceConfiguration.getGroupName(); + endpointName = remoteServiceConfiguration.getEndpointName(); + requestHandlerSource = remoteServiceConfiguration.getRequestHandlerSource().getHandle(); + metric = remoteServiceConfiguration.getMetric(); + marshallerFactory = serviceConfiguration.getMarshallerFactory(); + marshallingConfiguration = serviceConfiguration.getMarshallingConfiguration(); + } + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/SAClassTable.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/SAClassTable.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/SAClassTable.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,81 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ + +package org.jboss.remoting3.multiplex.sa; + +import org.jboss.marshalling.ClassTable; +import org.jboss.marshalling.Unmarshaller; +import org.jboss.marshalling.Marshaller; +import org.jboss.remoting3.QualifiedName; +import java.io.IOException; +import java.util.Map; +import java.util.IdentityHashMap; +import java.util.Collections; +import java.util.List; +import java.util.ArrayList; + +/** + * + */ +public final class SAClassTable implements ClassTable { + + private static final Map<Class<?>, Writer> writerMap; + private static final List<Class<?>> readerList; + + private static void add(Map<Class<?>, Writer> map, List<Class<?>> list, int id, Class<?> clazz) { + map.put(clazz, new IdWriter(id)); + list.add(id, clazz); + } + + static { + Map<Class<?>, Writer> map = new IdentityHashMap<Class<?>, Writer>(); + List<Class<?>> list = new ArrayList<Class<?>>(); + + // These are the classes that comprise messages for this service + add(map, list, 0, ServiceAdvertisement.class); + add(map, list, 1, ServiceAdvertisementReply.class); + add(map, list, 2, QualifiedName.class); + + writerMap = Collections.unmodifiableMap(map); + readerList = Collections.unmodifiableList(list); + } + + private static final class IdWriter implements Writer { + private final byte id; + + public IdWriter(final int id) { + this.id = (byte) id; + } + + public void writeClass(final Marshaller marshaller, final Class<?> clazz) throws IOException { + marshaller.writeByte(id); + } + } + + public Writer getClassWriter(final Class<?> clazz) throws IOException { + return writerMap.get(clazz); + } + + public Class<?> readClass(final Unmarshaller unmarshaller) throws IOException, ClassNotFoundException { + return readerList.get(unmarshaller.read()); + } +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisement.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisement.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisement.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,30 @@ +/* + * 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.remoting3.multiplex.sa; + +/** + * + */ +public final class ServiceAdvertisement { + +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementReply.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementReply.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementReply.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,30 @@ +/* + * 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.remoting3.multiplex.sa; + +/** + * + */ +public final class ServiceAdvertisementReply { + +} Added: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementRequestListener.java =================================================================== --- remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementRequestListener.java (rev 0) +++ remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/multiplex/sa/ServiceAdvertisementRequestListener.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,37 @@ +/* + * 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.remoting3.multiplex.sa; + +import org.jboss.remoting3.RequestContext; +import org.jboss.remoting3.RemoteExecutionException; +import org.jboss.remoting3.AbstractRequestListener; + +/** + * + */ +public final class ServiceAdvertisementRequestListener extends AbstractRequestListener<ServiceAdvertisement, ServiceAdvertisementReply> { + + public void handleRequest(final RequestContext<ServiceAdvertisementReply> context, final ServiceAdvertisement request) throws RemoteExecutionException { + + } +} Copied: remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting3/protocol (from rev 4858, remoting3-multiplex/trunk/main/src/main/java/org/jboss/remoting/protocol) Copied: remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3 (from rev 4848, remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting) Copied: remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex (from rev 4848, remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting/protocol/multiplex) Deleted: remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java =================================================================== --- remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting/protocol/multiplex/ConnectionTestCase.java 2009-01-29 20:44:43 UTC (rev 4848) +++ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -1,181 +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.nio.ByteBuffer; -import java.util.Collections; -import java.util.LinkedList; -import java.util.List; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.CountDownLatch; -import java.io.IOException; -import junit.framework.TestCase; -import org.jboss.remoting.core.EndpointImpl; -import org.jboss.remoting.test.support.LoggingHelper; -import org.jboss.remoting.LocalServiceConfiguration; -import org.jboss.remoting.RequestListener; -import org.jboss.remoting.ClientContext; -import org.jboss.remoting.ServiceContext; -import org.jboss.remoting.RequestContext; -import org.jboss.remoting.RemoteExecutionException; -import org.jboss.remoting.ClientSource; -import org.jboss.remoting.Client; -import org.jboss.remoting.QualifiedName; -import org.jboss.remoting.spi.NamedServiceRegistry; -import org.jboss.remoting.spi.RequestHandlerSource; -import org.jboss.remoting.spi.Handle; -import org.jboss.xnio.BufferAllocator; -import org.jboss.xnio.IoUtils; -import org.jboss.xnio.Xnio; -import org.jboss.xnio.CloseableExecutor; -import org.jboss.xnio.ChannelSource; -import org.jboss.xnio.IoHandlerFactory; -import org.jboss.xnio.IoFuture; -import org.jboss.xnio.Buffers; -import org.jboss.xnio.log.Logger; -import org.jboss.xnio.channels.AllocatedMessageChannel; -import org.jboss.xnio.channels.Channels; -import org.jboss.xnio.nio.NioXnio; -import org.jboss.river.RiverMarshallerFactory; -import org.jboss.marshalling.MarshallingConfiguration; - -/** - * - */ -public final class ConnectionTestCase extends TestCase { - static { - LoggingHelper.init(); - } - - public static final Logger log = Logger.getLogger(ConnectionTestCase.class.getSimpleName()); - - public void testConnection() throws Throwable { - final String REQUEST = "request"; - final String REPLY = "reply"; - final List<Throwable> problems = Collections.synchronizedList(new LinkedList<Throwable>()); - final CloseableExecutor closeableExecutor = IoUtils.closeableExecutor(Executors.newCachedThreadPool(), 500L, TimeUnit.MILLISECONDS); - try { - final BufferAllocator<ByteBuffer> allocator = Buffers.createHeapByteBufferAllocator(1024); - final Xnio xnio = NioXnio.create(); - try { - final EndpointImpl remoteEndpoint = new EndpointImpl(closeableExecutor, "left-side"); - try { - final EndpointImpl endpoint = new EndpointImpl(closeableExecutor, "right-side"); - try { - final CountDownLatch latch = new CountDownLatch(1); - final MultiplexConfiguration configuration = new MultiplexConfiguration(); - configuration.setAllocator(allocator); - configuration.setExecutor(closeableExecutor); - configuration.setLinkMetric(10); - configuration.setMarshallerFactory(new RiverMarshallerFactory()); - final NamedServiceRegistry registry = new NamedServiceRegistry(); - configuration.setNamedServiceRegistry(registry); - final MarshallingConfiguration marshallingConfiguration = new MarshallingConfiguration(); - configuration.setMarshallingConfiguration(marshallingConfiguration); - final LocalServiceConfiguration<Object, Object> localServiceConfiguration = new LocalServiceConfiguration<Object, Object>(new RequestListener<Object, Object>() { - public void handleClientOpen(final ClientContext context) { - log.debug("Client open"); - } - - public void handleServiceOpen(final ServiceContext context) { - } - - public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException { - try { - context.sendReply(REPLY); - } catch (IOException e) { - log.error(e, "Failed to send reply"); - problems.add(e); - } - } - - public void handleServiceClose(final ServiceContext context) { - } - - public void handleClientClose(final ClientContext context) { - log.debug("Client closed"); - latch.countDown(); - } - - public String toString() { - return "TestListener"; - } - }, Object.class, Object.class); - localServiceConfiguration.setServiceType("connection.test"); - localServiceConfiguration.setGroupName("testgroup"); - localServiceConfiguration.setMetric(10); - final Handle<RequestHandlerSource> requestHandlerSourceHandle = remoteEndpoint.registerService(localServiceConfiguration); - try { - registry.registerService(QualifiedName.parse("/test/connectiontest"), requestHandlerSourceHandle.getResource()); - final IoHandlerFactory<AllocatedMessageChannel> handlerFactory = MultiplexProtocol.createServer(remoteEndpoint, configuration); - final ChannelSource<AllocatedMessageChannel> channelSource = Channels.convertStreamToAllocatedMessage(xnio.createPipeServer(Channels.convertStreamToAllocatedMessage(handlerFactory, 16384, 16384)), 16384, 16384); - final IoFuture<MultiplexConnection> future = MultiplexProtocol.connect(endpoint, configuration, channelSource); - final MultiplexConnection connection = future.get(); - try { - final Handle<RequestHandlerSource> remoteHandlerSource = connection.openRemoteService(QualifiedName.parse("/test/connectiontest")); - try { - final ClientSource<Object, Object> clientSource = endpoint.createClientSource(remoteHandlerSource.getResource(), Object.class, Object.class); - try { - final Client<Object,Object> client = clientSource.createClient(); - try { - final IoFuture<Object> futureReply = client.send(REQUEST); - assertEquals(IoFuture.Status.DONE, futureReply.await(1L, TimeUnit.SECONDS)); - assertEquals(REPLY, futureReply.get()); - client.close(); - clientSource.close(); - remoteHandlerSource.close(); - connection.close(); - assertTrue(latch.await(1L, TimeUnit.SECONDS)); - } finally { - IoUtils.safeClose(client); - } - } finally { - IoUtils.safeClose(clientSource); - } - } finally { - IoUtils.safeClose(remoteHandlerSource); - } - } finally { - IoUtils.safeClose(connection); - } - } finally { - IoUtils.safeClose(requestHandlerSourceHandle); - } - } finally { - IoUtils.safeClose(endpoint); - } - } finally { - IoUtils.safeClose(remoteEndpoint); - } - } finally { - IoUtils.safeClose(xnio); - } - } finally { - IoUtils.safeClose(closeableExecutor); - } - for (Throwable t : problems) { - throw t; - } - } -} Copied: remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java (from rev 4858, remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting/protocol/multiplex/ConnectionTestCase.java) =================================================================== --- remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java (rev 0) +++ remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/multiplex/ConnectionTestCase.java 2009-02-26 21:52:06 UTC (rev 4859) @@ -0,0 +1,178 @@ +/* + * 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.remoting3.multiplex; + +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.CountDownLatch; +import java.io.IOException; +import junit.framework.TestCase; +import org.jboss.remoting3.LocalServiceConfiguration; +import org.jboss.remoting3.RequestListener; +import org.jboss.remoting3.ClientContext; +import org.jboss.remoting3.ServiceContext; +import org.jboss.remoting3.RequestContext; +import org.jboss.remoting3.RemoteExecutionException; +import org.jboss.remoting3.ClientSource; +import org.jboss.remoting3.Client; +import org.jboss.remoting3.QualifiedName; +import org.jboss.remoting3.Endpoint; +import org.jboss.remoting3.Remoting; +import org.jboss.remoting3.spi.NamedServiceRegistry; +import org.jboss.remoting3.spi.RequestHandlerSource; +import org.jboss.remoting3.spi.Handle; +import org.jboss.xnio.BufferAllocator; +import org.jboss.xnio.IoUtils; +import org.jboss.xnio.Xnio; +import org.jboss.xnio.CloseableExecutor; +import org.jboss.xnio.ChannelSource; +import org.jboss.xnio.IoHandlerFactory; +import org.jboss.xnio.IoFuture; +import org.jboss.xnio.Buffers; +import org.jboss.xnio.log.Logger; +import org.jboss.xnio.channels.AllocatedMessageChannel; +import org.jboss.xnio.channels.Channels; +import org.jboss.xnio.nio.NioXnio; +import org.jboss.marshalling.MarshallingConfiguration; +import org.jboss.marshalling.river.RiverMarshallerFactory; + +/** + * + */ +public final class ConnectionTestCase extends TestCase { + + public static final Logger log = Logger.getLogger(ConnectionTestCase.class.getSimpleName()); + + public void testConnection() throws Throwable { + final String REQUEST = "request"; + final String REPLY = "reply"; + final List<Throwable> problems = Collections.synchronizedList(new LinkedList<Throwable>()); + final CloseableExecutor closeableExecutor = IoUtils.closeableExecutor(Executors.newCachedThreadPool(), 500L, TimeUnit.MILLISECONDS); + try { + final BufferAllocator<ByteBuffer> allocator = Buffers.createHeapByteBufferAllocator(1024); + final Xnio xnio = NioXnio.create(); + try { + final Endpoint remoteEndpoint = Remoting.createEndpoint(closeableExecutor, "left-side"); + try { + final Endpoint endpoint = Remoting.createEndpoint(closeableExecutor, "right-side"); + try { + final CountDownLatch latch = new CountDownLatch(1); + final MultiplexConfiguration configuration = new MultiplexConfiguration(); + configuration.setAllocator(allocator); + configuration.setExecutor(closeableExecutor); + configuration.setLinkMetric(10); + configuration.setMarshallerFactory(new RiverMarshallerFactory()); + final NamedServiceRegistry registry = new NamedServiceRegistry(); + configuration.setNamedServiceRegistry(registry); + final MarshallingConfiguration marshallingConfiguration = new MarshallingConfiguration(); + configuration.setMarshallingConfiguration(marshallingConfiguration); + final LocalServiceConfiguration<Object, Object> localServiceConfiguration = new LocalServiceConfiguration<Object, Object>(new RequestListener<Object, Object>() { + public void handleClientOpen(final ClientContext context) { + log.debug("Client open"); + } + + public void handleServiceOpen(final ServiceContext context) { + } + + public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException { + try { + context.sendReply(REPLY); + } catch (IOException e) { + log.error(e, "Failed to send reply"); + problems.add(e); + } + } + + public void handleServiceClose(final ServiceContext context) { + } + + public void handleClientClose(final ClientContext context) { + log.debug("Client closed"); + latch.countDown(); + } + + public String toString() { + return "TestListener"; + } + }, Object.class, Object.class); + localServiceConfiguration.setServiceType("connection.test"); + localServiceConfiguration.setGroupName("testgroup"); + localServiceConfiguration.setMetric(10); + final Handle<RequestHandlerSource> requestHandlerSourceHandle = remoteEndpoint.registerService(localServiceConfiguration); + try { + registry.registerService(QualifiedName.parse("/test/connectiontest"), requestHandlerSourceHandle.getResource()); + final IoHandlerFactory<AllocatedMessageChannel> handlerFactory = MultiplexProtocol.createServer(configuration); + final ChannelSource<AllocatedMessageChannel> channelSource = Channels.convertStreamToAllocatedMessage(xnio.createPipeServer(Channels.convertStreamToAllocatedMessage(handlerFactory, 16384, 16384)), 16384, 16384); + final IoFuture<MultiplexConnection> future = MultiplexProtocol.connect(configuration, channelSource); + final MultiplexConnection connection = future.get(); + try { + final Handle<RequestHandlerSource> remoteHandlerSource = connection.openRemoteService(QualifiedName.parse("/test/connectiontest")); + try { + final ClientSource<Object, Object> clientSource = endpoint.createClientSource(remoteHandlerSource.getResource(), Object.class, Object.class); + try { + final Client<Object,Object> client = clientSource.createClient(); + try { + final IoFuture<? extends Object> futureReply = client.send(REQUEST); + assertEquals(IoFuture.Status.DONE, futureReply.await(1L, TimeUnit.SECONDS)); + assertEquals(REPLY, futureReply.get()); + client.close(); + clientSource.close(); + remoteHandlerSource.close(); + connection.close(); + assertTrue(latch.await(1L, TimeUnit.SECONDS)); + } finally { + IoUtils.safeClose(client); + } + } finally { + IoUtils.safeClose(clientSource); + } + } finally { + IoUtils.safeClose(remoteHandlerSource); + } + } finally { + IoUtils.safeClose(connection); + } + } finally { + IoUtils.safeClose(requestHandlerSourceHandle); + } + } finally { + IoUtils.safeClose(endpoint); + } + } finally { + IoUtils.safeClose(remoteEndpoint); + } + } finally { + IoUtils.safeClose(xnio); + } + } finally { + IoUtils.safeClose(closeableExecutor); + } + for (Throwable t : problems) { + throw t; + } + } +} Copied: remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting3/protocol (from rev 4858, remoting3-multiplex/trunk/main/src/test/java/org/jboss/remoting/protocol)