Author: david.lloyd(a)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(); - } - } -}