Author: david.lloyd(a)jboss.com Date: 2010-03-03 11:54:23 -0500 (Wed, 03 Mar 2010) New Revision: 5792 Modified: remoting3/trunk/jboss-remoting/pom.xml remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/EndpointImpl.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/LocalRequestHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RemoteClassTable.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Remoting.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RequestListener.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractClientMessageHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractMessageHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientAuthenticationHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientGreetingHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientOpenListener.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/GreetingUtils.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnection.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionProvider.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteProtocol.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerAuthenticationHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerGreetingHandler.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerOpenListener.java remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/EndpointTestCase.java remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/InvocationTestBase.java remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/StringRot13RequestListener.java remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/socket/SocketUsageExamples.java Log: JBREM-1201, remove handleClose; JBRAM-1199 add marshaller version to negotiation; also fix some problems in the close path and some minor marshalling issues Modified: remoting3/trunk/jboss-remoting/pom.xml =================================================================== --- remoting3/trunk/jboss-remoting/pom.xml 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/pom.xml 2010-03-03 16:54:23 UTC (rev 5792) @@ -31,8 +31,8 @@ <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> - <xnio.version>2.1.0.CR1</xnio.version> - <jbmar.version>1.3.0.CR1</jbmar.version> + <xnio.version>2.1.0.CR2-SNAPSHOT</xnio.version> + <jbmar.version>1.3.0.CR2-SNAPSHOT</jbmar.version> </properties> <groupId>org.jboss.remoting</groupId> @@ -54,15 +54,15 @@ </dependency> <dependency> <groupId>org.jboss.marshalling</groupId> - <artifactId>marshalling-api</artifactId> + <artifactId>jboss-marshalling</artifactId> <version>${jbmar.version}</version> <scope>compile</scope> </dependency> <dependency> <groupId>org.jboss.marshalling</groupId> - <artifactId>river</artifactId> + <artifactId>jboss-marshalling-river</artifactId> <version>${jbmar.version}</version> - <scope>test</scope> + <scope>compile</scope> </dependency> <dependency> <groupId>org.testng</groupId> @@ -90,6 +90,10 @@ <name>java.util.logging.manager</name> <value>org.jboss.logmanager.LogManager</value> </property> + <property> + <name>jboss.remoting.leakdebugging</name> + <value>true</value> + </property> </systemProperties> </configuration> </plugin> Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/EndpointImpl.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/EndpointImpl.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/EndpointImpl.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -612,6 +612,7 @@ if (connectionProviders.putIfAbsent(uriScheme, provider) != null) { throw new DuplicateRegistrationException("URI scheme '" + uriScheme + "' is already registered to a provider"); } + log.trace("Adding registration for connection provider named %s: %s", name, provider); final Registration handle = new MapRegistration<ConnectionProvider>(connectionProviders, uriScheme, provider); return handle; } @@ -643,6 +644,7 @@ if (map.putIfAbsent(name, provider) != null) { throw new DuplicateRegistrationException(type.getDescription() + " '" + name + "' is already registered"); } + log.trace("Adding registration for %s named %s: %s", type, name, provider); return new MapRegistration<T>(map, name, provider); } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/LocalRequestHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/LocalRequestHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/LocalRequestHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -90,11 +90,7 @@ } protected void closeAction() throws IOException { - try { - requestListener.handleClose(); - } catch (Throwable t) { - log.error(t, "Unexpected exception in request listener handleClose() method"); - } + clientContext.close(); } public String toString() { Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RemoteClassTable.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RemoteClassTable.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RemoteClassTable.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -168,9 +168,5 @@ // no action necessary. } } - - public void handleClose() { - // --- - } } } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Remoting.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Remoting.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Remoting.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -25,6 +25,8 @@ import java.io.IOException; import java.io.InputStreamReader; import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.Executor; import java.util.concurrent.Executors; @@ -160,6 +162,7 @@ addServices(endpoint, ProtocolServiceType.CLASS_RESOLVER, props); addServices(endpoint, ProtocolServiceType.OBJECT_RESOLVER, props); addServices(endpoint, ProtocolServiceType.CLASS_EXTERNALIZER_FACTORY, props); + final List<RemotingServiceDescriptor<?>> connectionProviders = new ArrayList<RemotingServiceDescriptor<?>>(); for (RemotingServiceDescriptor<?> descriptor : ServiceLoader.load(RemotingServiceDescriptor.class)) { final String name = descriptor.getName(); final Class<?> serviceType = descriptor.getType(); @@ -171,7 +174,7 @@ } try { if (serviceType == ConnectionProviderFactory.class) { - endpoint.addConnectionProvider(name, (ConnectionProviderFactory) service); + connectionProviders.add(descriptor); } else if (serviceType == ClassTable.class) { endpoint.addProtocolService(ProtocolServiceType.CLASS_TABLE, name, (ClassTable) service); } else if (serviceType == ObjectTable.class) { @@ -184,7 +187,7 @@ endpoint.addProtocolService(ProtocolServiceType.CLASS_EXTERNALIZER_FACTORY, name, (ClassExternalizerFactory) service); } } catch (DuplicateRegistrationException e) { - log.warn("Duplicate registration for '" + name + "' of " + serviceType); + log.warn("Duplicate registration for %s '%s'", serviceType, name); } } final Map<String, ProviderDescriptor> found = new HashMap<String, ProviderDescriptor>(); @@ -199,9 +202,20 @@ try { endpoint.addProtocolService(ProtocolServiceType.MARSHALLER_PROVIDER_DESCRIPTOR, name, found.get(name)); } catch (DuplicateRegistrationException e) { - log.warn("Duplicate registration for '" + name + "' of " + MarshallerFactory.class); + log.warn("Duplicate registration for marshaller factory '%s'", name); } } + // last but not least, install all the protocol service providers which may depend on prior items + for (RemotingServiceDescriptor<?> descriptor : connectionProviders) { + final String name = descriptor.getName(); + try { + endpoint.addConnectionProvider(name, (ConnectionProviderFactory) descriptor.getService(props)); + } catch (IOException e) { + throw new RuntimeException(e); + } catch (DuplicateRegistrationException e) { + log.warn("Duplicate registration for connection provider '%s'", name); + } + } ok = true; return endpoint; } finally { Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RequestListener.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RequestListener.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/RequestListener.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -46,10 +46,4 @@ * @throws RemoteExecutionException if the execution failed in some way */ void handleRequest(RequestContext<O> context, I request) throws RemoteExecutionException; - - /** - * Handle the client closing. Free up any resources. This method is called after the close has occurred, - * so exceptions thrown will be ignored. - */ - void handleClose(); } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractClientMessageHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractClientMessageHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractClientMessageHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -23,6 +23,7 @@ package org.jboss.remoting3.remote; import java.io.IOException; +import java.nio.channels.ClosedChannelException; import org.jboss.remoting3.spi.ConnectionHandlerFactory; import org.jboss.xnio.Result; @@ -35,6 +36,9 @@ } public void handleException(final IOException e) { + if (e instanceof ClosedChannelException) { + return; + } result.setException(e); super.handleException(e); } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractMessageHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractMessageHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/AbstractMessageHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -23,6 +23,7 @@ package org.jboss.remoting3.remote; import java.io.IOException; +import java.nio.channels.ClosedChannelException; import org.jboss.xnio.IoUtils; import org.jboss.xnio.channels.MessageHandler; @@ -36,14 +37,18 @@ public void handleEof() { try { remoteConnection.getChannel().shutdownReads(); + return; } catch (IOException e) { RemoteConnectionHandler.log.trace(e, "Failed to shut down reads for %s", remoteConnection); + IoUtils.safeClose(remoteConnection); } - remoteConnection.readDone(); - IoUtils.safeClose(remoteConnection); } public void handleException(final IOException e) { + if (e instanceof ClosedChannelException) { + // ignore; just means there was a race. + return; + } RemoteConnectionHandler.log.trace(e, "Received exception from %s", remoteConnection); IoUtils.safeClose(remoteConnection); } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientAuthenticationHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientAuthenticationHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientAuthenticationHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -25,7 +25,6 @@ import java.io.IOException; import java.nio.ByteBuffer; import org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.Marshalling; import org.jboss.remoting3.CloseHandler; import org.jboss.remoting3.spi.ConnectionHandler; import org.jboss.remoting3.spi.ConnectionHandlerContext; @@ -125,7 +124,7 @@ factoryResult.setResult(new ConnectionHandlerFactory() { public ConnectionHandler createInstance(final ConnectionHandlerContext connectionContext) { // this happens immediately. - final MarshallerFactory marshallerFactory = Marshalling.getMarshallerFactory("river"); + final MarshallerFactory marshallerFactory = remoteConnection.getProviderDescriptor().getMarshallerFactory(); final RemoteConnectionHandler connectionHandler = new RemoteConnectionHandler(connectionContext, remoteConnection, marshallerFactory); remoteConnection.addCloseHandler(new CloseHandler<Object>() { public void handleClose(final Object closed) { Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientGreetingHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientGreetingHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientGreetingHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -27,9 +27,11 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import org.jboss.remoting3.ProtocolException; import org.jboss.remoting3.RemotingOptions; import org.jboss.remoting3.spi.ConnectionHandlerFactory; import org.jboss.xnio.Buffers; +import org.jboss.xnio.IoUtils; import org.jboss.xnio.OptionMap; import org.jboss.xnio.Result; @@ -53,33 +55,56 @@ public void handleMessage(final ByteBuffer buffer) { List<String> saslMechs = new ArrayList<String>(); String remoteEndpointName = "endpoint"; + final int[] ourVersions = connection.getProviderDescriptor().getSupportedVersions(); + int bestVersion = -1; switch (buffer.get()) { case RemoteProtocol.GREETING: { + RemoteConnectionHandler.log.warn("Client received greeting message"); while (buffer.hasRemaining()) { final byte type = buffer.get(); final int len = buffer.get() & 0xff; + final ByteBuffer data = Buffers.slice(buffer, len); switch (type) { case RemoteProtocol.GREETING_VERSION: { // We only support version zero, so knowing the other side's version is not useful presently - buffer.get(); - if (len > 1) Buffers.skip(buffer, len - 1); break; } case RemoteProtocol.GREETING_SASL_MECH: { - saslMechs.add(Buffers.getModifiedUtf8(Buffers.slice(buffer, len))); + saslMechs.add(Buffers.getModifiedUtf8(data)); break; } case RemoteProtocol.GREETING_ENDPOINT_NAME: { - remoteEndpointName = Buffers.getModifiedUtf8(Buffers.slice(buffer, len)); + remoteEndpointName = Buffers.getModifiedUtf8(data); break; } + case RemoteProtocol.GREETING_MARSHALLER_VERSION: { + final int remoteVersion = data.getInt(); + // is it better than the best one? if not, don't bother + if (remoteVersion <= bestVersion) { + break; + } + // do we support it? if not, skip + for (int ourVersion : ourVersions) { + if (ourVersion == remoteVersion) { + bestVersion = remoteVersion; + break; + } + } + break; + } default: { // unknown, skip it for forward compatibility. - Buffers.skip(buffer, len); break; } } } + // Check the greeting + if (bestVersion == -1) { + // no matches. + factoryResult.setException(new ProtocolException("No matching Marshalling versions could be found")); + IoUtils.safeClose(connection); + return; + } // OK now send our authentication request final OptionMap optionMap = connection.getOptionMap(); final String userName = optionMap.get(RemotingOptions.AUTH_USER_NAME); Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientOpenListener.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientOpenListener.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ClientOpenListener.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -25,6 +25,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; +import org.jboss.marshalling.ProviderDescriptor; import org.jboss.remoting3.spi.ConnectionHandlerFactory; import org.jboss.remoting3.spi.ConnectionProviderContext; import org.jboss.xnio.ChannelListener; @@ -42,12 +43,14 @@ private final ConnectionProviderContext connectionProviderContext; private final Result<ConnectionHandlerFactory> factoryResult; private final CallbackHandler callbackHandler; + private final ProviderDescriptor providerDescriptor; - ClientOpenListener(final OptionMap optionMap, final ConnectionProviderContext connectionProviderContext, final Result<ConnectionHandlerFactory> factoryResult, final CallbackHandler callbackHandler) { + ClientOpenListener(final OptionMap optionMap, final ConnectionProviderContext connectionProviderContext, final Result<ConnectionHandlerFactory> factoryResult, final CallbackHandler callbackHandler, final ProviderDescriptor providerDescriptor) { this.optionMap = optionMap; this.connectionProviderContext = connectionProviderContext; this.factoryResult = factoryResult; this.callbackHandler = callbackHandler; + this.providerDescriptor = providerDescriptor; } public void handleEvent(final ConnectedStreamChannel<InetSocketAddress> channel) { @@ -56,12 +59,17 @@ } catch (IOException e) { // ignore } - final RemoteConnection connection = new RemoteConnection(connectionProviderContext.getExecutor(), channel, optionMap); - + final RemoteConnection connection = new RemoteConnection(connectionProviderContext.getExecutor(), channel, optionMap, providerDescriptor); // Send client greeting packet... final ByteBuffer buffer = connection.allocate(); // length placeholder buffer.putInt(0); + buffer.put(RemoteProtocol.GREETING); + // marshaller versions + final int[] versions = providerDescriptor.getSupportedVersions(); + for (int version : versions) { + GreetingUtils.writeInt(buffer, RemoteProtocol.GREETING_MARSHALLER_VERSION, version); + } // version ID GreetingUtils.writeByte(buffer, RemoteProtocol.GREETING_VERSION, RemoteProtocol.VERSION); // that's it! @@ -85,6 +93,7 @@ return; } } + RemoteConnectionHandler.log.warn("Client sent greeting message"); connection.free(buffer); channel.resumeReads(); return; Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/GreetingUtils.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/GreetingUtils.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/GreetingUtils.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -50,4 +50,10 @@ buffer.put((byte) 1); buffer.put(value); } + + static void writeInt(ByteBuffer buffer, byte type, int value) { + buffer.put(type); + buffer.put((byte) 4); + buffer.putInt(value); + } } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnection.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnection.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnection.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -28,6 +28,7 @@ import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.concurrent.Executor; +import org.jboss.marshalling.ProviderDescriptor; import org.jboss.remoting3.spi.AbstractHandleableCloseable; import org.jboss.xnio.Buffers; import org.jboss.xnio.IoUtils; @@ -39,15 +40,17 @@ final class RemoteConnection extends AbstractHandleableCloseable<RemoteConnection> implements Closeable { private final ConnectedStreamChannel<InetSocketAddress> channel; + private final ProviderDescriptor providerDescriptor; private final Pool<ByteBuffer> bufferPool = Buffers.createHeapByteBufferAllocator(4096); private final MessageHandler.Setter messageHandlerSetter; private final OptionMap optionMap; private boolean readDone; private final Object writeLock = new Object(); - RemoteConnection(final Executor executor, final ConnectedStreamChannel<InetSocketAddress> channel, final OptionMap optionMap) { + RemoteConnection(final Executor executor, final ConnectedStreamChannel<InetSocketAddress> channel, final OptionMap optionMap, final ProviderDescriptor providerDescriptor) { super(executor); this.channel = channel; + this.providerDescriptor = providerDescriptor; messageHandlerSetter = Channels.createMessageReader(channel, optionMap); this.optionMap = optionMap; } @@ -57,24 +60,10 @@ try { shutdownWritesBlocking(); } catch (IOException e) { - readDone = true; - writeLock.notifyAll(); IoUtils.safeClose(channel); return; } - while (! readDone) { - try { - writeLock.wait(); - } catch (InterruptedException e) { - readDone = true; - writeLock.notifyAll(); - IoUtils.safeClose(channel); - Thread.currentThread().interrupt(); - throw new InterruptedIOException(); - } - } } - channel.close(); } OptionMap getOptionMap() { @@ -101,13 +90,13 @@ try { sendBlockingNoClose(buffer); } catch (IOException e) { - IoUtils.safeClose(this); + IoUtils.safeClose(channel); throw e; } catch (RuntimeException e) { - IoUtils.safeClose(this); + IoUtils.safeClose(channel); throw e; } catch (Error e) { - IoUtils.safeClose(this); + IoUtils.safeClose(channel); throw e; } } @@ -139,13 +128,13 @@ channel.awaitWritable(); } } catch (IOException e) { - IoUtils.safeClose(this); + IoUtils.safeClose(channel); throw e; } catch (RuntimeException e) { - IoUtils.safeClose(this); + IoUtils.safeClose(channel); throw e; } catch (Error e) { - IoUtils.safeClose(this); + IoUtils.safeClose(channel); throw e; } } @@ -202,12 +191,7 @@ channel.shutdownReads(); } - void readDone() { - synchronized (writeLock) { - readDone = true; - writeLock.notifyAll(); - } + ProviderDescriptor getProviderDescriptor() { + return providerDescriptor; } - - } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -67,7 +67,7 @@ private final AtomicBoolean closed = new AtomicBoolean(); - public RemoteConnectionHandler(final ConnectionHandlerContext connectionContext, final RemoteConnection remoteConnection, final MarshallerFactory marshallerFactory) { + RemoteConnectionHandler(final ConnectionHandlerContext connectionContext, final RemoteConnection remoteConnection, final MarshallerFactory marshallerFactory) { super(connectionContext.getConnectionProviderContext().getExecutor()); this.connectionContext = connectionContext; this.remoteConnection = remoteConnection; Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionProvider.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionProvider.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteConnectionProvider.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -26,6 +26,7 @@ import java.net.InetSocketAddress; import java.net.URI; import java.net.UnknownHostException; +import org.jboss.marshalling.ProviderDescriptor; import org.jboss.remoting3.RemotingOptions; import org.jboss.remoting3.security.ServerAuthenticationProvider; import org.jboss.remoting3.spi.ConnectionHandlerFactory; @@ -52,10 +53,16 @@ private final ConnectionProviderContext connectionProviderContext; private final Connector<InetSocketAddress, ? extends ConnectedStreamChannel<InetSocketAddress>> connector; private final ProviderInterface providerInterface = new ProviderInterface(); + private final ProviderDescriptor providerDescriptor; RemoteConnectionProvider(final ConnectionProviderContext connectionProviderContext, final Connector<InetSocketAddress, ? extends ConnectedStreamChannel<InetSocketAddress>> connector) { this.connectionProviderContext = connectionProviderContext; this.connector = connector; + final ProviderDescriptor providerDescriptor = connectionProviderContext.getProtocolServiceProvider(ProtocolServiceType.MARSHALLER_PROVIDER_DESCRIPTOR, "river"); + if (providerDescriptor == null) { + throw new IllegalArgumentException("River marshalling protocol is not installed"); + } + this.providerDescriptor = providerDescriptor; } public Cancellable connect(final URI uri, final OptionMap connectOptions, final Result<ConnectionHandlerFactory> result, final CallbackHandler callbackHandler) throws IllegalArgumentException { @@ -71,7 +78,7 @@ // Open a client channel final IoFuture<? extends ConnectedStreamChannel<InetSocketAddress>> futureChannel; try { - futureChannel = connector.connectTo(new InetSocketAddress(InetAddress.getByName(host), port), new ClientOpenListener(connectOptions, connectionProviderContext, result, callbackHandler), null); + futureChannel = connector.connectTo(new InetSocketAddress(InetAddress.getByName(host), port), new ClientOpenListener(connectOptions, connectionProviderContext, result, callbackHandler, providerDescriptor), null); } catch (UnknownHostException e) { result.setException(e); return IoUtils.nullCancellable(); @@ -90,7 +97,7 @@ if (authenticationProvider == null) { throw new IllegalArgumentException("Missing authentication provider: " + (providerName == null ? "default" : providerName)); } - return new ServerOpenListener(optionMap, connectionProviderContext); + return new ServerOpenListener(optionMap, connectionProviderContext, providerDescriptor); } } } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteProtocol.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteProtocol.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/RemoteProtocol.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -67,6 +67,7 @@ static final byte GREETING_VERSION = 0; // sent by client & server static final byte GREETING_SASL_MECH = 1; // sent by server static final byte GREETING_ENDPOINT_NAME = 2; // sent by client & server + static final byte GREETING_MARSHALLER_VERSION = 3; // sent by client & server /** * Create an instance of the connection provider for the "remote" protocol. Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerAuthenticationHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerAuthenticationHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerAuthenticationHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -25,8 +25,6 @@ import java.io.IOException; import java.nio.ByteBuffer; import org.jboss.marshalling.MarshallerFactory; -import org.jboss.marshalling.Marshalling; -import org.jboss.marshalling.MarshallingConfiguration; import org.jboss.remoting3.CloseHandler; import org.jboss.remoting3.spi.ConnectionHandler; import org.jboss.remoting3.spi.ConnectionHandlerContext; @@ -70,7 +68,7 @@ remoteConnection.sendAuthMessage(RemoteProtocol.AUTH_COMPLETE, challenge); connectionProviderContext.accept(new ConnectionHandlerFactory() { public ConnectionHandler createInstance(final ConnectionHandlerContext connectionContext) { - final MarshallerFactory marshallerFactory = Marshalling.getMarshallerFactory("river"); + final MarshallerFactory marshallerFactory = remoteConnection.getProviderDescriptor().getMarshallerFactory(); final RemoteConnectionHandler connectionHandler = new RemoteConnectionHandler(connectionContext, remoteConnection, marshallerFactory); remoteConnection.addCloseHandler(new CloseHandler<Object>() { public void handleClose(final Object closed) { @@ -97,12 +95,4 @@ } } } - - public void handleEof() { - IoUtils.safeClose(remoteConnection); - } - - public void handleException(final IOException e) { - IoUtils.safeClose(remoteConnection); - } } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerGreetingHandler.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerGreetingHandler.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerGreetingHandler.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -22,7 +22,6 @@ package org.jboss.remoting3.remote; -import java.io.IOException; import java.nio.ByteBuffer; import java.util.Map; import java.util.Set; @@ -50,19 +49,35 @@ public void handleMessage(final ByteBuffer buffer) { switch (buffer.get()) { case RemoteProtocol.GREETING: { + RemoteConnectionHandler.log.trace("Server received greeting message"); + final int[] ourVersions = connection.getProviderDescriptor().getSupportedVersions(); + int bestVersion = -1; while (buffer.hasRemaining()) { final byte type = buffer.get(); final int len = buffer.get() & 0xff; + final ByteBuffer data = Buffers.slice(buffer, len); switch (type) { case RemoteProtocol.GREETING_VERSION: { // We only support version zero, so knowing the other side's version is not useful presently - buffer.get(); - if (len > 1) Buffers.skip(buffer, len - 1); break; } + case RemoteProtocol.GREETING_MARSHALLER_VERSION: { + final int remoteVersion = data.getInt(); + // is it better than the best one? if not, don't bother + if (remoteVersion <= bestVersion) { + break; + } + // do we support it? if not, skip + for (int ourVersion : ourVersions) { + if (ourVersion == remoteVersion) { + bestVersion = remoteVersion; + break; + } + } + break; + } default: { // unknown, skip it for forward compatibility. - Buffers.skip(buffer, len); break; } } Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerOpenListener.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerOpenListener.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/remote/ServerOpenListener.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -30,6 +30,7 @@ import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; +import org.jboss.marshalling.ProviderDescriptor; import org.jboss.remoting3.RemotingOptions; import org.jboss.remoting3.security.ServerAuthenticationProvider; import org.jboss.remoting3.spi.ConnectionProviderContext; @@ -48,10 +49,12 @@ private final OptionMap optionMap; private final ConnectionProviderContext connectionProviderContext; + private final ProviderDescriptor providerDescriptor; - ServerOpenListener(final OptionMap optionMap, final ConnectionProviderContext connectionProviderContext) { + ServerOpenListener(final OptionMap optionMap, final ConnectionProviderContext connectionProviderContext, final ProviderDescriptor providerDescriptor) { this.optionMap = optionMap; this.connectionProviderContext = connectionProviderContext; + this.providerDescriptor = providerDescriptor; } public void handleEvent(final ConnectedStreamChannel<InetSocketAddress> channel) { @@ -60,7 +63,7 @@ } catch (IOException e) { // ignore } - final RemoteConnection connection = new RemoteConnection(connectionProviderContext.getExecutor(), channel, optionMap); + final RemoteConnection connection = new RemoteConnection(connectionProviderContext.getExecutor(), channel, optionMap, providerDescriptor); // Calculate available server mechanisms final Sequence<String> mechs = optionMap.get(Options.SASL_MECHANISMS); @@ -81,8 +84,14 @@ final ByteBuffer buffer = connection.allocate(); // length placeholder buffer.putInt(0); + buffer.put(RemoteProtocol.GREETING); // version ID GreetingUtils.writeByte(buffer, RemoteProtocol.GREETING_VERSION, RemoteProtocol.VERSION); + // marshaller versions + final int[] versions = providerDescriptor.getSupportedVersions(); + for (int version : versions) { + GreetingUtils.writeInt(buffer, RemoteProtocol.GREETING_MARSHALLER_VERSION, version); + } // SASL server mechs for (String name : serverMechanisms) { GreetingUtils.writeString(buffer, RemoteProtocol.GREETING_SASL_MECH, name); @@ -110,6 +119,7 @@ return; } } + RemoteConnectionHandler.log.warn("Server sent greeting message"); connection.free(buffer); channel.resumeReads(); return; Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java =================================================================== --- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/spi/AbstractHandleableCloseable.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -50,7 +50,6 @@ private final StackTraceElement[] backtrace; private final Object closeLock = new Object(); - private Thread closingThread; private State state = State.OPEN; private Map<Key, CloseHandler<? super T>> closeHandlers = null; @@ -115,7 +114,6 @@ switch (state) { case OPEN: { state = State.CLOSING; - closingThread = Thread.currentThread(); closeHandlers = this.closeHandlers; this.closeHandlers = null; break; @@ -138,7 +136,6 @@ } finally { synchronized (closeLock) { state = State.CLOSED; - closingThread = null; closeLock.notifyAll(); } } @@ -244,13 +241,13 @@ try { super.finalize(); } finally { - if (! isOpen()) { + if (isOpen()) { if (LEAK_DEBUGGING) { final Throwable t = new LeakThrowable(); t.setStackTrace(backtrace); log.warn(t, "Leaked a %s instance: %s", getClass().getName(), this); } else { - log.warn("Leaked a %s instance: %s", getClass().getName(), this); + log.trace("Leaked a %s instance: %s", getClass().getName(), this); } IoUtils.safeClose(this); } Modified: remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/EndpointTestCase.java =================================================================== --- remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/EndpointTestCase.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/EndpointTestCase.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -27,6 +27,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.jboss.remoting3.Client; +import org.jboss.remoting3.CloseHandler; import org.jboss.remoting3.Endpoint; import org.jboss.remoting3.RemoteExecutionException; import org.jboss.remoting3.Remoting; @@ -88,11 +89,12 @@ throw new RemoteExecutionException(e); } } - - public void handleClose() { + }, Object.class, Object.class); + localClient.addCloseHandler(new CloseHandler<Client<Object, Object>>() { + public void handleClose(final Client<Object, Object> closed) { log.info("Listener closed"); } - }, Object.class, Object.class); + }); try { assertEquals(replyObj, localClient.invoke(requestObj)); } finally { @@ -121,11 +123,12 @@ throw new RemoteExecutionException(e); } } - - public void handleClose() { + }, Object.class, Object.class); + localClient.addCloseHandler(new CloseHandler<Client<Object, Object>>() { + public void handleClose(final Client<Object, Object> closed) { log.info("Listener closed"); } - }, Object.class, Object.class); + }); try { assertEquals(replyObj, localClient.send(requestObj).get()); } finally { Modified: remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/InvocationTestBase.java =================================================================== --- remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/InvocationTestBase.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/jboss-remoting/src/test/java/org/jboss/remoting3/test/InvocationTestBase.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -23,10 +23,12 @@ package org.jboss.remoting3.test; import java.io.IOException; +import org.jboss.marshalling.river.RiverMarshaller; import org.jboss.remoting3.Client; import org.jboss.remoting3.ClientConnector; import org.jboss.remoting3.ClientContext; import org.jboss.remoting3.ClientListener; +import org.jboss.remoting3.CloseHandler; import org.jboss.remoting3.Connection; import org.jboss.remoting3.Endpoint; import org.jboss.remoting3.Registration; @@ -56,6 +58,7 @@ public void setUp() throws IOException { enter(); try { + Thread.currentThread().setContextClassLoader(RiverMarshaller.class.getClassLoader()); endpoint = Remoting.getConfiguredEndpoint(); } finally { exit(); @@ -83,6 +86,11 @@ final Registration registration = endpoint.serviceBuilder().setGroupName("foo").setServiceType("test1").setRequestType(InvocationTestObject.class). setReplyType(InvocationTestObject.class).setClientListener(new ClientListener<InvocationTestObject, InvocationTestObject>() { public RequestListener<InvocationTestObject, InvocationTestObject> handleClientOpen(final ClientContext clientContext) { + clientContext.addCloseHandler(new CloseHandler<ClientContext>() { + public void handleClose(final ClientContext closed) { + log.info("Client closed"); + } + }); return new RequestListener<InvocationTestObject, InvocationTestObject>() { public void handleRequest(final RequestContext<InvocationTestObject> objectRequestContext, final InvocationTestObject request) throws RemoteExecutionException { try { @@ -92,10 +100,6 @@ throw new RemoteExecutionException(e); } } - - public void handleClose() { - log.info("Listener closed"); - } }; } }).register(); @@ -130,6 +134,11 @@ final Registration registration = endpoint.serviceBuilder().setGroupName("foo").setServiceType("test2").setRequestType(InvocationTestObject.class). setReplyType(InvocationTestObject.class).setClientListener(new ClientListener<InvocationTestObject, InvocationTestObject>() { public RequestListener<InvocationTestObject, InvocationTestObject> handleClientOpen(final ClientContext clientContext) { + clientContext.addCloseHandler(new CloseHandler<ClientContext>() { + public void handleClose(final ClientContext closed) { + log.info("Listener closed"); + } + }); return new RequestListener<InvocationTestObject, InvocationTestObject>() { public void handleRequest(final RequestContext<InvocationTestObject> objectRequestContext, final InvocationTestObject request) throws RemoteExecutionException { try { @@ -140,10 +149,6 @@ throw new RemoteExecutionException(e); } } - - public void handleClose() { - log.info("Listener closed"); - } }; } }).register(); @@ -179,6 +184,11 @@ final Registration registration = endpoint.serviceBuilder().setGroupName("foo").setServiceType("test3").setRequestType(ClientConnector.class). setReplyType(InvocationTestObject.class).setClientListener(new ClientListener<ClientConnector, InvocationTestObject>() { public RequestListener<ClientConnector, InvocationTestObject> handleClientOpen(final ClientContext clientContext) { + clientContext.addCloseHandler(new CloseHandler<ClientContext>() { + public void handleClose(final ClientContext closed) { + log.info("Listener closed"); + } + }); return new RequestListener<ClientConnector, InvocationTestObject>() { public void handleRequest(final RequestContext<InvocationTestObject> objectRequestContext, final ClientConnector request) throws RemoteExecutionException { try { @@ -189,10 +199,6 @@ throw new RemoteExecutionException(e); } } - - public void handleClose() { - log.info("Listener closed"); - } }; } }).register(); @@ -201,7 +207,7 @@ try { final Client<ClientConnector, InvocationTestObject> client = connection.openClient("test3", "*", ClientConnector.class, InvocationTestObject.class).get(); try { - client.invoke(connection.createClientConnector(new RequestListener<InvocationTestObject, InvocationTestObject>() { + final ClientConnector<InvocationTestObject, InvocationTestObject> clientConnector = connection.createClientConnector(new RequestListener<InvocationTestObject, InvocationTestObject>() { public void handleRequest(final RequestContext<InvocationTestObject> requestContext, final InvocationTestObject request) throws RemoteExecutionException { try { log.info("Got request %s, sending reply %s", request, replyObj); @@ -210,11 +216,19 @@ throw new RemoteExecutionException(e); } } - - public void handleClose() { + }, InvocationTestObject.class, InvocationTestObject.class); + final ClientContext context = clientConnector.getClientContext(); + context.addCloseHandler(new CloseHandler<ClientContext>() { + public void handleClose(final ClientContext closed) { log.info("Inner listener closed"); } - }, InvocationTestObject.class, InvocationTestObject.class)); + }); + try { + client.invoke(clientConnector); + } finally { + IoUtils.safeClose(context); + context.awaitClosedUninterruptibly(); + } } finally { IoUtils.safeClose(client); client.awaitClosedUninterruptibly(); Modified: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/StringRot13RequestListener.java =================================================================== --- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/StringRot13RequestListener.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/StringRot13RequestListener.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -28,9 +28,6 @@ } } - public void handleClose() { - } - private char rot13(final char i) { if (i >= 'A' && i <= 'M' || i >= 'a' && i <= 'm') { return (char) (i + 13); Modified: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/socket/SocketUsageExamples.java =================================================================== --- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/socket/SocketUsageExamples.java 2010-03-03 02:53:41 UTC (rev 5791) +++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/socket/SocketUsageExamples.java 2010-03-03 16:54:23 UTC (rev 5792) @@ -354,9 +354,6 @@ public static class DrNickRequestListener implements RequestListener<String, String> { - public void handleClose() { - } - public void handleRequest(RequestContext<String> context, String request) throws RemoteExecutionException { try { log.info(this + ": got request: " + request); @@ -376,9 +373,6 @@ public static class DrFrankensteinRequestListener implements RequestListener<String, String> { - public void handleClose() { - } - public void handleRequest(RequestContext<String> context, String request) throws RemoteExecutionException { try { log.info(this + ": got request: " + request); @@ -437,9 +431,6 @@ this.endpoint = endpoint; } - public void handleClose() { - } - public void handleRequest(RequestContext<RequestWrapper> context, RequestWrapper request) throws RemoteExecutionException { log.info(this + ": got request: " + request); Connection connection = null; @@ -482,16 +473,16 @@ private Client<Object, Object> client; private int counter; - public void handleClose() { - if (client != null) { - try { - client.close(); - client = null; - } catch (IOException e) { - log.warn(this + " unable to close Client " + client); - } - } - } +// public void handleClose() { +// if (client != null) { +// try { +// client.close(); +// client = null; +// } catch (IOException e) { +// log.warn(this + " unable to close Client " + client); +// } +// } +// } @SuppressWarnings("unchecked") @Override @@ -561,9 +552,6 @@ private static Logger log = Logger.getLogger(CallbackReceiverRequestListener.class); private ArrayList<Object> callbacks = new ArrayList<Object>(); - public void handleClose() { - } - public void handleRequest(RequestContext<Object> context, Object callback) throws RemoteExecutionException { log.info(this + ": got callback: " + callback); callbacks.add(callback);