[jboss-remoting-commits] JBoss Remoting SVN: r5568 - in remoting3/trunk: samples/src/main/java/org/jboss/remoting3/samples/protocol/basic and 2 other directories.
jboss-remoting-commits at lists.jboss.org
jboss-remoting-commits at lists.jboss.org
Fri Oct 23 17:19:18 EDT 2009
Author: david.lloyd at jboss.com
Date: 2009-10-23 17:19:18 -0400 (Fri, 23 Oct 2009)
New Revision: 5568
Added:
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/LocalBasicExample2Main.java
Removed:
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicConfiguration.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicHandlerReplyConsumer.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicProtocol.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicRequestHandler.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerReplyTransmitter.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerRequestConsumer.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/FutureBasicReply.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexClientExample.java
remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexServerExample.java
remoting3/trunk/samples/src/test/java/org/jboss/remoting3/samples/protocol/basic/BasicTestCase.java
Modified:
remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/CopyOnWriteHashMap.java
remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Endpoint.java
remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/EndpointImpl.java
remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistration.java
remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistrationListener.java
Log:
Fix wildcard lookups of group name; make the samples work or nuke them; add a local-connect sample
Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/CopyOnWriteHashMap.java
===================================================================
--- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/CopyOnWriteHashMap.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/CopyOnWriteHashMap.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -34,9 +34,17 @@
import java.util.concurrent.ConcurrentMap;
final class CopyOnWriteHashMap<K, V> implements ConcurrentMap<K, V> {
- private final Object writeLock = new Object();
+ private final Object writeLock;
private volatile Map<K, V> map = emptyMap();
+ CopyOnWriteHashMap() {
+ this(new Object());
+ }
+
+ CopyOnWriteHashMap(final Object writeLock) {
+ this.writeLock = writeLock;
+ }
+
public V putIfAbsent(final K key, final V value) {
if (key == null) {
throw new NullPointerException("key is null");
Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Endpoint.java
===================================================================
--- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Endpoint.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/Endpoint.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -55,7 +55,7 @@
* @param configuration the configuration to use
* @throws IOException if an error occurs
*/
- <I, O> SimpleCloseable registerService(LocalServiceConfiguration<I, O> configuration) throws IOException;
+ <I, O> Registration registerService(LocalServiceConfiguration<I, O> configuration) throws IOException;
/**
* Add a service registration listener which is called whenever a local service is registered.
@@ -66,7 +66,7 @@
* @param flags the flags to apply to the listener
* @return a handle which may be used to remove the listener registration
*/
- SimpleCloseable addServiceRegistrationListener(ServiceRegistrationListener listener, Set<ListenerFlag> flags);
+ Registration addServiceRegistrationListener(ServiceRegistrationListener listener, Set<ListenerFlag> flags);
/**
* Create a client that uses the given request handler to handle its requests.
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 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/EndpointImpl.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -25,7 +25,6 @@
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -33,14 +32,12 @@
import java.util.Map;
import java.util.Queue;
import java.util.Set;
+import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
import org.jboss.remoting3.spi.AbstractHandleableCloseable;
-import org.jboss.remoting3.spi.AbstractSimpleCloseable;
import org.jboss.remoting3.spi.ConnectionHandler;
import org.jboss.remoting3.spi.ConnectionHandlerFactory;
import org.jboss.remoting3.spi.ConnectionProvider;
@@ -82,6 +79,14 @@
return new CopyOnWriteHashMap<K, V>();
}
+ static <K, V> ConcurrentMap<K, V> concurrentMap(Object lock) {
+ return new CopyOnWriteHashMap<K, V>(lock);
+ }
+
+ static <T> Set<T> concurrentSet(Object lock) {
+ return Collections.<T>newSetFromMap(EndpointImpl.<T, Boolean>concurrentMap(lock));
+ }
+
static <K, V> Map<K, V> hashMap() {
return new HashMap<K, V>();
}
@@ -106,10 +111,10 @@
* Snapshot lock. Hold this lock while reading or updating {@link #serviceListenerRegistrations} or while updating
* {@link #registeredLocalServices}. Allows atomic snapshot of existing service registrations and listener add.
*/
- private final Lock serviceRegistrationLock = new ReentrantLock();
+ private final Object serviceRegistrationLock = new Object();
- private final Set<ListenerRegistration<ServiceRegistrationListener>> serviceListenerRegistrations = hashSet();
- private final Map<String, ServiceRegistration> registeredLocalServices = hashMap();
+ private final Set<ListenerRegistration<ServiceRegistrationListener>> serviceListenerRegistrations = Collections.newSetFromMap(EndpointImpl.<ListenerRegistration<ServiceRegistrationListener>, Boolean>concurrentMap(serviceRegistrationLock));
+ private final ConcurrentMap<String, ConcurrentMap<String, ServiceRegistration>> registeredLocalServices = concurrentMap(serviceRegistrationLock);
private final ConcurrentMap<String, ConnectionProvider<?>> connectionProviders = concurrentMap();
private final ConcurrentMap<String, MarshallingProtocol> marshallingProtocols = concurrentMap();
@@ -183,7 +188,7 @@
return localRequestHandler;
}
- public <I, O> SimpleCloseable registerService(final LocalServiceConfiguration<I, O> configuration) throws IOException {
+ public <I, O> Registration registerService(final LocalServiceConfiguration<I, O> configuration) throws IOException {
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
sm.checkPermission(REGISTER_SERVICE_PERM);
@@ -201,12 +206,13 @@
ServiceURI.validateServiceType(serviceType);
ServiceURI.validateGroupName(groupName);
checkOpen();
- final String serviceKey = serviceType.toLowerCase() + ":" + groupName.toLowerCase();
+ final String canonServiceType = serviceType.toLowerCase();
+ final String canonGroupName = groupName.toLowerCase();
final Class<I> requestClass = configuration.getRequestClass();
final Class<O> replyClass = configuration.getReplyClass();
final ClientListener<I, O> clientListener = configuration.getClientListener();
final Executor executor = this.executor;
- final Map<String, ServiceRegistration> registeredLocalServices = this.registeredLocalServices;
+ final ConcurrentMap<String, ConcurrentMap<String, ServiceRegistration>> registeredLocalServices = this.registeredLocalServices;
final RequestHandlerConnector requestHandlerConnector = new RequestHandlerConnector() {
public Cancellable createRequestHandler(final Result<RequestHandler> result) throws SecurityException {
try {
@@ -227,54 +233,47 @@
};
final ServiceRegistration registration = new ServiceRegistration(serviceType, groupName, name, optionMap, requestHandlerConnector);
// this handle is used to remove the service registration
- final AbstractSimpleCloseable newHandle = new AbstractSimpleCloseable(executor) {
- protected void closeAction() throws IOException {
- final Lock lock = serviceRegistrationLock;
- lock.lock();
- try {
- registeredLocalServices.remove(serviceKey);
- } finally {
- lock.unlock();
+ final Registration handle = new Registration() {
+ public void close() {
+ synchronized (serviceRegistrationLock) {
+ final ConcurrentMap<String, ServiceRegistration> submap = registeredLocalServices.get(serviceType);
+ if (submap != null) {
+ submap.remove(groupName, registration);
+ }
}
}
};
- registration.setHandle(newHandle);
- final List<ListenerRegistration<ServiceRegistrationListener>> serviceListenerRegistrations;
- final Lock lock = serviceRegistrationLock;
- // actually register the service, and while we have the lock, snag a copy of the registration listener list
- lock.lock();
- try {
- if (registeredLocalServices.containsKey(serviceKey)) {
- throw new ServiceRegistrationException("ListenerRegistration of service of type \"" + serviceType + "\" in group \"" + groupName + "\" duplicates an already-registered service's specification");
+ registration.setHandle(handle);
+ final Iterator<ListenerRegistration<ServiceRegistrationListener>> serviceListenerRegistrations;
+ final Object lock = serviceRegistrationLock;
+ synchronized (lock) {
+ // actually register the service, and while we have the lock, snag a copy of the registration listener list
+ final ConcurrentMap<String, ServiceRegistration> submap;
+ if (registeredLocalServices.containsKey(canonServiceType)) {
+ submap = registeredLocalServices.get(canonServiceType);
+ if (submap.containsKey(canonGroupName)) {
+ throw new ServiceRegistrationException("ListenerRegistration of service of type \"" + serviceType + "\" in group \"" + groupName + "\" duplicates an already-registered service's specification");
+ }
+ } else {
+ submap = concurrentMap(lock);
+ registeredLocalServices.put(canonServiceType, submap);
}
- registeredLocalServices.put(serviceKey, registration);
- serviceListenerRegistrations = new ArrayList<ListenerRegistration<ServiceRegistrationListener>>(this.serviceListenerRegistrations);
- } finally {
- lock.unlock();
+ submap.put(canonGroupName, registration);
+ // snapshot
+ serviceListenerRegistrations = this.serviceListenerRegistrations.iterator();
}
- // this registration closes the service registration when the endpoint is closed
- final WeakCloseable lrhCloseable = new WeakCloseable(newHandle);
- final Key key = addCloseHandler(new CloseHandler<Object>() {
- public void handleClose(final Object closed) {
- IoUtils.safeClose(lrhCloseable);
- }
- });
- // this registration removes the prior registration if the service registration is closed
- newHandle.addCloseHandler(new CloseHandler<Object>() {
- public void handleClose(final Object closed) {
- key.remove();
- }
- });
// notify all service listener registrations that were registered at the time the service was created
final ServiceRegistrationListener.ServiceInfo serviceInfo = new ServiceRegistrationListener.ServiceInfo();
serviceInfo.setGroupName(groupName);
serviceInfo.setServiceType(serviceType);
serviceInfo.setOptionMap(optionMap);
- serviceInfo.setRegistrationHandle(newHandle);
+ serviceInfo.setRegistrationHandle(handle);
serviceInfo.setRequestHandlerConnector(requestHandlerConnector);
executor.execute(new Runnable() {
public void run() {
- for (final ListenerRegistration<ServiceRegistrationListener> slr : serviceListenerRegistrations) {
+ final Iterator<ListenerRegistration<ServiceRegistrationListener>> iter = serviceListenerRegistrations;
+ while (iter.hasNext()) {
+ final ListenerRegistration<ServiceRegistrationListener> slr = iter.next();
final ServiceRegistrationListener registrationListener = slr.getResource();
try {
registrationListener.serviceRegistered(slr, serviceInfo.clone());
@@ -284,7 +283,7 @@
}
}
});
- return newHandle;
+ return handle;
}
private static void logListenerError(final Throwable t) {
@@ -324,36 +323,45 @@
return client;
}
- public SimpleCloseable addServiceRegistrationListener(final ServiceRegistrationListener listener, final Set<ListenerFlag> flags) {
+ public Registration addServiceRegistrationListener(final ServiceRegistrationListener listener, final Set<ListenerFlag> flags) {
final SecurityManager sm = System.getSecurityManager();
if (sm != null) {
sm.checkPermission(ADD_SERVICE_LISTENER_PERM);
}
+ final List<ServiceRegistration> services;
final ListenerRegistration<ServiceRegistrationListener> registration = new ListenerRegistration<ServiceRegistrationListener>(listener);
- final Lock lock = serviceRegistrationLock;
- final Collection<ServiceRegistration> services;
- lock.lock();
- try {
+ synchronized (serviceRegistrationLock) {
+ serviceListenerRegistrations.add(registration);
if (flags == null || ! flags.contains(ListenerFlag.INCLUDE_OLD)) {
- services = new ArrayList<ServiceRegistration>(registeredLocalServices.values());
+ // need to make a copy of the whole list
+ services = new ArrayList<ServiceRegistration>();
+ for (Map.Entry<String, ConcurrentMap<String, ServiceRegistration>> entry : registeredLocalServices.entrySet()) {
+ for (Map.Entry<String, ServiceRegistration> subEntry : entry.getValue().entrySet()) {
+ services.add(subEntry.getValue());
+ }
+ }
} else {
- services = Collections.emptySet();
+ services = null;
}
- } finally {
- lock.unlock();
}
- serviceListenerRegistrations.add(registration);
- for (ServiceRegistration service : services) {
- final ServiceRegistrationListener.ServiceInfo serviceInfo = new ServiceRegistrationListener.ServiceInfo();
- serviceInfo.setGroupName(service.getGroupName());
- serviceInfo.setOptionMap(service.getOptionMap());
- serviceInfo.setRegistrationHandle(service.getHandle());
- serviceInfo.setRequestHandlerConnector(service.getRequestHandlerConnector());
- serviceInfo.setServiceType(service.getServiceType());
- listener.serviceRegistered(registration, serviceInfo);
- if (! registration.isOpen()) {
- break;
- }
+ if (services != null) {
+ executor.execute(new Runnable() {
+ public void run() {
+ for (ServiceRegistration service : services) {
+ final ServiceRegistrationListener.ServiceInfo serviceInfo = new ServiceRegistrationListener.ServiceInfo();
+ serviceInfo.setGroupName(service.getGroupName());
+ serviceInfo.setOptionMap(service.getOptionMap());
+ serviceInfo.setRegistrationHandle(service.getHandle());
+ serviceInfo.setRequestHandlerConnector(service.getRequestHandlerConnector());
+ serviceInfo.setServiceType(service.getServiceType());
+ try {
+ listener.serviceRegistered(registration, serviceInfo);
+ } catch (Throwable t) {
+ logListenerError(t);
+ }
+ }
+ }
+ });
}
return registration;
}
@@ -491,26 +499,43 @@
private final class LocalConnectionContext implements ConnectionContext {
public void openService(final String serviceType, final String groupName, final OptionMap optionMap, final ServiceResult serviceResult) {
- final ServiceRegistration registration = registeredLocalServices.get(serviceType + ":" + groupName);
- if (registration != null) {
- registration.getRequestHandlerConnector().createRequestHandler(new Result<RequestHandler>() {
- public void setResult(final RequestHandler result) {
- serviceResult.opened(result, registration.getOptionMap());
- }
+ final String canonServiceType = serviceType.toLowerCase();
+ final String canonGroupName = groupName.toLowerCase();
+ final ConcurrentMap<String, ServiceRegistration> submap = registeredLocalServices.get(canonServiceType);
+ if (submap == null) {
+ serviceResult.notFound();
+ return;
+ }
+ final ServiceRegistration registration;
+ if (canonGroupName.equals("*")) {
+ final Iterator<Map.Entry<String, ServiceRegistration>> iter = submap.entrySet().iterator();
+ if (! iter.hasNext()) {
+ serviceResult.notFound();
+ return;
+ }
+ registration = iter.next().getValue();
+ } else {
+ registration = submap.get(canonGroupName);
+ if (registration == null) {
+ serviceResult.notFound();
+ return;
+ }
+ }
+ registration.getRequestHandlerConnector().createRequestHandler(new Result<RequestHandler>() {
+ public void setResult(final RequestHandler result) {
+ serviceResult.opened(result, registration.getOptionMap());
+ }
- public void setException(final IOException exception) {
- log.warn(exception, "Unexpected exception on service lookup");
- serviceResult.notFound();
- }
+ public void setException(final IOException exception) {
+ log.warn(exception, "Unexpected exception on service lookup");
+ serviceResult.notFound();
+ }
- public void setCancelled() {
- log.warn("Unexpected cancellation on service lookup");
- serviceResult.notFound();
- }
- });
- } else {
- serviceResult.notFound();
- }
+ public void setCancelled() {
+ log.warn("Unexpected cancellation on service lookup");
+ serviceResult.notFound();
+ }
+ });
}
}
@@ -550,28 +575,19 @@
}
}
- private final class ListenerRegistration<T> extends AbstractSimpleCloseable {
+ private final class ListenerRegistration<T> implements Registration {
private final T resource;
private ListenerRegistration(final T resource) {
- super(executor);
this.resource = resource;
}
- protected void closeAction() throws IOException {
- final Lock lock = serviceRegistrationLock;
- lock.lock();
- try {
+ public void close() {
+ synchronized (serviceRegistrationLock) {
serviceListenerRegistrations.remove(this);
- } finally {
- lock.unlock();
}
}
- protected boolean isOpen() {
- return super.isOpen();
- }
-
T getResource() {
return resource;
}
@@ -675,15 +691,16 @@
private class LoopbackConnectionHandler implements ConnectionHandler {
- public Cancellable open(final String serviceName, final String groupName, final Result<RequestHandler> result) {
- // the loopback connection opens a local service
- // local services are registered as RequestHandlerConnectors
- final ServiceRegistration registration = registeredLocalServices.get(serviceName + ":" + groupName);
- if (registration != null) {
- registration.getRequestHandlerConnector().createRequestHandler(result);
- } else {
- result.setException(new ServiceNotFoundException(ServiceURI.create(serviceName, groupName, name), "No such service located"));
- }
+ public Cancellable open(final String serviceType, final String groupName, final Result<RequestHandler> result) {
+ localConnectionContext.openService(serviceType, groupName, OptionMap.EMPTY, new ConnectionContext.ServiceResult() {
+ public void opened(final RequestHandler requestHandler, final OptionMap optionMap) {
+ result.setResult(requestHandler);
+ }
+
+ public void notFound() {
+ result.setException(new ServiceNotFoundException(ServiceURI.create(serviceType, groupName, name), "No such service located"));
+ }
+ });
return NULL_CANCELLABLE;
}
Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistration.java
===================================================================
--- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistration.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistration.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -36,7 +36,7 @@
private final String endpointName;
private final OptionMap optionMap;
private final RequestHandlerConnector requestHandlerConnector;
- private volatile SimpleCloseable handle;
+ private volatile Registration handle;
ServiceRegistration(final String serviceType, final String groupName, final String endpointName, final OptionMap optionMap, final RequestHandlerConnector requestHandlerConnector) {
this.requestHandlerConnector = requestHandlerConnector;
@@ -86,11 +86,11 @@
return requestHandlerConnector;
}
- public SimpleCloseable getHandle() {
+ public Registration getHandle() {
return handle;
}
- void setHandle(final SimpleCloseable handle) {
+ void setHandle(final Registration handle) {
this.handle = handle;
}
}
Modified: remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistrationListener.java
===================================================================
--- remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistrationListener.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/jboss-remoting/src/main/java/org/jboss/remoting3/ServiceRegistrationListener.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -23,7 +23,6 @@
package org.jboss.remoting3;
import org.jboss.remoting3.spi.RequestHandlerConnector;
-import org.jboss.remoting3.spi.RequestHandler;
import org.jboss.xnio.OptionMap;
/**
@@ -40,7 +39,7 @@
* @param listenerHandle the handle to this listener
* @param info the servce information
*/
- void serviceRegistered(SimpleCloseable listenerHandle, ServiceInfo info);
+ void serviceRegistered(Registration listenerHandle, ServiceInfo info);
/**
* Information about a registered service.
@@ -51,7 +50,7 @@
private String serviceType;
private String groupName;
private RequestHandlerConnector requestHandlerConnector;
- private SimpleCloseable registrationHandle;
+ private Registration registrationHandle;
private OptionMap optionMap;
/**
@@ -137,7 +136,7 @@
*
* @return the registration handle
*/
- public SimpleCloseable getRegistrationHandle() {
+ public Registration getRegistrationHandle() {
return registrationHandle;
}
@@ -146,7 +145,7 @@
*
* @param registrationHandle the registration handle
*/
- public void setRegistrationHandle(final SimpleCloseable registrationHandle) {
+ public void setRegistrationHandle(final Registration registrationHandle) {
this.registrationHandle = registrationHandle;
}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicConfiguration.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicConfiguration.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicConfiguration.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,80 +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.remoting3.samples.protocol.basic;
-
-import org.jboss.marshalling.MarshallerFactory;
-import org.jboss.marshalling.MarshallingConfiguration;
-import org.jboss.xnio.BufferAllocator;
-import java.util.concurrent.Executor;
-import java.nio.ByteBuffer;
-
-/**
- *
- */
-public final class BasicConfiguration {
- private MarshallerFactory marshallerFactory;
- private MarshallingConfiguration marshallingConfiguration;
- private int linkMetric;
- private Executor executor;
- private BufferAllocator<ByteBuffer> allocator;
-
- public MarshallerFactory getMarshallerFactory() {
- return marshallerFactory;
- }
-
- public void setMarshallerFactory(final MarshallerFactory marshallerFactory) {
- this.marshallerFactory = marshallerFactory;
- }
-
- public MarshallingConfiguration getMarshallingConfiguration() {
- return marshallingConfiguration;
- }
-
- public void setMarshallingConfiguration(final MarshallingConfiguration marshallingConfiguration) {
- this.marshallingConfiguration = marshallingConfiguration;
- }
-
- public int getLinkMetric() {
- return linkMetric;
- }
-
- public void setLinkMetric(final int linkMetric) {
- this.linkMetric = linkMetric;
- }
-
- public Executor getExecutor() {
- return executor;
- }
-
- public void setExecutor(final Executor executor) {
- this.executor = executor;
- }
-
- public BufferAllocator<ByteBuffer> getAllocator() {
- return allocator;
- }
-
- public void setAllocator(final BufferAllocator<ByteBuffer> allocator) {
- this.allocator = allocator;
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicHandlerReplyConsumer.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicHandlerReplyConsumer.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicHandlerReplyConsumer.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,144 +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.remoting3.samples.protocol.basic;
-
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.Lock;
-import java.util.Queue;
-import java.io.IOException;
-import org.jboss.marshalling.Unmarshaller;
-import org.jboss.xnio.channels.StreamChannel;
-import org.jboss.xnio.IoUtils;
-import org.jboss.xnio.log.Logger;
-import org.jboss.remoting3.spi.ReplyHandler;
-import org.jboss.remoting3.spi.SpiUtils;
-import org.jboss.remoting3.RemoteExecutionException;
-import org.jboss.remoting3.ReplyException;
-import org.jboss.remoting3.IndeterminateOutcomeException;
-
-/**
- *
- */
-final class BasicHandlerReplyConsumer implements Runnable {
-
- private static final Logger log = Logger.getLogger(BasicHandlerReplyConsumer.class);
-
- private final AtomicInteger replySequence;
- private final Unmarshaller unmarshaller;
- private final StreamChannel streamChannel;
- private final Lock reqLock;
- private final Queue<ReplyHandler> replyQueue;
-
- public BasicHandlerReplyConsumer(final Unmarshaller unmarshaller, final StreamChannel streamChannel, final Lock reqLock, final Queue<ReplyHandler> replyQueue) {
- this.unmarshaller = unmarshaller;
- this.streamChannel = streamChannel;
- this.reqLock = reqLock;
- this.replyQueue = replyQueue;
- replySequence = new AtomicInteger();
- }
-
- public void run() {
- try {
- for (;;) {
- final int type = unmarshaller.read();
- switch (type) {
- case -1: {
- // done.
- return;
- }
- case 1: {
- // reply - success
- reqLock.lock();
- try {
- replySequence.getAndIncrement();
- final ReplyHandler replyHandler = replyQueue.remove();
- final Object reply;
- try {
- reply = unmarshaller.readObject();
- } catch (Exception e) {
- SpiUtils.safeHandleException(replyHandler, new ReplyException("Failed to read reply from server", e));
- return;
- }
- SpiUtils.safeHandleReply(replyHandler, reply);
- break;
- } finally {
- reqLock.unlock();
- }
- }
- case 2: {
- // reply - cancelled
- reqLock.lock();
- try {
- final int id = unmarshaller.readInt();
- if (id != replySequence.getAndIncrement()) {
- replySequence.decrementAndGet();
- break;
- }
- final ReplyHandler replyHandler = replyQueue.remove();
- SpiUtils.safeHandleCancellation(replyHandler);
- break;
- } finally {
- reqLock.unlock();
- }
- }
- case 3: {
- // reply - exception
- reqLock.lock();
- try {
- replySequence.getAndIncrement();
- final ReplyHandler replyHandler = replyQueue.remove();
- final Throwable e;
- try {
- e = (Throwable) unmarshaller.readObject();
- } catch (Exception e2) {
- SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Failed to read exception from server", e2));
- return;
- }
- SpiUtils.safeHandleException(replyHandler, new RemoteExecutionException("Remote execution failed", e));
- break;
- } finally {
- reqLock.unlock();
- }
- }
- default: {
- // invalid byte
- throw new IOException("Read an invalid byte from the server");
- }
- }
- }
- } catch (Exception e) {
- log.error(e, "Error receiving reply");
- } finally {
- IoUtils.safeClose(streamChannel);
- reqLock.lock();
- try {
- while (replyQueue.size() > 0) {
- ReplyHandler replyHandler = replyQueue.remove();
- SpiUtils.safeHandleException(replyHandler, new IndeterminateOutcomeException("Connection terminated; operation outcome unknown"));
- }
- } finally {
- reqLock.unlock();
- }
- }
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicProtocol.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicProtocol.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicProtocol.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,82 +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.remoting3.samples.protocol.basic;
-
-import org.jboss.remoting3.spi.RequestHandler;
-import org.jboss.remoting3.spi.ReplyHandler;
-import org.jboss.xnio.channels.StreamChannel;
-import org.jboss.xnio.channels.ChannelOutputStream;
-import org.jboss.xnio.channels.ChannelInputStream;
-import org.jboss.marshalling.MarshallingConfiguration;
-import org.jboss.marshalling.MarshallerFactory;
-import org.jboss.marshalling.Unmarshaller;
-import org.jboss.marshalling.Marshaller;
-import org.jboss.marshalling.Marshalling;
-import java.io.IOException;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.Executor;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.Queue;
-import java.util.LinkedList;
-
-/**
- * A very basic example protocol.
- */
-public final class BasicProtocol {
-
- private BasicProtocol() {
- }
-
- public static final void createServer(final Handle<RequestHandler> requestHandlerHandle, final StreamChannel streamChannel, final BasicConfiguration configuration) throws IOException {
- final RequestHandler requestHandler = requestHandlerHandle.getResource();
- final MarshallingConfiguration marshallerConfiguration = configuration.getMarshallingConfiguration();
- final MarshallerFactory marshallerFactory = configuration.getMarshallerFactory();
- final Marshaller marshaller = marshallerFactory.createMarshaller(marshallerConfiguration);
- final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallerConfiguration);
- final Executor executor = configuration.getExecutor();
- marshaller.start(Marshalling.createByteOutput(new ChannelOutputStream(streamChannel)));
- unmarshaller.start(Marshalling.createByteInput(new ChannelInputStream(streamChannel)));
- final BlockingQueue<FutureBasicReply> replyQueue = new LinkedBlockingQueue<FutureBasicReply>();
- // todo - handle rejected execution...
- executor.execute(new BasicServerReplyTransmitter(replyQueue, marshaller, streamChannel, requestHandlerHandle));
- // todo - handle rejected execution...
- executor.execute(new BasicServerRequestConsumer(unmarshaller, requestHandler, replyQueue, streamChannel, requestHandlerHandle));
- }
-
- public static final Handle<RequestHandler> createClient(final StreamChannel streamChannel, final BasicConfiguration configuration) throws IOException {
- final MarshallingConfiguration marshallerConfiguration = configuration.getMarshallingConfiguration();
- final MarshallerFactory marshallerFactory = configuration.getMarshallerFactory();
- final Marshaller marshaller = marshallerFactory.createMarshaller(marshallerConfiguration);
- final Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(marshallerConfiguration);
- final Executor executor = configuration.getExecutor();
- marshaller.start(Marshalling.createByteOutput(new ChannelOutputStream(streamChannel)));
- unmarshaller.start(Marshalling.createByteInput(new ChannelInputStream(streamChannel)));
- final Lock reqLock = new ReentrantLock();
- final Queue<ReplyHandler> replyQueue = new LinkedList<ReplyHandler>();
- // todo - handle rejected execution...
- executor.execute(new BasicHandlerReplyConsumer(unmarshaller, streamChannel, reqLock, replyQueue));
- return new BasicRequestHandler(reqLock, marshaller, replyQueue, streamChannel, executor).getHandle();
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicRequestHandler.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicRequestHandler.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicRequestHandler.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,101 +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.remoting3.samples.protocol.basic;
-
-import org.jboss.remoting3.spi.RequestHandler;
-import org.jboss.remoting3.spi.ReplyHandler;
-import org.jboss.remoting3.spi.RemoteRequestContext;
-import org.jboss.remoting3.spi.SpiUtils;
-import org.jboss.marshalling.Marshaller;
-import org.jboss.xnio.channels.StreamChannel;
-import org.jboss.xnio.IoUtils;
-import org.jboss.xnio.log.Logger;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.Executor;
-import java.util.Queue;
-import java.io.IOException;
-
-/**
- *
- */
-final class BasicRequestHandler extends AbstractAutoCloseable<RequestHandler> implements RequestHandler {
-
- private static final Logger log = Logger.getLogger("org.jboss.remoting.basic");
-
- private final AtomicInteger requestSequence;
- private final Lock reqLock;
- private final Marshaller marshaller;
- private final Queue<ReplyHandler> replyQueue;
- private final StreamChannel streamChannel;
-
- public BasicRequestHandler(final Lock reqLock, final Marshaller marshaller, final Queue<ReplyHandler> replyQueue, final StreamChannel streamChannel, final Executor executor) {
- super(executor);
- this.reqLock = reqLock;
- this.marshaller = marshaller;
- this.replyQueue = replyQueue;
- this.streamChannel = streamChannel;
- requestSequence = new AtomicInteger();
- }
-
- public RemoteRequestContext receiveRequest(final Object request, final ReplyHandler replyHandler) {
- reqLock.lock();
- try {
- marshaller.write(2);
- marshaller.writeObject(request);
- marshaller.flush();
- final int id = requestSequence.getAndIncrement();
- replyQueue.add(replyHandler);
- return new RemoteRequestContext() {
- public RemoteRequestContext cancel() {
- reqLock.lock();
- try {
- marshaller.write(3);
- marshaller.writeInt(id);
- marshaller.flush();
- } catch (IOException e) {
- log.error(e, "Error writing cancel request");
- IoUtils.safeClose(BasicRequestHandler.this);
- } finally {
- reqLock.unlock();
- }
- return this;
- }
- };
- } catch (IOException e) {
- SpiUtils.safeHandleException(replyHandler, e);
- IoUtils.safeClose(this);
- return SpiUtils.getBlankRemoteRequestContext();
- } finally {
- reqLock.unlock();
- }
- }
-
- protected void closeAction() throws IOException {
- streamChannel.close();
- }
-
- public String toString() {
- return "basic protocol handler <" + Integer.toHexString(hashCode()) + ">";
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerReplyTransmitter.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerReplyTransmitter.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerReplyTransmitter.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -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.remoting3.samples.protocol.basic;
-
-import java.util.concurrent.BlockingQueue;
-import org.jboss.marshalling.Marshaller;
-import org.jboss.xnio.channels.StreamChannel;
-import org.jboss.xnio.IoUtils;
-import org.jboss.xnio.log.Logger;
-import org.jboss.remoting3.spi.RequestHandler;
-
-/**
- *
- */
-final class BasicServerReplyTransmitter implements Runnable {
-
- private static final Logger log = Logger.getLogger(BasicServerReplyTransmitter.class);
-
- private final BlockingQueue<FutureBasicReply> replyQueue;
- private final Marshaller marshaller;
- private final StreamChannel streamChannel;
- private final Handle<RequestHandler> requestHandlerHandle;
-
- public BasicServerReplyTransmitter(final BlockingQueue<FutureBasicReply> replyQueue, final Marshaller marshaller, final StreamChannel streamChannel, final Handle<RequestHandler> requestHandlerHandle) {
- this.replyQueue = replyQueue;
- this.marshaller = marshaller;
- this.streamChannel = streamChannel;
- this.requestHandlerHandle = requestHandlerHandle;
- }
-
- public void run() {
- try {
- for (;;) {
- final FutureBasicReply futureBasicReply = replyQueue.take();
- OUT: for (;;) switch (futureBasicReply.awaitInterruptibly()) {
- case DONE: {
- marshaller.write(1);
- marshaller.writeObject(futureBasicReply.get());
- marshaller.flush();
- break OUT;
- }
- case CANCELLED: {
- marshaller.write(2);
- marshaller.writeInt(futureBasicReply.id);
- marshaller.flush();
- break OUT;
- }
- case FAILED: {
- marshaller.write(3);
- marshaller.writeObject(futureBasicReply.getException());
- marshaller.flush();
- break OUT;
- }
- case WAITING: {
- // spurious wakeup, try again
- continue;
- }
- }
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- log.trace(e, "Interrupted");
- } catch (Exception e) {
- log.error(e, "Error in reply transmitter");
- } finally {
- IoUtils.safeClose(streamChannel);
- IoUtils.safeClose(requestHandlerHandle);
- }
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerRequestConsumer.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerRequestConsumer.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/BasicServerRequestConsumer.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,117 +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.remoting3.samples.protocol.basic;
-
-import org.jboss.marshalling.Unmarshaller;
-import org.jboss.remoting3.spi.RequestHandler;
-import org.jboss.remoting3.spi.RemoteRequestContext;
-import org.jboss.remoting3.spi.ReplyHandler;
-import org.jboss.xnio.channels.StreamChannel;
-import org.jboss.xnio.IoUtils;
-import org.jboss.xnio.log.Logger;
-import java.util.concurrent.BlockingQueue;
-import java.io.IOException;
-
-/**
- *
- */
-final class BasicServerRequestConsumer implements Runnable {
-
- private static final Logger log = Logger.getLogger(BasicServerRequestConsumer.class);
-
- private final Unmarshaller unmarshaller;
- private final RequestHandler requestHandler;
- private final BlockingQueue<FutureBasicReply> replyQueue;
- private final StreamChannel streamChannel;
- private final Handle<RequestHandler> requestHandlerHandle;
-
- public BasicServerRequestConsumer(final Unmarshaller unmarshaller, final RequestHandler requestHandler, final BlockingQueue<FutureBasicReply> replyQueue, final StreamChannel streamChannel, final Handle<RequestHandler> requestHandlerHandle) {
- this.unmarshaller = unmarshaller;
- this.requestHandler = requestHandler;
- this.replyQueue = replyQueue;
- this.streamChannel = streamChannel;
- this.requestHandlerHandle = requestHandlerHandle;
- }
-
- public void run() {
- try {
- int requestSequence = 0;
- for (;;) {
- final int id = unmarshaller.read();
- switch (id) {
- case -1: {
- // done.
- return;
- }
- case 2: {
- // two-way request
- final int requestId = requestSequence++;
- final Object request = unmarshaller.readObject();
- final FutureBasicReply future = new FutureBasicReply(requestId);
- replyQueue.add(future);
- final RemoteRequestContext requestContext = requestHandler.receiveRequest(request, new ReplyHandler() {
-
- public void handleReply(final Object reply) {
- future.setResult(reply);
- }
-
- public void handleException(final IOException exception) {
- future.setException(exception);
- }
-
- public void handleCancellation() {
- future.finishCancel();
- }
- });
- future.requestContext = requestContext;
- break;
- }
- case 3: {
- // cancel request
- final int requestId = unmarshaller.readInt();
- // simply iterate over the outstanding requests until we match or are past it...
- for (FutureBasicReply future : replyQueue) {
- final int queuedId = future.id;
- if (queuedId == requestId) {
- future.cancel();
- break;
- } else if (queuedId > requestId) {
- break;
- }
- }
- break;
- }
- default: {
- // invalid byte
- throw new IOException("Read an invalid byte from the client");
- }
- }
- }
- } catch (Exception e) {
- log.error(e, "Connection failed");
- } finally {
- IoUtils.safeClose(streamChannel);
- IoUtils.safeClose(requestHandlerHandle);
- }
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/FutureBasicReply.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/FutureBasicReply.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/protocol/basic/FutureBasicReply.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,58 +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.remoting3.samples.protocol.basic;
-
-import org.jboss.xnio.AbstractIoFuture;
-import org.jboss.xnio.IoFuture;
-import org.jboss.remoting3.spi.RemoteRequestContext;
-import java.io.IOException;
-
-/**
- *
- */
-final class FutureBasicReply extends AbstractIoFuture<Object> {
-
- final int id;
- RemoteRequestContext requestContext;
-
- public FutureBasicReply(final int id) {
- this.id = id;
- }
-
- protected boolean setException(final IOException exception) {
- return super.setException(exception);
- }
-
- protected boolean setResult(final Object result) {
- return super.setResult(result);
- }
-
- protected boolean finishCancel() {
- return super.finishCancel();
- }
-
- public IoFuture<Object> cancel() {
- requestContext.cancel();
- return this;
- }
-}
Copied: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/LocalBasicExample2Main.java (from rev 5492, remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/LocalBasicExampleMain.java)
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/LocalBasicExample2Main.java (rev 0)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/LocalBasicExample2Main.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -0,0 +1,71 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2009, 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.samples.simple;
+
+import java.net.URI;
+import org.jboss.remoting3.Client;
+import org.jboss.remoting3.Endpoint;
+import org.jboss.remoting3.Remoting;
+import org.jboss.remoting3.Connection;
+import org.jboss.remoting3.LocalServiceConfiguration;
+import org.jboss.remoting3.Registration;
+import org.jboss.xnio.IoUtils;
+import org.jboss.xnio.OptionMap;
+
+/**
+ *
+ */
+public final class LocalBasicExample2Main {
+
+ private LocalBasicExample2Main() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ final Endpoint endpoint = Remoting.createEndpoint("simple");
+ try {
+ final LocalServiceConfiguration<String, String> config = LocalServiceConfiguration.create(new StringRot13ClientListener(), String.class, String.class);
+ config.setGroupName("main");
+ config.setServiceType("simple.rot13");
+ final Registration handle = endpoint.registerService(config);
+ try {
+ final Connection connection = endpoint.connect(new URI("local:///"), OptionMap.EMPTY).get();
+ try {
+ final Client<String, String> client = connection.openClient("simple.rot13", "*", String.class, String.class).get();
+ try {
+ final String original = "The Secret Message";
+ final String result = client.invoke(original);
+ System.out.printf("The secret message \"%s\" became \"%s\"!\n", original.trim(), result.trim());
+ } finally {
+ IoUtils.safeClose(client);
+ }
+ } finally {
+ IoUtils.safeClose(connection);
+ }
+ } finally {
+ IoUtils.safeClose(handle);
+ }
+ } finally {
+ IoUtils.safeClose(endpoint);
+ }
+ }
+}
\ No newline at end of file
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexClientExample.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexClientExample.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexClientExample.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -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.remoting3.samples.simple;
-
-import org.jboss.remoting3.Endpoint;
-import org.jboss.remoting3.Remoting;
-import org.jboss.xnio.OptionMap;
-import org.jboss.remoting3.Connection;
-import org.jboss.remoting3.Client;
-import org.jboss.xnio.IoUtils;
-import java.io.IOException;
-import java.net.URI;
-
-import java.util.logging.Logger;
-import java.util.logging.Level;
-
-/**
- *
- */
-public final class MultiplexClientExample {
-
- static {
- final Logger l = Logger.getLogger("");
- l.getHandlers()[0].setLevel(Level.ALL);
- l.setLevel(Level.INFO);
- }
-
- private MultiplexClientExample() {
- }
-
- public static void main(String[] args) {
- try {
- final Endpoint endpoint = Remoting.createEndpoint("example-client-endpoint");
- try {
- final Connection connection = endpoint.connect(URI.create(args[0]), OptionMap.EMPTY).get();
- try {
- final Client<String,String> client = connection.openClient("samples.rot13", "*", String.class, String.class).get();
- try {
- final String original = "The Secret Message\n";
- final String result = client.invoke(original);
- System.out.printf("The secret message \"%s\" became \"%s\"!\n", original.trim(), result.trim());
- } finally {
- IoUtils.safeClose(client);
- }
- } finally {
- IoUtils.safeClose(connection);
- }
- } finally {
- IoUtils.safeClose(endpoint);
- }
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-}
Deleted: remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexServerExample.java
===================================================================
--- remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexServerExample.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/main/java/org/jboss/remoting3/samples/simple/MultiplexServerExample.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,97 +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.remoting3.samples.simple;
-
-import org.jboss.remoting3.Remoting;
-import org.jboss.remoting3.Endpoint;
-import org.jboss.remoting3.LocalServiceConfiguration;
-import org.jboss.remoting3.SimpleCloseable;
-import org.jboss.remoting3.spi.ConnectionProviderRegistration;
-import org.jboss.remoting3.multiplex.MultiplexConnectionProviderFactory;
-import org.jboss.remoting3.multiplex.MultiplexServerFactory;
-import org.jboss.xnio.IoUtils;
-import org.jboss.xnio.CloseableExecutor;
-import org.jboss.xnio.Xnio;
-import org.jboss.xnio.TcpServer;
-import org.jboss.xnio.OptionMap;
-import org.jboss.xnio.channels.Channels;
-import java.io.IOException;
-
-import java.util.logging.Logger;
-import java.util.logging.Level;
-
-/**
- *
- */
-public final class MultiplexServerExample {
-
- static {
- final Logger l = Logger.getLogger("");
- l.getHandlers()[0].setLevel(Level.ALL);
- l.setLevel(Level.INFO);
- }
-
- private MultiplexServerExample() {
- }
-
- public static void main(String[] args) throws InterruptedException, IOException {
- final CloseableExecutor executor = Remoting.createExecutor(10);
- try {
- Xnio xnio = Xnio.create();
- try {
- final Endpoint endpoint = Remoting.createEndpoint(executor, "example-endpoint");
- try {
- final LocalServiceConfiguration<String, String> config = LocalServiceConfiguration.create(new StringRot13ClientListener(), String.class, String.class);
- config.setGroupName("main");
- config.setServiceType("simple.rot13");
- final SimpleCloseable handle = endpoint.registerService(config);
- try {
- // now create the server...
- final MultiplexConnectionProviderFactory multiplexConnectionProviderFactory = new MultiplexConnectionProviderFactory(xnio.createTcpConnector(OptionMap.EMPTY));
- final ConnectionProviderRegistration<MultiplexServerFactory> cpHandle = endpoint.addConnectionProvider("multiplex", multiplexConnectionProviderFactory);
- try {
- final TcpServer tcpServer = xnio.createTcpServer(Channels.createAllocatedMessageChannel(cpHandle.getProviderInterface().getServerListener(), OptionMap.EMPTY)).create();
- try {
- // now just wait for 15 seconds, and then shut it all down
- Thread.sleep(15000L);
- } finally {
- IoUtils.safeClose(tcpServer);
- }
- } finally {
- IoUtils.safeClose(cpHandle);
- }
- } finally {
- IoUtils.safeClose(handle);
- }
- } finally {
- IoUtils.safeClose(endpoint);
- }
- } finally {
- IoUtils.safeClose(xnio);
- }
- } finally {
- IoUtils.safeClose(executor);
- }
- }
-
-}
Deleted: remoting3/trunk/samples/src/test/java/org/jboss/remoting3/samples/protocol/basic/BasicTestCase.java
===================================================================
--- remoting3/trunk/samples/src/test/java/org/jboss/remoting3/samples/protocol/basic/BasicTestCase.java 2009-10-23 19:34:47 UTC (rev 5567)
+++ remoting3/trunk/samples/src/test/java/org/jboss/remoting3/samples/protocol/basic/BasicTestCase.java 2009-10-23 21:19:18 UTC (rev 5568)
@@ -1,132 +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.remoting3.samples.protocol.basic;
-
-import junit.framework.TestCase;
-import org.jboss.xnio.Xnio;
-import org.jboss.xnio.IoUtils;
-import org.jboss.xnio.ChannelSource;
-import org.jboss.xnio.IoFuture;
-import org.jboss.xnio.nio.NioXnio;
-import org.jboss.xnio.channels.StreamChannel;
-import org.jboss.remoting3.Endpoint;
-import org.jboss.remoting3.Remoting;
-import org.jboss.remoting3.RequestContext;
-import org.jboss.remoting3.RemoteExecutionException;
-import org.jboss.remoting3.Client;
-import org.jboss.remoting3.spi.RequestHandler;
-import org.jboss.marshalling.MarshallingConfiguration;
-import org.jboss.marshalling.river.RiverMarshallerFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.io.IOException;
-
-/**
- *
- */
-public final class BasicTestCase extends TestCase {
-
- public static void testConnect() throws Throwable {
- ExecutorService executor = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
- try {
- Xnio xnio = NioXnio.create(executor, 2, 2, 2);
- try {
- final BasicConfiguration configuration = new BasicConfiguration();
- configuration.setExecutor(executor);
- configuration.setMarshallerFactory(new RiverMarshallerFactory());
- final MarshallingConfiguration marshallingConfiguration = new MarshallingConfiguration();
- configuration.setMarshallingConfiguration(marshallingConfiguration);
- final Endpoint endpoint = Remoting.createEndpoint(executor, "test");
- try {
- final Handle<RequestHandler> requestHandlerHandle = endpoint.createLocalRequestHandler(new AbstractRequestListener<Object, Object>() {
- public void handleRequest(final RequestContext<Object> context, final Object request) throws RemoteExecutionException {
- System.out.println("Got a request! " + request.toString());
- try {
- context.sendReply("GOOMBA");
- } catch (IOException e) {
- try {
- context.sendFailure("Failed", e);
- } catch (IOException e1) {
- // buh
- }
- }
- }
- }, Object.class, Object.class);
- try {
- final ChannelSource<StreamChannel> channelSource = xnio.createPipeServer(executor, IoUtils.singletonHandlerFactory(new IoHandler<StreamChannel>() {
- public void handleOpened(final StreamChannel channel) {
- try {
- System.out.println("Opening channel");
- BasicProtocol.createServer(requestHandlerHandle, channel, configuration);
- } catch (IOException e) {
- e.printStackTrace();
- IoUtils.safeClose(channel);
- }
- }
-
- public void handleReadable(final StreamChannel channel) {
- }
-
- public void handleWritable(final StreamChannel channel) {
- }
-
- public void handleClosed(final StreamChannel channel) {
- System.out.println("Closing channel");
- }
- }));
- final IoFuture<StreamChannel> futureChannel = channelSource.open(IoUtils.nullHandler());
- assertEquals(IoFuture.Status.DONE, futureChannel.await(1L, TimeUnit.SECONDS));
- final StreamChannel channel = futureChannel.get();
- try {
- final Handle<RequestHandler> clientHandlerHandle = BasicProtocol.createClient(channel, configuration);
- try {
- final Client<Object,Object> client = endpoint.createClient(clientHandlerHandle.getResource(), Object.class, Object.class);
- try {
- final IoFuture<? extends Object> futureReply = client.send("GORBA!");
- assertEquals(IoFuture.Status.DONE, futureReply.await(500L, TimeUnit.MILLISECONDS));
- System.out.println("Reply is:" + futureReply.get());
- } finally {
- IoUtils.safeClose(client);
- }
- } finally {
- IoUtils.safeClose(clientHandlerHandle);
- }
- } finally {
- IoUtils.safeClose(channel);
- }
- } finally {
- IoUtils.safeClose(requestHandlerHandle);
- }
- } finally {
- IoUtils.safeClose(endpoint);
- }
- } finally {
- IoUtils.safeClose(xnio);
- }
- } finally {
- executor.shutdownNow();
- }
- }
-}
More information about the jboss-remoting-commits
mailing list