[jboss-svn-commits] JBL Code SVN: r28794 - in labs/jbossesb/branches/JBESB_4_4_GA_CP/product: install and 13 other directories.
jboss-svn-commits at lists.jboss.org
jboss-svn-commits at lists.jboss.org
Tue Aug 4 11:28:36 EDT 2009
Author: kevin.conner at jboss.com
Date: 2009-08-04 11:28:34 -0400 (Tue, 04 Aug 2009)
New Revision: 28794
Added:
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMException.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransport.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransport.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMService.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMServiceMBean.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransportUnitTest.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransportUnitTest.java
Modified:
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/docs/AdministrationGuide.odt
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/docs/ProgrammersGuide.odt
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/conf/jbossesb-properties.xml
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/jboss-service.xml
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/TwoWayCourierImpl.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/tx/InVMXAResource.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMRegistryInterceptor.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/addressing/eprs/InVMEpr.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/common/Environment.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/couriers/CourierFactory.java
labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/tests/InVMCourierUnitTest.java
Log:
Added InVM transport mechanism, separating courier concerns from permanent/temporary endpoint functionality: JBESB-2758
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/docs/AdministrationGuide.odt
===================================================================
(Binary files differ)
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/docs/ProgrammersGuide.odt
===================================================================
(Binary files differ)
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/conf/jbossesb-properties.xml
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/conf/jbossesb-properties.xml 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/conf/jbossesb-properties.xml 2009-08-04 15:28:34 UTC (rev 28794)
@@ -88,6 +88,9 @@
<property name="org.jboss.soa.esb.ftp.remotedir" value="/tmp"/>
<property name="org.jboss.soa.esb.jms.connectionPool" value="20"/>
<property name="org.jboss.soa.esb.jms.sessionSleep" value="30"/>
+ <!-- expiry of messages to temporary invm destinations (default replies)
+ expire after the following timeout, specified in milliseconds (5 mins) -->
+ <property name="org.jboss.soa.esb.invm.expiryTime" value="300000"/>
<!-- Uncomment this to return stack traces when exposing an ESB service through WS
<property name="org.jboss.soa.esb.ws.returnStackTrace" value="true"/>
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/jboss-service.xml
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/jboss-service.xml 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/install/jboss-service.xml 2009-08-04 15:28:34 UTC (rev 28794)
@@ -9,6 +9,10 @@
name="jboss.esb:service=ESBRegistry">
<depends>jboss.esb:service=PropertyService</depends>
</mbean>
+ <mbean code="org.jboss.internal.soa.esb.services.registry.InVMService"
+ name="jboss.esb:service=InVM">
+ <depends>jboss.esb:service=PropertyService</depends>
+ </mbean>
<mbean code="org.jboss.internal.soa.esb.dependencies.JuddiRMIService"
name="jboss.esb:service=JuddiRMI"
description="Juddi RMI Service">
@@ -19,6 +23,7 @@
<mbean code="org.jboss.soa.esb.listeners.config.JBoss4ESBDeployer"
name="jboss.esb:service=ESBDeployer">
<depends>jboss.esb:service=ESBRegistry</depends>
+ <depends>jboss.esb:service=InVM</depends>
<depends>jboss.esb:service=JuddiRMI</depends>
</mbean>
<mbean code="org.jboss.soa.esb.listeners.message.MessageCounter"
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -19,34 +19,39 @@
package org.jboss.internal.soa.esb.addressing.eprs;
-import org.jboss.soa.esb.addressing.EPR;
import org.jboss.soa.esb.addressing.eprs.InVMEpr;
-import org.jboss.soa.esb.couriers.CourierFactory;
import java.net.URI;
import java.net.URISyntaxException;
public class DefaultInVMReplyToEpr extends InVMEpr {
- private static volatile long counter = 0;
+ private static ThreadLocal<Long> counter = new ThreadLocal<Long>() ;
public DefaultInVMReplyToEpr(InVMEpr epr) {
super(createEprAndCourier());
}
- private static EPR createEprAndCourier() {
+ private static InVMEpr createEprAndCourier() {
try {
- InVMEpr epr;
-
- synchronized (DefaultInVMReplyToEpr.class) {
- counter++;
- epr = new InVMEpr(URI.create("invm://thread-" + Thread.currentThread().getId() + "-" + counter));
- }
- CourierFactory.getInstance().addInVMCourier(epr);
-
- return epr;
+ final InVMEpr epr = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://thread-" + Thread.currentThread().getId() + "-" + getNextCounter()));
+ epr.setTemporaryEPR(true) ;
+ return epr ;
} catch (URISyntaxException e) {
throw new RuntimeException("Unexpected Exception creating replyTo InVMEpr.", e);
}
}
+
+ private static long getNextCounter() {
+ final Long current = counter.get() ;
+ final long value ;
+ if (current != null) {
+ value = current.longValue() + 1 ;
+ } else {
+ value = 0 ;
+ }
+
+ counter.set(Long.valueOf(value)) ;
+ return value ;
+ }
}
\ No newline at end of file
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -23,48 +23,19 @@
package org.jboss.internal.soa.esb.couriers;
import org.apache.log4j.Logger;
-import org.jboss.internal.soa.esb.couriers.tx.InVMXAResource;
-import org.jboss.internal.soa.esb.message.format.MessageSerializer;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTemporaryTransport;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTransport;
import org.jboss.soa.esb.addressing.eprs.InVMEpr;
-import org.jboss.soa.esb.common.TransactionStrategy;
-import org.jboss.soa.esb.common.TransactionStrategyException;
import org.jboss.soa.esb.couriers.CourierException;
-import org.jboss.soa.esb.message.ByReferenceMessage;
import org.jboss.soa.esb.message.Message;
-import org.jboss.soa.esb.Service;
-import java.io.IOException;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-/*
- * These need to be treated differently to other couriers or we'll end up with
- * garbage collection issues. We need to enforce a rule that a sender cannot
- * send (gets an error) if the server isn't listening. That's straightforward to
- * do here, but we need to check the impact on the rest of the codebase, since the
- * other couriers can be used asynchronously of the service lifecycle. The courier
- * instance should be purges when the service is finished with it.
- *
- * An alternative would be to use a WeakHashMap for maintaining the InVMCouriers in
- * the InternalCourierFactory. But I'd prefer to have explicit control over the courier
- * lifecycle rather than leaving to the vagueries of the gc. Plus, it could introduce
- * some interesting race conditions.
- */
-
public class InVMCourier implements PickUpOnlyCourier, DeliverOnlyCourier {
- protected InVMEpr epr;
+ private static Logger logger = Logger.getLogger(InVMCourier.class);
- protected static Logger logger = Logger.getLogger(InVMCourier.class);
+ private final InVMEpr epr;
- private Service service;
- private Queue<Object> messageQueue = new ConcurrentLinkedQueue<Object>();
-
- private long deliveryTimeout = 0;
-
- private boolean isActive = true;
- private boolean passByValue = true;
-
/**
* Objects of this class should only be instantiated by internal
* implementations
@@ -74,44 +45,9 @@
public InVMCourier(InVMEpr epr) {
this.epr = epr;
-
- /*
- * Normally we keep the link between sender and receiver threads. This
- * makes the model simpler and allows services to be transplanted
- * between transports. Plus, this is based on the experiences from CORBA
- * 2.0 to 2.3, where POA was introduced: by default all local
- * invocations then looked like remote invocations and weren't
- * necessarily handled by the same thread. But all good ORBs had a
- * workaround to go back to the old style, where the same thread did all
- * of the work in "lock step" (it was the same thread).
- */
-
- if (epr.getLockstep()) {
- deliveryTimeout = epr.getLockstepWaitTime();
- } else {
- deliveryTimeout = 0;
- }
-
- passByValue = epr.getPassByValue();
}
/**
- * Get the Service name owning this InVMCourier instance.
- * @return The owning Service name.
- */
- public Service getService() {
- return service;
- }
-
- /**
- * Set the Service name owning this InVMCourier instance.
- * @param service The owning Service name.
- */
- public void setService(Service service) {
- this.service = service;
- }
-
- /**
* package the ESB message into the queue. If this is a transactional interaction
* then the deliver will return immediately, but the message may not go into
* the queue if the transaction subsequently rolls back. The caller must monitor
@@ -125,126 +61,29 @@
*/
public boolean deliver(Message message) throws CourierException {
- assertCourierActive() ;
-
if (message == null) {
return false;
}
try
{
- /*
- * Are we operating within the scope of a global transaction?
- */
-
- if (isTransactional())
- {
- /*
- * If we are transactional, then hold off on the insertion until the transaction commits.
- * The XAResource is given the duty of doing the insert in that case.
- * If it doesn't commit, then the insert won't happen and the client needs to monitor
- * the transaction outcome.
- *
- * We could get away with using a Synchronization rather than an XAResource
- * since there's no durable component in this case. However, Synchronizations aren't
- * guaranteed to be called in all circumstances and unless you've read the JTS spec. it'll
- * make maintaining things that little bit more difficult. Using a Synchronization would
- * be ok with JBossTS, but not necessarily other implementations. We could end up with
- * a non-atomic outcome as a result.
- *
- * The downside of this is that we could force a 1PC transaction to become 2PC when in
- * fact there's no real difference to the 1PC participant. But the upside is we get
- * consistency!
- */
-
- TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true);
-
- txStrategy.enlistResource(new InVMXAResource(this, message, InVMXAResource.operation.INSERT));
-
- /*
- * Can't do lockstep wait here because otherwise the transaction may not terminate if this
- * is the transaction controller thread!
- */
-
- return true; // does not mean message got into the queue.
+ if (epr.isTemporaryEPR()) {
+ InVMTemporaryTransport.getInstance().deliver(epr, message) ;
+ } else {
+ InVMTransport.getInstance().deliver(epr, message) ;
}
- else
- {
- Object addedObject;
-
- if (passByValue) {
- try {
- addedObject = MessageSerializer.serialize(message);
- } catch (IOException ex) {
- logger.warn("Could not serialize message to pass by value.", ex);
- return false;
- }
- } else if (message instanceof ByReferenceMessage){
- addedObject = ((ByReferenceMessage)message).reference() ;
- } else {
- addedObject = message;
- }
-
- synchronized (messageQueue) {
-
- messageQueue.add(addedObject);
-
- // Notify 1 waiting pickup thread of the delivery...
- messageQueue.notify();
-
- if (deliveryTimeout > 0) {
- long start = System.currentTimeMillis();
-
- // Continue to wait until the "addedObject" has been removed from the queue,
- // or the delivery timeout expires...
- while(messageQueue.contains(addedObject)) {
- try {
- // Wait on notification from the pickup thread....
- messageQueue.wait(5); // Yes, it's a "magic" number, but we don't need to configure it or make a one-off constant!!!
- } catch (InterruptedException e) {
- logger.warn("Waiting delivery thread interupted while waiting on message pickup on InVM queue '" + epr.getAddr().getAddress() + "'. Exiting pickup wait state.");
- break;
- }
-
- // If the delivery timeout has expired...
- if(System.currentTimeMillis() > start + deliveryTimeout) {
- logger.warn("Timeout expired while waiting on message pickup on InVM queue '" + epr.getAddr().getAddress() + "'.");
- break;
- }
- }
- }
- }
-
- return true;
- }
+
+ return true;
}
catch (final Throwable ex)
{
logger.warn("InVMCourier delivery caught: "+ex);
+ logger.debug("InVMCourier delivery caught", ex);
return false;
}
}
- private Object addMessageToQueue(Message message) {
- Object addedObject;
-
- if (passByValue) {
- try {
- addedObject = MessageSerializer.serialize(message);
- messageQueue.add(addedObject);
- } catch (IOException ex) {
- logger.warn("Could not serialize message to pass by value.", ex);
- return false;
- }
- } else {
- addedObject = message;
- messageQueue.add(message);
- }
-
- return addedObject;
- }
-
/**
* Get a mesage from the queue or wait for the specified time in case one
* arrives.
@@ -261,177 +100,23 @@
// see associated test
public Message pickup(long millis) throws CourierException {
- assertCourierActive() ;
-
- Message message = null;
-
try
{
- millis = Math.max(millis, 100);
-
- Object messageObj = null;
-
- synchronized (messageQueue) {
- if (messageQueue.isEmpty()) {
- try {
- messageQueue.wait(millis);
- } catch (InterruptedException e) {
- logger.debug("Pickup thread '" + Thread.currentThread().getName() + "' interrupted while waiting on delivery notification or timeout.", e);
- }
- }
- if (!messageQueue.isEmpty()) {
- messageObj = messageQueue.remove();
- }
- }
-
- if(messageObj instanceof byte[]) {
- message = MessageSerializer.deserialize((byte[]) messageObj);
+ if (epr.isTemporaryEPR()) {
+ return InVMTemporaryTransport.getInstance().pickup(epr, millis) ;
} else {
- message = (Message) messageObj;
+ return InVMTransport.getInstance().pickup(epr, millis) ;
}
-
- if ((message != null) && isTransactional())
- {
- /*
- * Return the message, but don't remove it from the queue until the transaction
- * commits. If the transaction rolls back then the message may not go back into the
- * queue at the exact place it was originally: other messages may have been removed
- * successfully by other threads. Plus, we would have to maintain a before and after
- * image of the queue. This is more a compensation transaction.
- */
-
- TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true);
- InVMXAResource theResource = new InVMXAResource(this, message, InVMXAResource.operation.REMOVE);
-
- txStrategy.enlistResource(theResource);
-
- // now we can deliver the message.
- }
}
catch (final Throwable ex)
{
logger.warn("InVMCourier pickup caught: "+ex);
+ logger.debug("InVMCourier pickup caught", ex);
return null;
}
-
- return message;
}
public void cleanup() {
}
-
- private void assertCourierActive() throws CourierException {
- if (!isActive) {
- throw new CourierException("InVMCourier for EPR '" + epr.getAddr().getAddress() + "' is not active. Cannot pickup/deliver on this courier.");
- }
- }
-
- public boolean isActive() {
- return isActive;
- }
-
- public void setActive(boolean active) {
- isActive = active;
-
- if(!active && !messageQueue.isEmpty()) {
- logger.error("Deactivated non-empty InVM Queue. Undelivered messages will be lost.");
- messageQueue.clear();
- }
- }
-
- public void clean() {
- }
-
- public void reset() {
- try
- {
- if (isTransactional())
- logger.warn("InVMCourier reset called on transactional courier: ignoring reset for the sake of consistency!");
- else
- messageQueue.clear();
- }
- catch (final Throwable ex)
- {
- logger.debug("InVMCourier reset caught: "+ex);
- }
- }
-
- /**
- * Used by transactional resource to deliver the message to the queue
- * in the event of a transaction commit.
- * Almost the same as normal delivery, except no lockstep wait. Lockstep
- * does not make sense in the scope of queued transactional delivery.
- */
-
- public boolean doDeliver (Message message)
- {
- synchronized (messageQueue)
- {
- Object addedObject = addMessageToQueue(message);
-
- if (addedObject == null) {
- return false;
- }
-
- // Notify 1 waiting pickup thread of the delivery...
- messageQueue.notify();
- }
-
- return true;
- }
-
- /**
- * Used by transactional resource to place message back on the queue. We remove from the head
- * but insert to the tail, for speed. If individual messages need to be delivered in some well
- * defined order then this could cause problems. However, there are no ordering guarantees with
- * a range of transports, and particularly not in an asynchronous environment with arbitrary
- * network latency.
- *
- * If this does cause a problem then we should revisit. It will mean creating a fully transactional
- * queue (done already in Arjuna so we may be able to lift that code). But that will affect
- * performance. It may be easier to look at solving the ordering problem separately.
- */
-
- public boolean doRedeliver (Message message)
- {
- synchronized (messageQueue) {
- messageQueue.add(message);
- }
-
- return true;
- }
-
- /*
- * TODO this is used in a number of classes so should be a separate
- * util routine.
- */
-
- private boolean isTransactional() throws CourierException {
- boolean transactional;
- try
- {
- TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true);
- Object txHandle = ((txStrategy == null) ? null : txStrategy.getTransaction());
- boolean isActive = ((txStrategy == null) ? false : txStrategy.isActive());
-
- transactional = (txHandle != null);
-
- /*
- * Make sure the current transaction is still active! If we
- * have previously slept, then the timeout may be longer than that
- * associated with the transaction.
- */
-
- if (transactional && !isActive)
- {
- throw new CourierException("Associated transaction is no longer active!");
- }
- }
- catch (TransactionStrategyException ex)
- {
- throw new CourierException(ex);
- }
- return transactional;
- }
}
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/TwoWayCourierImpl.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/TwoWayCourierImpl.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/TwoWayCourierImpl.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -125,7 +125,7 @@
if (null == epr)
return null;
if (epr instanceof InVMEpr)
- return CourierFactory.getInstance().getInVMCourier((InVMEpr) epr);
+ return new InVMCourier((InVMEpr)epr);
if (epr instanceof JMSEpr)
return new JmsCourier((JMSEpr) epr, pickUpOnly);
if (epr instanceof FileEpr)
@@ -146,7 +146,7 @@
addr = epr.getAddr().getAddress();
if (addr.startsWith(InVMEpr.INVM_PROTOCOL))
- return CourierFactory.getInstance().getInVMCourier(new InVMEpr(epr));
+ return new InVMCourier(new InVMEpr(epr));
if (addr.startsWith(JMSEpr.JMS_PROTOCOL))
return new JmsCourier(new JMSEpr(epr), pickUpOnly);
if (addr.startsWith(JDBCEpr.JDBC_PROTOCOL))
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMException.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMException.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMException.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,72 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2006, 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.internal.soa.esb.couriers.transport;
+
+/**
+ * The exception raised for InVM transport errors.
+ *
+ * @author <a href='mailto:kevin.conner at jboss.com'>Kevin Conner</a>
+ */
+public class InVMException extends Exception
+{
+ /**
+ * serial version UID of this exception.
+ */
+ private static final long serialVersionUID = 3419916783144269048L;
+
+ /**
+ * Construct a default InVM exception.
+ */
+ public InVMException()
+ {
+ }
+
+ /**
+ * Construct an InVM exception with the specified message.
+ * @param message The exception message.
+ */
+ public InVMException(final String message)
+ {
+ super(message);
+ }
+
+ /**
+ * Construct an InVM exception with the associated cause.
+ * @param cause The associated cause.
+ */
+ public InVMException(final Throwable cause)
+ {
+ super(cause);
+ }
+
+ /**
+ * Construct an InVM exception with the specified message and associated cause.
+ * @param message The exception message.
+ * @param cause The associated cause.
+ */
+ public InVMException(String message, Throwable cause)
+ {
+ super(message, cause);
+ }
+}
+
\ No newline at end of file
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMException.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransport.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransport.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransport.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,1016 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2006, 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.internal.soa.esb.couriers.transport;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.log4j.Logger;
+import org.jboss.internal.soa.esb.couriers.tx.InVMXAResource;
+import org.jboss.internal.soa.esb.message.format.MessageSerializer;
+import org.jboss.soa.esb.addressing.eprs.InVMEpr;
+import org.jboss.soa.esb.common.Environment;
+import org.jboss.soa.esb.common.ModulePropertyManager;
+import org.jboss.soa.esb.common.TransactionStrategy;
+import org.jboss.soa.esb.common.TransactionStrategyException;
+import org.jboss.soa.esb.message.ByReferenceMessage;
+import org.jboss.soa.esb.message.Message;
+
+import com.arjuna.common.util.propertyservice.PropertyManager;
+
+/**
+ * Provide support for the InVM temporary transports.
+ *
+ * The InVM transport is divided into two distinct types, permanent transports which
+ * represent ESB aware services and temporary transports which represent default reply
+ * queues.
+ *
+ * This class provides the support for temporary queues while support for the
+ * registry queries/permanent transports are through the InVMTransport class.
+ *
+ * @author kevin
+ */
+public class InVMTemporaryTransport
+{
+ /**
+ * The logger for this instance.
+ */
+ private static final Logger LOGGER = Logger.getLogger(InVMTemporaryTransport.class) ;
+ /**
+ * The default expiry time.
+ */
+ private static final long DEFAULT_EXPIRY_TIME = 300000;
+ /**
+ * The expiry time for messages.
+ */
+ private final long expiryTime ;
+ /**
+ * Mapping of service ids to entries.
+ */
+ private final Map<String, InVMEntry> serviceIdToEntry = new HashMap<String, InVMEntry>() ;
+ /**
+ * Ordered entries for reaping.
+ */
+ private final LinkedHashSet<InVMQueueEntry> orderedEntries = new LinkedHashSet<InVMQueueEntry>() ;
+ /**
+ * The lock guarding access and modification to the structures.
+ */
+ private ReadWriteLock lock = new ReentrantReadWriteLock() ;
+
+ /**
+ * Reaper thread
+ */
+ private volatile ReaperThread reaperThread ;
+
+ /**
+ * Factory singleton instance.
+ */
+ private static InVMTemporaryTransport instance = new InVMTemporaryTransport();
+
+ /**
+ * Get the InVM Transport.
+ * @return The InVM Transport instance.
+ */
+ public static InVMTemporaryTransport getInstance()
+ {
+ return instance;
+ }
+
+ /**
+ * Deliver a message to the specified EPR.
+ * @param inVMEpr The EPR to receive the message.
+ * @param message The message to deliver.
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public void deliver(final InVMEpr inVMEpr, final Message message)
+ throws InVMException
+ {
+ final String serviceId = inVMEpr.getAddr().getAddress() ;
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Delivering message to " + serviceId) ;
+ }
+ final Object addedObject ;
+ if (inVMEpr.getPassByValue())
+ {
+ try
+ {
+ addedObject = MessageSerializer.serialize(message) ;
+ }
+ catch (final IOException ex)
+ {
+ throw new InVMException("Could not serialize message to pass by value.", ex) ;
+ }
+ }
+ else if (message instanceof ByReferenceMessage)
+ {
+ addedObject = ((ByReferenceMessage)message).reference() ;
+ }
+ else
+ {
+ addedObject = message ;
+ }
+
+ if (InVMTransport.isTransactional())
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Delivering transactional message to " + serviceId) ;
+ }
+ /*
+ * Can't do lockstep wait here because otherwise the transaction may not terminate if this
+ * is the transaction controller thread!
+ */
+ final TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true) ;
+ try
+ {
+ txStrategy.enlistResource(new InVMXAResource(inVMEpr, addedObject, InVMXAResource.operation.INSERT));
+ }
+ catch (final TransactionStrategyException tse)
+ {
+ throw new InVMException("Unexpected error enlisting transaction resource", tse) ;
+ }
+ }
+ else
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Delivering message direct to " + serviceId + " queue") ;
+ }
+
+ final long lockstep = (inVMEpr.getLockstep() ? inVMEpr.getLockstepWaitTime() : 0) ;
+
+ final InVMEntry entry ;
+ acquireReadLock() ;
+ try
+ {
+ entry = getEntry(serviceId) ;
+ entry.incDelivery() ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ deliver(entry, addedObject, lockstep) ;
+ }
+ }
+
+ /**
+ * Pickup a message for the specified EPR.
+ * @param inVMEpr The EPR to receive the message.
+ * @param millis The number of milliseconds to wait for a message.
+ * @return The message or null if nothing present within the timeout.
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public Message pickup(final InVMEpr inVMEpr, final long millis)
+ throws InVMException
+ {
+ final String serviceId = inVMEpr.getAddr().getAddress() ;
+ final InVMEntry entry ;
+ acquireReadLock() ;
+ try
+ {
+ entry = getEntry(serviceId) ;
+ entry.incPickup() ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+
+ final InVMQueueEntry queueEntry = entry.pickup(millis) ;
+ clearEntry(entry, queueEntry) ;
+ if (queueEntry != null)
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Pickup of message from " + serviceId) ;
+ }
+ final Object msgObject = queueEntry.getValue() ;
+ if (msgObject != null)
+ {
+ final Message message ;
+ try
+ {
+ if (msgObject instanceof byte[])
+ {
+ message = MessageSerializer.deserialize((byte[])msgObject) ;
+ }
+ else if (inVMEpr.getPassByValue())
+ {
+ // pass by reference but now expecting value.
+ message = ((Message)msgObject).copy() ;
+ }
+ else
+ {
+ message = (Message)msgObject ;
+ }
+ }
+ catch (final IOException ioe)
+ {
+ throw new InVMException("Failed to deserialise incoming message", ioe) ;
+ }
+
+ if (InVMTransport.isTransactional())
+ {
+ /*
+ * Return the message, but don't remove it from the queue until the transaction
+ * commits. If the transaction rolls back then the message may not go back into the
+ * queue at the exact place it was originally: other messages may have been removed
+ * successfully by other threads. Plus, we would have to maintain a before and after
+ * image of the queue. This is more a compensation transaction.
+ */
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Pickup enlisting transactional resource for service " + serviceId) ;
+ }
+ final TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true) ;
+ try
+ {
+ txStrategy.enlistResource(new InVMXAResource(inVMEpr, msgObject, InVMXAResource.operation.REMOVE)) ;
+ }
+ catch (final TransactionStrategyException tse)
+ {
+ throw new InVMException("Unexpected error enlisting transaction resource", tse) ;
+ }
+ }
+ return message ;
+ }
+ }
+ return null ;
+ }
+
+ /**
+ * Deliver an object as a consequence of a transaction. This will either be a
+ * rollback, placing the object back on the source queue, or a commit delivering
+ * to a target queue.
+ *
+ * @param inVMEpr The EPR to receive the message.
+ * @param msgObject The object to deliver.
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public void deliverTx(final InVMEpr inVMEpr, final Object msgObject)
+ throws InVMException
+ {
+ final String serviceId = inVMEpr.getAddr().getAddress() ;
+
+ final InVMEntry entry ;
+ acquireReadLock() ;
+ try
+ {
+ entry = getEntry(serviceId) ;
+ entry.incDelivery() ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Transactional redelivery of message to " + serviceId) ;
+ }
+
+ deliver(entry, msgObject, 0) ;
+ }
+
+ /**
+ * Start the reaper thread.
+ */
+ public void startReaper()
+ {
+ LOGGER.info("Starting reaper thread") ;
+ reaperThread = new ReaperThread() ;
+ reaperThread.start() ;
+ }
+
+ /**
+ * Stop the reaper thread.
+ */
+ public void stopReaper()
+ {
+ final ReaperThread currentReaperThread = reaperThread ;
+ reaperThread = null ;
+ if (currentReaperThread != null)
+ {
+ LOGGER.info("Stopping reaper thread") ;
+ currentReaperThread.shutdown() ;
+ }
+ }
+
+ /**
+ * Deliver to the queue, adding to the reaper list if necessary.
+ *
+ * N.B. Must not hold external lock while calling this method.
+ *
+ * @param entry The invm entry containing the queue.
+ * @param value The value being appended to the queue.
+ * @param lockstep The lockstep timeout or 0 if not required.
+ */
+ private void deliver(final InVMEntry entry, final Object value, final long lockstep)
+ throws InVMException
+ {
+ final InVMQueueEntry queueEntry = entry.deliver(value, lockstep, expiryTime) ;
+ if (entry.isActive(queueEntry))
+ {
+ final boolean notifyReaper ;
+ acquireWriteLock() ;
+ try
+ {
+ if (entry.isActive(queueEntry))
+ {
+ notifyReaper = orderedEntries.isEmpty() ;
+ orderedEntries.add(queueEntry) ;
+ }
+ else
+ {
+ notifyReaper = false ;
+ }
+ }
+ finally
+ {
+ releaseWriteLock() ;
+ }
+ if (notifyReaper)
+ {
+ final ReaperThread currentReaperThread = reaperThread ;
+ if (currentReaperThread != null)
+ {
+ currentReaperThread.entriesUpdated() ;
+ }
+ }
+ }
+ }
+
+ /**
+ * Get the size of the mapping for service ids to entries.
+ * @return The size of the mapping.
+ */
+ public int getServiceIdToEntrySize()
+ {
+ acquireReadLock() ;
+ try
+ {
+ return serviceIdToEntry.size() ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ }
+
+ /**
+ * Get the size of the ordered entries for reaping.
+ * @return The size of the ordered entries.
+ */
+ public int getOrderedEntriesSize()
+ {
+ acquireReadLock() ;
+ try
+ {
+ return orderedEntries.size() ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ }
+
+ /**
+ * Acquire a read lock for accessing the data.
+ */
+ private void acquireReadLock()
+ {
+ lock.readLock().lock() ;
+ }
+
+ /**
+ * Release a read lock for accessing the data.
+ */
+ private void releaseReadLock()
+ {
+ lock.readLock().unlock() ;
+ }
+
+ /**
+ * Acquire a write lock for accessing the data.
+ */
+ private void acquireWriteLock()
+ {
+ lock.writeLock().lock() ;
+ }
+
+ /**
+ * Release a write lock for accessing the data.
+ */
+ private void releaseWriteLock()
+ {
+ lock.writeLock().unlock() ;
+ }
+
+ /**
+ * Promote to a write lock for accessing the data.
+ */
+ private void promoteToWriteLock()
+ {
+ lock.readLock().unlock() ;
+ lock.writeLock().lock() ;
+ }
+
+ /**
+ * Demote to a read lock for accessing the data.
+ */
+ private void demoteToReadLock()
+ {
+ lock.readLock().lock() ;
+ lock.writeLock().unlock() ;
+ }
+
+ /**
+ * Get an entry representing the serviceId.
+ *
+ * Must hold read lock when calling this method.
+ *
+ * @param serviceId The service id to locate.
+ * @return The entry associated with the service id.
+ */
+ private InVMEntry getEntry(final String serviceId)
+ {
+ InVMEntry entry = serviceIdToEntry.get(serviceId) ;
+ if (entry == null)
+ {
+ promoteToWriteLock() ;
+ try
+ {
+ entry = new InVMEntry(serviceId) ;
+ final InVMEntry current = serviceIdToEntry.put(serviceId, entry) ;
+ if ((current != null) && (current != entry))
+ {
+ entry = current ;
+ serviceIdToEntry.put(serviceId, current) ;
+ }
+ }
+ finally
+ {
+ demoteToReadLock() ;
+ }
+ }
+ return entry ;
+ }
+
+ /**
+ * Clear the queue entry.
+ *
+ * @param entry The service entry to clear.
+ * @param queueEntry The queue entry to clear or null if not required.
+ */
+ private void clearEntry(final InVMEntry entry, final InVMQueueEntry queueEntry)
+ {
+ acquireWriteLock() ;
+ try
+ {
+ if (queueEntry != null)
+ {
+ orderedEntries.remove(queueEntry) ;
+ }
+
+ if (entry.isFree())
+ {
+ final String serviceId = entry.getServiceId() ;
+ final InVMEntry current = serviceIdToEntry.remove(serviceId) ;
+ if ((current != null) && (current != entry))
+ {
+ serviceIdToEntry.put(serviceId, current) ;
+ }
+ }
+ }
+ finally
+ {
+ releaseWriteLock() ;
+ }
+ }
+
+ /**
+ * Data representing an entry in the message queue.
+ * @author kevin
+ */
+ private static class InVMQueueEntry
+ {
+ /**
+ * The associated InVM entry.
+ */
+ private final InVMEntry entry ;
+ /**
+ * The value enqueued.
+ */
+ private final Object value ;
+ /**
+ * Condition associated with lockstep.
+ */
+ private final Condition condition ;
+ /**
+ * Expiry of this entry in milli seconds.
+ */
+ private final long expiry ;
+ /**
+ * The flag indicating that this entry is now inactive.
+ */
+ private boolean inactive ;
+
+ /**
+ * Construct the InVM queue entry.
+ *
+ * @param entry The associated InVM entry.
+ * @param value The value being enqueued.
+ * @param condition The condition representing the lockstep or null if not required.
+ * @param expiry The expiry time of this entry.
+ */
+ InVMQueueEntry(final InVMEntry entry, final Object value, final Condition condition,
+ final long expiry)
+ {
+ this.entry = entry ;
+ this.value = value ;
+ this.condition = condition ;
+ this.expiry = expiry ;
+ }
+
+ /**
+ * Get the associated entry.
+ * @return The associated entry.
+ */
+ InVMEntry getEntry()
+ {
+ return entry ;
+ }
+
+ /**
+ * Get the enqueued value.
+ * @return The enqueued value.
+ */
+ Object getValue()
+ {
+ return value ;
+ }
+
+ /**
+ * Get the lockstep condition.
+ * @return The lockstep condition or null if not necessary.
+ */
+ Condition getCondition()
+ {
+ return condition ;
+ }
+
+ /**
+ * Get the expiry time of this entry.
+ * @return The expiry time.
+ */
+ long getExpiryTime()
+ {
+ return expiry ;
+ }
+
+ /**
+ * Set the inactive flag.
+ * Must be invoked with InVMEntry lock
+ */
+ void setInactive()
+ {
+ inactive = true ;
+ }
+
+ /**
+ * Get the inactive flag.
+ * Must be invoked with InVMEntry lock
+ */
+ boolean isInactive()
+ {
+ return inactive ;
+ }
+ }
+
+ /**
+ * The InVM entries in the data structures.
+ * @author kevin
+ */
+ private static class InVMEntry
+ {
+ /**
+ * The lock for the queue.
+ */
+ private final Lock lock = new ReentrantLock() ;
+ /**
+ * The condition on which to await messages.
+ */
+ private final Condition waitingCondition = lock.newCondition() ;
+ /**
+ * Entries in the queue.
+ */
+ private final Queue<InVMQueueEntry> entries = new LinkedList<InVMQueueEntry>() ;
+ /**
+ * The service id for this entry.
+ */
+ private final String serviceId ;
+ /**
+ * The number of waiters on the waiting condition.
+ */
+ private int numWaiters ;
+ /**
+ * The number of deliveries about to occur.
+ */
+ private int numDeliveries ;
+ /**
+ * The number of pickups about to occur.
+ */
+ private int numPickups ;
+
+ /**
+ * Create an entry with the specified service id.
+ * @param serviceId The service id.
+ */
+ InVMEntry(final String serviceId)
+ {
+ this.serviceId = serviceId ;
+ }
+
+ /**
+ * Get the service id associated with this entry.
+ * @return the service id.
+ */
+ String getServiceId()
+ {
+ return serviceId ;
+ }
+
+ /**
+ * Increment the number of deliveries about to occur.
+ * Can hold external read lock.
+ */
+ void incDelivery()
+ {
+ lock.lock() ;
+ try
+ {
+ numDeliveries++ ;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Increment the number of pickups about to occur.
+ * Can hold external read lock.
+ */
+ void incPickup()
+ {
+ lock.lock() ;
+ try
+ {
+ numPickups++ ;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+
+ /**
+ * Is this entry free?
+ * @return true if there are no entries, waiters, deliveries or pickups about to occur.
+ */
+ boolean isFree()
+ {
+ lock.lock() ;
+ try
+ {
+ return ((numDeliveries == 0) && (numPickups == 0) && (numWaiters == 0) && (entries.size() == 0)) ;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Deliver the specified value onto the queue.
+ *
+ * N.B. Must not hold external lock while calling this method.
+ *
+ * @param value The value being appended to the queue.
+ * @param lockstep The lockstep timeout or 0 if not required.
+ * @param expiryTime The expiry time for the message.
+ *
+ * @throws InVMException for InVM transport specific errors.
+ */
+ InVMQueueEntry deliver(final Object value, final long lockstep, final long expiryTime)
+ throws InVMException
+ {
+ lock.lock() ;
+ try
+ {
+ numDeliveries-- ;
+ final Condition condition = (lockstep > 0 ? lock.newCondition() : null) ;
+ final InVMQueueEntry queueEntry = new InVMQueueEntry(this, value, condition, System.currentTimeMillis() + expiryTime) ;
+ if (!entries.offer(queueEntry))
+ {
+ throw new InVMException("Failed to append message to InVM queue") ;
+ }
+ if (numWaiters > 0)
+ {
+ waitingCondition.signal() ;
+ }
+
+ if (condition != null)
+ {
+ try
+ {
+ condition.await(lockstep, TimeUnit.MILLISECONDS) ;
+ }
+ catch (final InterruptedException ie)
+ {
+ LOGGER.warn("Waiting delivery thread interupted while waiting on message pickup on InVM queue '" + serviceId + "'. Exiting pickup wait state.") ;
+ }
+ }
+ return queueEntry ;
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+
+ /**
+ * Pickup an entry from the queue.
+ *
+ * N.B. Must not hold external lock while calling this method.
+ *
+ * @param millis The number of milliseconds to wait for a message.
+ * @return The message or null if nothing present within the timeout.
+ *
+ * @throws InVMException for InVM transport specific errors.
+ */
+ InVMQueueEntry pickup(final long millis)
+ throws InVMException
+ {
+ final long end = System.currentTimeMillis() + millis ;
+ lock.lock() ;
+ try
+ {
+ numPickups-- ;
+ if (entries.isEmpty())
+ {
+ final long delay = end - System.currentTimeMillis() ;
+ if (delay > 0)
+ {
+ numWaiters++ ;
+ try
+ {
+ waitingCondition.await(delay, TimeUnit.MILLISECONDS) ;
+ }
+ catch (final InterruptedException ioe) {} // ignore
+ finally
+ {
+ numWaiters-- ;
+ }
+ }
+ }
+ final InVMQueueEntry entry = entries.poll() ;
+ if (entry != null)
+ {
+ entry.setInactive() ;
+ final Condition condition = entry.getCondition() ;
+ if (condition != null)
+ {
+ condition.signal() ;
+ }
+ return entry ;
+ }
+ return null ;
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+
+ /**
+ * Is the specified queue entry still active?
+ * @param queueEntry The queue entry to check.
+ * @return true if still active, false otherwise.
+ */
+ boolean isActive(final InVMQueueEntry queueEntry)
+ {
+ lock.lock() ;
+ try
+ {
+ return !queueEntry.isInactive() ;
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+
+ /**
+ * Expire the specified entry.
+ * @param queueEntry The queue entry to expire.
+ */
+ void expire(final InVMQueueEntry queueEntry)
+ {
+ lock.lock() ;
+ try
+ {
+ if (entries.remove(queueEntry))
+ {
+ queueEntry.setInactive() ;
+ }
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+ }
+
+ /**
+ * The reaper thread for expired tasks.
+ * @author kevin
+ */
+ private class ReaperThread extends Thread
+ {
+ /**
+ * Shutdown flag.
+ */
+ private AtomicBoolean shutdown = new AtomicBoolean() ;
+ /**
+ * Lock for notifications.
+ */
+ private Lock lock = new ReentrantLock() ;
+ /**
+ * Notify condition.
+ */
+ private Condition notifyCondition = lock.newCondition() ;
+ /**
+ * The notify flag.
+ */
+ private boolean notify ;
+
+ public void run()
+ {
+ while(!shutdown.get())
+ {
+ long nextExpiry = Long.MAX_VALUE ;
+ acquireWriteLock() ;
+ try
+ {
+ final Iterator<InVMQueueEntry> orderedEntriesIter = orderedEntries.iterator() ;
+ if (orderedEntriesIter.hasNext())
+ {
+ final long now = System.currentTimeMillis() ;
+ do
+ {
+ final InVMQueueEntry queueEntry = orderedEntriesIter.next() ;
+ final long expiryTime = queueEntry.getExpiryTime() ;
+ if (expiryTime <= now)
+ {
+ orderedEntriesIter.remove() ;
+ final InVMEntry entry = queueEntry.getEntry();
+ entry.expire(queueEntry) ;
+ clearEntry(queueEntry.getEntry(), null) ;
+ }
+ else
+ {
+ nextExpiry = expiryTime ;
+ break ;
+ }
+ }
+ while(orderedEntriesIter.hasNext()) ;
+ }
+ }
+ finally
+ {
+ releaseWriteLock() ;
+ }
+ LOGGER.debug("Reaper thread next expiry: " + nextExpiry) ;
+ lock.lock() ;
+ try
+ {
+ if (!notify && !shutdown.get())
+ {
+ final long delay = nextExpiry - System.currentTimeMillis() ;
+ if (delay > 0)
+ {
+ try
+ {
+ notifyCondition.await(delay, TimeUnit.MILLISECONDS) ;
+ }
+ catch (final InterruptedException ie) {} // ignore
+ }
+ }
+ notify = false ;
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+ }
+
+ /**
+ * Notify the reaper that entries have been updated.
+ * Must not be called with locks held.
+ */
+ void entriesUpdated()
+ {
+ lock.lock() ;
+ try
+ {
+ notify = true ;
+ notifyCondition.signal() ;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Indicate a reaper shutdown.
+ */
+ void shutdown()
+ {
+ shutdown.set(true) ;
+ lock.lock() ;
+ try
+ {
+ notifyCondition.signal() ;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+ }
+
+ {
+ final PropertyManager prop = ModulePropertyManager.getPropertyManager(ModulePropertyManager.TRANSPORTS_MODULE);
+ final String value = prop.getProperty(Environment.INVM_EXPIRY_TIME);
+
+ long expiryTime = DEFAULT_EXPIRY_TIME ;
+ if (value != null)
+ {
+ try
+ {
+ final long parsedExpiryTime = Long.parseLong(value);
+ if (parsedExpiryTime > 0)
+ {
+ expiryTime = parsedExpiryTime ;
+ }
+ else
+ {
+ LOGGER.warn("Invalid InVM expiry time, using default") ;
+ }
+ }
+ catch (final NumberFormatException ex)
+ {
+ LOGGER.warn("Failed to parse InVM expiry time, using default") ;
+ }
+ }
+ this.expiryTime = expiryTime ;
+ LOGGER.debug("InVM expiry time set to " + expiryTime) ;
+ }
+}
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransport.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransport.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransport.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransport.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,951 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2006, 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.internal.soa.esb.couriers.transport;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.log4j.Logger;
+import org.jboss.internal.soa.esb.couriers.tx.InVMXAResource;
+import org.jboss.internal.soa.esb.message.format.MessageSerializer;
+import org.jboss.soa.esb.addressing.EPR;
+import org.jboss.soa.esb.addressing.eprs.InVMEpr;
+import org.jboss.soa.esb.common.TransactionStrategy;
+import org.jboss.soa.esb.common.TransactionStrategyException;
+import org.jboss.soa.esb.message.ByReferenceMessage;
+import org.jboss.soa.esb.message.Message;
+import org.jboss.soa.esb.services.registry.ServiceNotFoundException;
+
+/**
+ * Provide support for the InVM transports and EPRs.
+ *
+ * The InVM transport is divided into two distinct types, permanent transports which
+ * represent ESB aware services and temporary transports which represent default reply
+ * queues.
+ *
+ * This class provides the interface and support for registry queries/permanent transports
+ * while support for the temporary queues are through the InVMTemporaryTransport class.
+ *
+ * @author kevin
+ */
+public class InVMTransport
+{
+ /**
+ * The logger for this instance.
+ */
+ private static final Logger LOGGER = Logger.getLogger(InVMTransport.class) ;
+
+ /**
+ * Mapping of service ids to entries.
+ */
+ private final Map<String, InVMEntry> serviceIdToEntry = new HashMap<String, InVMEntry>() ;
+ /**
+ * Mapping of category names to service names and entries.
+ */
+ private final Map<String, Map<String, InVMEntry>> categoryToNameToEntry = new HashMap<String, Map<String,InVMEntry>>() ;
+ /**
+ * The lock guarding access and modification to the structures.
+ */
+ private ReadWriteLock lock = new ReentrantReadWriteLock() ;
+
+ /**
+ * Factory singleton instance.
+ */
+ private static InVMTransport instance = new InVMTransport();
+
+ /**
+ * Get the InVM Transport.
+ * @return The InVM Transport instance.
+ */
+ public static InVMTransport getInstance()
+ {
+ return instance;
+ }
+
+ /**
+ * Register the specific EPR against the service category/name.
+ *
+ * @param category The service category.
+ * @param name The service name.
+ * @param epr The associated InVM EPR.
+ *
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public void registerEPR(final String category, final String name, final InVMEpr epr)
+ throws InVMException
+ {
+ if (epr.isTemporaryEPR())
+ {
+ throw new InVMException("Attempt to register temporary EPR in permanent registry") ;
+ }
+
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Registering EPR " + epr + " for category " + category + ", name " + name) ;
+ }
+ final String serviceId = epr.getServiceId() ;
+ acquireWriteLock() ;
+ try
+ {
+ final InVMEntry existingEntry = serviceIdToEntry.get(serviceId) ;
+ Map<String, InVMEntry> nameToEntry = categoryToNameToEntry.get(category) ;
+ if (nameToEntry == null)
+ {
+ if (existingEntry != null)
+ {
+ throw new InVMException("Service " + serviceId + " registered under a multiple categories") ;
+ }
+ nameToEntry = new HashMap<String, InVMEntry>() ;
+ categoryToNameToEntry.put(category, nameToEntry) ;
+ }
+
+ InVMEntry entry = nameToEntry.get(name) ;
+ if (entry == null)
+ {
+ if (existingEntry != null)
+ {
+ throw new InVMException("Service " + serviceId + " registered under a multiple names") ;
+ }
+ entry = new InVMEntry(serviceId) ;
+ nameToEntry.put(name, entry) ;
+ }
+ else if ((existingEntry != null) && (existingEntry != entry))
+ {
+ throw new InVMException("Service " + serviceId + " registered under a multiple names") ;
+ }
+
+ if (entry.addEPR(epr))
+ {
+ serviceIdToEntry.put(serviceId, entry) ;
+ }
+ }
+ finally
+ {
+ releaseWriteLock() ;
+ }
+ }
+
+ /**
+ * Unregister the service category/name.
+ *
+ * @param category The service category.
+ * @param name The service name.
+ * @return true if the service was removed, false otherwise
+ */
+ public boolean unRegisterService(final String category, final String name)
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Unregistering service category " + category + ", name " + name) ;
+ }
+ acquireWriteLock() ;
+ try
+ {
+ Map<String, InVMEntry> nameToEntry = categoryToNameToEntry.get(category) ;
+ if (nameToEntry == null)
+ {
+ return false ;
+ }
+
+ InVMEntry entry = nameToEntry.remove(name) ;
+ if (entry == null)
+ {
+ return false ;
+ }
+ entry.shutdown() ;
+ if (nameToEntry.isEmpty())
+ {
+ categoryToNameToEntry.remove(category) ;
+ }
+ serviceIdToEntry.remove(entry.getServiceId()) ;
+ return true ;
+ }
+ finally
+ {
+ releaseWriteLock() ;
+ }
+ }
+
+ /**
+ * Unregister the specific EPR from the service category/name.
+ *
+ * @param category The service category.
+ * @param name The service name.
+ * @param epr The associated InVM EPR.
+ *
+ * @throws ServiceNotFoundException if the service/EPR combination is not found.
+ */
+ public void unRegisterEPR(final String category, final String name, final InVMEpr epr)
+ throws ServiceNotFoundException
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Unregistering EPR " + epr + " for category " + category + ", name " + name) ;
+ }
+ acquireWriteLock() ;
+ try
+ {
+ Map<String, InVMEntry> nameToEntry = categoryToNameToEntry.get(category) ;
+ if (nameToEntry == null)
+ {
+ throw new ServiceNotFoundException("Could not locate any services for category " + category) ;
+ }
+
+ InVMEntry entry = nameToEntry.get(name) ;
+ if (entry == null)
+ {
+ throw new ServiceNotFoundException("Could not locate service " + category + ", " + name) ;
+ }
+ if (entry.removeEPR(epr))
+ {
+ nameToEntry.remove(name) ;
+ entry.shutdown() ;
+ if (nameToEntry.isEmpty())
+ {
+ categoryToNameToEntry.remove(category) ;
+ }
+ serviceIdToEntry.remove(entry.getServiceId()) ;
+ }
+ }
+ finally
+ {
+ releaseWriteLock() ;
+ }
+ }
+
+ /**
+ * Retrieve a list of all known services.
+ * @return The known services.
+ */
+ public List<String> findAllServices()
+ {
+ final List<String> result = new ArrayList<String>() ;
+
+ acquireReadLock() ;
+ try
+ {
+ for(Map<String, InVMEntry> nameToEntry: categoryToNameToEntry.values())
+ {
+ result.addAll(nameToEntry.keySet()) ;
+ }
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ return result ;
+ }
+
+ /**
+ * Retrieve a list of all known services in a specified category.
+ * @param category The category to query.
+ * @return The known services.
+ */
+ public List<String> findServices(final String category)
+ {
+ final List<String> result = new ArrayList<String>() ;
+
+ acquireReadLock() ;
+ try
+ {
+ final Map<String, InVMEntry> nameToEntry = categoryToNameToEntry.get(category) ;
+ if (nameToEntry != null)
+ {
+ result.addAll(nameToEntry.keySet()) ;
+ }
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ return result ;
+ }
+
+ /**
+ * Retrieve a list of EPRs for the specified category and name.
+ * @param category The service category to query.
+ * @param name The service name to query.
+ * @return The known EPRs.
+ */
+ public List<EPR> findEPRs(final String category, final String name)
+ {
+ final List<EPR> result = new ArrayList<EPR>() ;
+
+ acquireReadLock() ;
+ try
+ {
+ final Map<String, InVMEntry> nameToEntry = categoryToNameToEntry.get(category) ;
+ if (nameToEntry != null)
+ {
+ final InVMEntry entry = nameToEntry.get(name) ;
+ if (entry != null)
+ {
+ result.addAll(entry.getEPRs()) ;
+ }
+ }
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ return result ;
+ }
+
+ /**
+ * Retrieve an EPR for the specified category and name.
+ * @param category The service category to query.
+ * @param name The service name to query.
+ * @return An EPR for the specified category and name or null if not registered.
+ */
+ public EPR findEPR(final String category, final String name)
+ {
+ acquireReadLock() ;
+ try
+ {
+ final Map<String, InVMEntry> nameToEntry = categoryToNameToEntry.get(category) ;
+ if (nameToEntry != null)
+ {
+ final InVMEntry entry = nameToEntry.get(name) ;
+ if (entry != null)
+ {
+ return entry.getEPRs().get(0) ;
+ }
+ }
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ return null;
+ }
+
+ /**
+ * Deliver a message to the specified EPR.
+ * @param inVMEpr The EPR to receive the message.
+ * @param message The message to deliver.
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public void deliver(final InVMEpr inVMEpr, final Message message)
+ throws InVMException
+ {
+ final String serviceId = inVMEpr.getServiceId() ;
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Delivering message to " + serviceId) ;
+ }
+ final InVMEntry entry ;
+ final boolean passByValue ;
+ final long lockstep ;
+ acquireReadLock() ;
+ try
+ {
+ entry = serviceIdToEntry.get(serviceId) ;
+ passByValue = entry.isPassByValue() ;
+ lockstep = entry.getLockstep() ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ if (entry == null)
+ {
+ throw new InVMException("Could not locate service entry for epr " + inVMEpr) ;
+ }
+
+ final Object addedObject ;
+ if (passByValue)
+ {
+ try
+ {
+ addedObject = MessageSerializer.serialize(message) ;
+ }
+ catch (final IOException ex)
+ {
+ throw new InVMException("Could not serialize message to pass by value.", ex) ;
+ }
+ }
+ else if (message instanceof ByReferenceMessage)
+ {
+ addedObject = ((ByReferenceMessage)message).reference() ;
+ }
+ else
+ {
+ addedObject = message ;
+ }
+
+ if (isTransactional())
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Delivering transactional message to " + serviceId) ;
+ }
+ /*
+ * Can't do lockstep wait here because otherwise the transaction may not terminate if this
+ * is the transaction controller thread!
+ */
+ final TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true) ;
+ try
+ {
+ txStrategy.enlistResource(new InVMXAResource(inVMEpr, addedObject, InVMXAResource.operation.INSERT));
+ }
+ catch (final TransactionStrategyException tse)
+ {
+ throw new InVMException("Unexpected error enlisting transaction resource", tse) ;
+ }
+ }
+ else
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Delivering message direct to " + serviceId + " queue") ;
+ }
+ entry.deliver(addedObject, lockstep) ;
+ }
+ }
+
+ /**
+ * Pickup a message for the specified EPR.
+ * @param inVMEpr The EPR to receive the message.
+ * @param millis The number of milliseconds to wait for a message.
+ * @return The message or null if nothing present within the timeout.
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public Message pickup(final InVMEpr inVMEpr, final long millis)
+ throws InVMException
+ {
+ final String serviceId = inVMEpr.getServiceId() ;
+ final InVMEntry entry ;
+ acquireReadLock() ;
+ try
+ {
+ entry = serviceIdToEntry.get(serviceId) ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ if (entry == null)
+ {
+ throw new InVMException("Could not locate service entry for epr " + inVMEpr) ;
+ }
+
+ final Object msgObject = entry.pickup(millis) ;
+ if (msgObject != null)
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Pickup of message from " + serviceId) ;
+ }
+ final Message message ;
+ try
+ {
+ if (msgObject instanceof byte[])
+ {
+ message = MessageSerializer.deserialize((byte[])msgObject) ;
+ }
+ else if (inVMEpr.getPassByValue())
+ {
+ // pass by reference but now expecting value.
+ message = ((Message)msgObject).copy() ;
+ }
+ else
+ {
+ message = (Message)msgObject ;
+ }
+ }
+ catch (final IOException ioe)
+ {
+ throw new InVMException("Failed to deserialise incoming message", ioe) ;
+ }
+
+ if (isTransactional())
+ {
+ /*
+ * Return the message, but don't remove it from the queue until the transaction
+ * commits. If the transaction rolls back then the message may not go back into the
+ * queue at the exact place it was originally: other messages may have been removed
+ * successfully by other threads. Plus, we would have to maintain a before and after
+ * image of the queue. This is more a compensation transaction.
+ */
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Pickup enlisting transactional resource for service " + serviceId) ;
+ }
+ final TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true) ;
+ try
+ {
+ txStrategy.enlistResource(new InVMXAResource(inVMEpr, msgObject, InVMXAResource.operation.REMOVE)) ;
+ }
+ catch (final TransactionStrategyException tse)
+ {
+ throw new InVMException("Unexpected error enlisting transaction resource", tse) ;
+ }
+ }
+ return message ;
+ }
+ return null ;
+ }
+
+ /**
+ * Deliver an object as a consequence of a transaction. This will either be a
+ * rollback, placing the object back on the source queue, or a commit delivering
+ * to a target queue.
+ *
+ * @param inVMEpr The EPR to receive the message.
+ * @param msgObject The object to deliver.
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public void deliverTx(final InVMEpr inVMEpr, final Object msgObject)
+ throws InVMException
+ {
+ final String serviceId = inVMEpr.getServiceId() ;
+ final InVMEntry entry ;
+ acquireReadLock() ;
+ try
+ {
+ entry = serviceIdToEntry.get(serviceId) ;
+ }
+ finally
+ {
+ releaseReadLock() ;
+ }
+ if (entry == null)
+ {
+ throw new InVMException("Could not locate service entry for epr " + inVMEpr) ;
+ }
+
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Transactional redelivery of message to " + serviceId) ;
+ }
+ entry.deliver(msgObject, 0) ;
+ }
+
+ /**
+ * Acquire a read lock for accessing the data.
+ */
+ private void acquireReadLock()
+ {
+ lock.readLock().lock() ;
+ }
+
+ /**
+ * Release a read lock for accessing the data.
+ */
+ private void releaseReadLock()
+ {
+ lock.readLock().unlock() ;
+ }
+
+ /**
+ * Acquire a write lock for accessing the data.
+ */
+ private void acquireWriteLock()
+ {
+ lock.writeLock().lock() ;
+ }
+
+ /**
+ * Release a write lock for accessing the data.
+ */
+ private void releaseWriteLock()
+ {
+ lock.writeLock().unlock() ;
+ }
+
+ /**
+ * TODO this is used in a number of classes so should be a separate
+ * util routine.
+ */
+ static boolean isTransactional()
+ throws InVMException
+ {
+ boolean transactional;
+ try
+ {
+ TransactionStrategy txStrategy = TransactionStrategy.getTransactionStrategy(true);
+ Object txHandle = ((txStrategy == null) ? null : txStrategy.getTransaction());
+ boolean isActive = ((txStrategy == null) ? false : txStrategy.isActive());
+
+ transactional = (txHandle != null);
+
+ /*
+ * Make sure the current transaction is still active! If we
+ * have previously slept, then the timeout may be longer than that
+ * associated with the transaction.
+ */
+
+ if (transactional && !isActive)
+ {
+ throw new InVMException("Associated transaction is no longer active!");
+ }
+ }
+ catch (final TransactionStrategyException ex)
+ {
+ throw new InVMException(ex);
+ }
+ return transactional;
+ }
+
+ /**
+ * Data representing an entry in the message queue.
+ * @author kevin
+ */
+ private static class InVMQueueEntry
+ {
+ /**
+ * The value enqueued.
+ */
+ private final Object value ;
+ /**
+ * Condition associated with lockstep.
+ */
+ private final Condition condition ;
+
+ /**
+ * Construct the InVM queue entry.
+ * @param value The value being enqueued.
+ * @param condition The condition representing the lockstep or null if not required.
+ */
+ InVMQueueEntry(final Object value, final Condition condition)
+ {
+ this.value = value ;
+ this.condition = condition ;
+ }
+
+ /**
+ * Get the enqueued value.
+ * @return The enqueued value.
+ */
+ Object getValue()
+ {
+ return value ;
+ }
+
+ /**
+ * Get the lockstep condition.
+ * @return The lockstep condition or null if not necessary.
+ */
+ Condition getCondition()
+ {
+ return condition ;
+ }
+ }
+
+ /**
+ * The InVM entries in the data structures.
+ * @author kevin
+ */
+ private static class InVMEntry
+ {
+ /**
+ * The lock for the queue.
+ */
+ private final Lock lock = new ReentrantLock() ;
+ /**
+ * The condition on which to await messages.
+ */
+ private final Condition waitingCondition = lock.newCondition() ;
+ /**
+ * The number of waiters on the waiting condition.
+ */
+ private int numWaiters ;
+ /**
+ * Flag indicating service has been shutdown.
+ */
+ private boolean shutdown ;
+ /**
+ * Entries in the queue.
+ */
+ private final Queue<InVMQueueEntry> entries = new LinkedList<InVMQueueEntry>() ;
+ /**
+ * Current EPRs.
+ */
+ private final List<InVMEpr> eprs = new LinkedList<InVMEpr>() ;
+ /**
+ * The service id for this entry.
+ */
+ private final String serviceId ;
+ /**
+ * The number of EPRs supporting pass by value.
+ */
+ private int numPassByValue ;
+ /**
+ * The maximum lockstep value.
+ */
+ private long lockstep ;
+
+ /**
+ * Create an entry with the specified service id.
+ * @param serviceId The service id.
+ */
+ InVMEntry(final String serviceId)
+ {
+ this.serviceId = serviceId ;
+ }
+
+ /**
+ * Get the service id associated with this entry.
+ * @return The service id.
+ */
+ String getServiceId()
+ {
+ return serviceId ;
+ }
+
+ /**
+ * Get the EPRs associated with this entry.
+ * Must hold external lock before accessing.
+ *
+ * @return the service EPRs.
+ */
+ List<InVMEpr> getEPRs()
+ {
+ return eprs ;
+ }
+
+ /**
+ * Add an EPR into this entry.
+ * Must hold external write lock before accessing.
+ *
+ * @param epr the EPR to register with this entry.
+ * @return true if this is the first epr, false otherwise.
+ */
+ boolean addEPR(final InVMEpr epr)
+ {
+ boolean result = eprs.isEmpty() ;
+ eprs.add(epr) ;
+ if (epr.getPassByValue())
+ {
+ numPassByValue ++ ;
+ }
+ if (epr.getLockstep())
+ {
+ final long eprLockstep = epr.getLockstepWaitTime() ;
+ if (eprLockstep > lockstep)
+ {
+ lockstep = eprLockstep ;
+ }
+ }
+ return result ;
+ }
+
+ /**
+ * Remove an EPR from this entry.
+ * Must hold external write lock before accessing.
+ *
+ * @param epr the EPR to remove from this entry.
+ * @return true if this is the last epr, false otherwise.
+ * @throws ServiceNotFoundException if the EPR cannot be located.
+ */
+ boolean removeEPR(final InVMEpr epr)
+ throws ServiceNotFoundException
+ {
+ if (!eprs.remove(epr))
+ {
+ throw new ServiceNotFoundException("Could not locate the EPR in the current service") ;
+ }
+ if (epr.getPassByValue())
+ {
+ numPassByValue-- ;
+ }
+ if (epr.getLockstep())
+ {
+ if (epr.getLockstepWaitTime() == lockstep)
+ {
+ lockstep = 0 ;
+ for(InVMEpr inVMEpr: eprs)
+ {
+ if (inVMEpr.getLockstep())
+ {
+ final long eprLockstep = inVMEpr.getLockstepWaitTime() ;
+ if (eprLockstep > lockstep)
+ {
+ lockstep = eprLockstep ;
+ }
+ }
+ }
+ }
+ }
+ return eprs.isEmpty() ;
+ }
+
+ /**
+ * Shutdown this entry, releasing any waiters.
+ * Can hold external write lock before accessing.
+ */
+ public void shutdown()
+ {
+ lock.lock() ;
+ try
+ {
+ shutdown = true ;
+ if (numWaiters > 0)
+ {
+ waitingCondition.signalAll() ;
+ }
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+
+ /**
+ * Should we pass the message by value?
+ * Must hold external lock before querying.
+ *
+ * @return true if should pass by value, false otherwise.
+ */
+ public boolean isPassByValue()
+ {
+ return numPassByValue > 0 ;
+ }
+
+ /**
+ * Get the lockstep value.
+ * Must hold external lock before querying.
+ *
+ * @return the lockstep value or 0 if not active.
+ */
+ public long getLockstep()
+ {
+ return lockstep ;
+ }
+
+ /**
+ * Deliver the specified value onto the queue.
+ *
+ * N.B. Must not hold external lock while calling this method.
+ *
+ * @param value The value being appended to the queue.
+ * @param lockstep The lockstep timeout or 0 if not required.
+ *
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public void deliver(final Object value, final long lockstep)
+ throws InVMException
+ {
+ lock.lock() ;
+ try
+ {
+ if (shutdown)
+ {
+ throw new InVMException("InVM Transport already shutdown") ;
+ }
+ final Condition condition = (lockstep > 0 ? lock.newCondition() : null) ;
+ final InVMQueueEntry queueEntry = new InVMQueueEntry(value, condition) ;
+ if (!entries.offer(queueEntry))
+ {
+ throw new InVMException("Failed to append message to InVM queue") ;
+ }
+ if (numWaiters > 0)
+ {
+ waitingCondition.signal() ;
+ }
+
+ if (condition != null)
+ {
+ try
+ {
+ condition.await(lockstep, TimeUnit.MILLISECONDS) ;
+ }
+ catch (final InterruptedException ie)
+ {
+ LOGGER.warn("Waiting delivery thread interupted while waiting on message pickup on InVM queue '" + serviceId + "'. Exiting pickup wait state.") ;
+ }
+ }
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+
+ /**
+ * Pickup an entry from the queue.
+ *
+ * N.B. Must not hold external lock while calling this method.
+ *
+ * @param millis The number of milliseconds to wait for a message.
+ * @return The message or null if nothing present within the timeout.
+ *
+ * @throws InVMException for InVM transport specific errors.
+ */
+ public Object pickup(final long millis)
+ throws InVMException
+ {
+ final long end = System.currentTimeMillis() + millis ;
+ lock.lock() ;
+ try
+ {
+ if (shutdown)
+ {
+ throw new InVMException("InVM Transport already shutdown") ;
+ }
+ if (entries.isEmpty())
+ {
+ final long delay = end - System.currentTimeMillis() ;
+ if (delay > 0)
+ {
+ numWaiters++ ;
+ try
+ {
+ waitingCondition.await(delay, TimeUnit.MILLISECONDS) ;
+ }
+ catch (final InterruptedException ioe)
+ {
+ throw new InVMException("Interrupted during wait") ;
+ }
+ }
+ }
+ final InVMQueueEntry entry = entries.poll() ;
+ if (entry != null)
+ {
+ final Object result = entry.getValue() ;
+ final Condition condition = entry.getCondition() ;
+ if (condition != null)
+ {
+ condition.signal() ;
+ }
+ return result ;
+ }
+ return null ;
+ }
+ finally
+ {
+ lock.unlock() ;
+ }
+ }
+ }
+}
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransport.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/tx/InVMXAResource.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/tx/InVMXAResource.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/couriers/tx/InVMXAResource.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -22,13 +22,16 @@
package org.jboss.internal.soa.esb.couriers.tx;
+import java.io.Serializable;
+
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import org.apache.log4j.Logger;
-import org.jboss.internal.soa.esb.couriers.InVMCourier;
-import org.jboss.soa.esb.message.Message;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTemporaryTransport;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTransport;
+import org.jboss.soa.esb.addressing.eprs.InVMEpr;
/**
* This XAResource instance controls the InVM queue manipulation under the control
@@ -49,14 +52,19 @@
* to make this an XAResource.
*/
-public class InVMXAResource implements XAResource
+public class InVMXAResource implements XAResource, Serializable
{
+ /**
+ * Serial version UID for this class.
+ */
+ private static final long serialVersionUID = 77430212548543969L;
+
public enum operation { INSERT, REMOVE };
- public InVMXAResource (InVMCourier courier, Message msg, final operation op)
+ public InVMXAResource (final InVMEpr inVMEpr, final Object msgObject, final operation op)
{
- _theCourier = courier;
- _msg = msg;
+ this.inVMEpr = inVMEpr ;
+ this.msgObject = msgObject ;
_opcode = op;
}
@@ -76,7 +84,7 @@
try
{
- problem = !_theCourier.doDeliver(_msg);
+ deliverTx() ;
}
catch (final Exception ex)
{
@@ -151,7 +159,7 @@
try
{
- problem = !_theCourier.doRedeliver(_msg);
+ deliverTx() ;
}
catch (final Exception ex)
{
@@ -169,6 +177,19 @@
}
}
+ private void deliverTx()
+ throws Exception
+ {
+ if (inVMEpr.isTemporaryEPR())
+ {
+ InVMTemporaryTransport.getInstance().deliverTx(inVMEpr, msgObject) ;
+ }
+ else
+ {
+ InVMTransport.getInstance().deliverTx(inVMEpr, msgObject) ;
+ }
+ }
+
public boolean setTransactionTimeout (int seconds) throws XAException
{
_timeout = seconds;
@@ -180,20 +201,15 @@
{
}
- public Message getMessage ()
- {
- return _msg;
- }
-
public boolean isSameRM (XAResource xares) throws XAException
{
return (xares == this);
}
- private InVMCourier _theCourier;
- private Message _msg;
- private operation _opcode;
- private int _timeout;
+ private transient InVMEpr inVMEpr;
+ private transient Object msgObject;
+ private transient operation _opcode;
+ private transient int _timeout;
protected static Logger _logger = Logger.getLogger(InVMXAResource.class);
}
\ No newline at end of file
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMRegistryInterceptor.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMRegistryInterceptor.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMRegistryInterceptor.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -19,17 +19,15 @@
*/
package org.jboss.internal.soa.esb.services.registry;
-import org.jboss.soa.esb.Service;
import org.jboss.soa.esb.addressing.EPR;
import org.jboss.soa.esb.addressing.eprs.InVMEpr;
-import org.jboss.soa.esb.couriers.CourierFactory;
import org.jboss.soa.esb.services.registry.AbstractRegistryInterceptor;
import org.jboss.soa.esb.services.registry.RegistryException;
import org.jboss.soa.esb.services.registry.ServiceNotFoundException;
-import org.jboss.internal.soa.esb.couriers.InVMCourier;
+import org.jboss.internal.soa.esb.couriers.transport.InVMException;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTransport;
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.List;
/**
* In VM registry.
@@ -38,44 +36,20 @@
*/
public class InVMRegistryInterceptor extends AbstractRegistryInterceptor {
- private Map<Service, List<RegsitryEntry>> regEntries = new ConcurrentHashMap<Service, List<RegsitryEntry>>();
-
public void registerEPR(String serviceCategoryName, String serviceName, String serviceDescription, EPR epr, String eprDescription) throws RegistryException {
if (epr instanceof InVMEpr) {
- Service service = new Service(serviceCategoryName, serviceName);
- RegsitryEntry regEntry = new RegsitryEntry(service, serviceDescription, epr, eprDescription);
- List<RegsitryEntry> serviceEntries = getServiceEntries(service);
-
- if(serviceEntries == null) {
- serviceEntries = new ArrayList<RegsitryEntry>();
- regEntries.put(service, serviceEntries);
+ try {
+ InVMTransport.getInstance().registerEPR(serviceCategoryName, serviceName, (InVMEpr)epr) ;
+ } catch (final InVMException invme) {
+ throw new RegistryException("Unexpected error registering service EPR", invme) ;
}
-
- if(serviceEntries.isEmpty()) {
- InVMCourier existingInVMCourier = CourierFactory.getInstance().getInVMCourier((InVMEpr) epr);
-
- if(existingInVMCourier != null) {
- throw new RegistryException("Invalid attempt to register InVM EPR '" + epr + "' for Service '" + service + "'. This InVM EPR is already registered for Service '" + existingInVMCourier.getService() + "'." );
- }
- InVMCourier newInVMCourier = CourierFactory.getInstance().addInVMCourier((InVMEpr) epr);
- newInVMCourier.setService(service);
- } else {
- // Verify that the InVM EPRs on the other registrations for this service
- // are exactly the same...
- EPR existingEpr = serviceEntries.get(0).getEpr();
- if(!epr.equals(existingEpr)) {
- throw new RegistryException("Invalid attempt to register different InVM EPRs for Service '" + service + "'. EPR '" + existingEpr + "' already registered. Attempt to register '" + epr + "'.");
- }
- }
-
- serviceEntries.add(regEntry);
} else {
getRegistry().registerEPR(serviceCategoryName, serviceName, serviceDescription, epr, eprDescription) ;
}
}
public void unRegisterService(String category, String serviceName) throws RegistryException, ServiceNotFoundException {
- final boolean entryRemoved = unRegisterInVMService(category, serviceName);
+ final boolean entryRemoved = InVMTransport.getInstance().unRegisterService(category, serviceName) ;
try {
getRegistry().unRegisterService(category, serviceName) ;
} catch (final ServiceNotFoundException snfe) {
@@ -85,77 +59,28 @@
}
}
- private boolean unRegisterInVMService(String category, String serviceName) throws RegistryException, ServiceNotFoundException {
- if (isValidService(category, serviceName)) {
- return _unRegisterEPR(category, serviceName, true);
- } else {
- return false ;
- }
- }
-
public void unRegisterEPR(String serviceCategoryName, String serviceName, EPR epr) throws RegistryException, ServiceNotFoundException {
if (epr instanceof InVMEpr) {
- _unRegisterEPR(serviceCategoryName, serviceName, false);
+ InVMTransport.getInstance().unRegisterEPR(serviceCategoryName, serviceName, (InVMEpr)epr);
} else {
getRegistry().unRegisterEPR(serviceCategoryName, serviceName, epr) ;
}
}
- private boolean _unRegisterEPR(String serviceCategoryName, String serviceName, boolean removeAll) {
- Service service = new Service(serviceCategoryName, serviceName);
- List<RegsitryEntry> serviceEntries = getServiceEntries(service);
-
- if(serviceEntries != null) {
- InVMEpr removedInVMEpr = null;
-
- if(!serviceEntries.isEmpty()) {
- // Just remove one (decrement the count)...
- removedInVMEpr = (InVMEpr) serviceEntries.remove(0).getEpr();
- }
-
- if(removeAll) {
- serviceEntries.clear();
- }
-
- // Zap the InVM courier if there are no more Services using it...
- if(removedInVMEpr != null && serviceEntries.isEmpty()) {
- CourierFactory.getInstance().removeInVMCourier(removedInVMEpr);
- }
-
- return (removedInVMEpr != null);
- }
-
- return false;
- }
-
public List<String> findAllServices() throws RegistryException {
- final List<String> services = findInVMServices(null);
+ final List<String> services = InVMTransport.getInstance().findAllServices();
services.addAll(getRegistry().findAllServices()) ;
return services ;
}
public List<String> findServices(String serviceCategoryName) throws RegistryException {
- final List<String> services = findInVMServices(serviceCategoryName);
+ final List<String> services = InVMTransport.getInstance().findServices(serviceCategoryName);
services.addAll(getRegistry().findServices(serviceCategoryName)) ;
return services ;
}
- private List<String> findInVMServices(String serviceCategoryName) throws RegistryException {
- Iterator<Service> iterator = regEntries.keySet().iterator();
- List<String> services = new ArrayList<String>();
-
- while (iterator.hasNext()) {
- Service service = iterator.next();
- if(serviceCategoryName == null || service.getCategory().equals(serviceCategoryName)) {
- services.add(service.getName());
- }
- }
-
- return services;
- }
-
public List<EPR> findEPRs(String serviceCategoryName, String serviceName) throws RegistryException, ServiceNotFoundException {
- List<EPR> eprs = findInVMEPRs(serviceCategoryName, serviceName) ;
+ List<EPR> eprs = InVMTransport.getInstance().findEPRs(serviceCategoryName, serviceName) ;
if (eprs.size() == 0) {
return getRegistry().findEPRs(serviceCategoryName, serviceName) ;
} else {
@@ -169,46 +94,12 @@
}
public EPR findEPR(String serviceCategoryName, String serviceName) throws RegistryException, ServiceNotFoundException {
- List<EPR> eprs = findInVMEPRs(serviceCategoryName, serviceName);
+ final EPR epr = InVMTransport.getInstance().findEPR(serviceCategoryName, serviceName);
- if(!eprs.isEmpty()) {
- return eprs.get(0);
+ if(epr != null) {
+ return epr;
} else {
return getRegistry().findEPR(serviceCategoryName, serviceName);
}
}
-
- private List<EPR> findInVMEPRs(String serviceCategoryName, String serviceName) throws RegistryException, ServiceNotFoundException {
- List<EPR> eprs = new ArrayList<EPR>();
- if (isValidService(serviceCategoryName, serviceName)) {
- Service service = new Service(serviceCategoryName, serviceName);
- List<RegsitryEntry> serviceEntries = getServiceEntries(service);
-
- if(serviceEntries != null) {
- for (RegsitryEntry serviceEntry : serviceEntries) {
- eprs.add(serviceEntry.getEpr());
- }
- }
- }
- return eprs ;
- }
-
- private synchronized List<RegsitryEntry> getServiceEntries(Service service) {
- List<RegsitryEntry> serviceEntries = null; // = regEntries.get(service);
- Set<Map.Entry<Service, List<RegsitryEntry>>> entries = regEntries.entrySet();
-
- for (Map.Entry<Service, List<RegsitryEntry>> entry : entries) {
- if(entry.getKey().equals(service)) {
- serviceEntries = entry.getValue();
- break;
- }
- }
-
- return serviceEntries;
- }
-
- private boolean isValidService(final String category, final String name) {
- return ((category != null) && (category.length() > 0) &&
- (name != null) && (name.length() > 0)) ;
- }
}
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMService.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMService.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMService.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,47 @@
+/*
+* JBoss, Home of Professional Open Source
+* Copyright 2006, 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.internal.soa.esb.services.registry;
+
+import org.jboss.internal.soa.esb.couriers.transport.InVMTemporaryTransport;
+import org.jboss.system.ServiceMBeanSupport;
+
+/**
+ * Service to control the InVM Temporary reaper thread.
+ *
+ * @author <a href="Kevin.Conner at jboss.com">Kevin Conner</a>
+ */
+public class InVMService extends ServiceMBeanSupport implements InVMServiceMBean
+{
+ @Override
+ protected void startService()
+ throws Exception
+ {
+ InVMTemporaryTransport.getInstance().startReaper() ;
+ }
+
+ @Override
+ protected void stopService()
+ throws Exception
+ {
+ InVMTemporaryTransport.getInstance().stopReaper() ;
+ }
+}
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMService.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMServiceMBean.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMServiceMBean.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMServiceMBean.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,31 @@
+/*
+* JBoss, Home of Professional Open Source
+* Copyright 2006, 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.internal.soa.esb.services.registry;
+
+import org.jboss.system.ServiceMBean;
+
+/**
+ * @author <a href="Kevin.Conner at jboss.com">Kevin Conner</a>
+ */
+public interface InVMServiceMBean extends ServiceMBean
+{
+}
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/services/registry/InVMServiceMBean.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/addressing/eprs/InVMEpr.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/addressing/eprs/InVMEpr.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/addressing/eprs/InVMEpr.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -77,15 +77,11 @@
public static final String PASS_BY_VALUE = "passByValue";
- private boolean _lockstep = false;
-
- private boolean _lockstepTime = false;
-
- private boolean _passByValue = false;
+ public static final String TEMPORARY_EPR = "temporaryEPR";
public InVMEpr(EPR epr)
{
- setAddr(epr.getAddr());
+ super(epr) ;
}
public InVMEpr(EPR epr, Element header)
@@ -109,17 +105,19 @@
if (tag.equals(LOCKSTEP_ENDPOINT_TAG))
{
getAddr().addExtension(LOCKSTEP_ENDPOINT_TAG, nl.item(i).getTextContent());
- _lockstep = true;
}
- if (tag.equals(LOCKSTEP_WAIT_TIME_TAG))
+ else if (tag.equals(LOCKSTEP_WAIT_TIME_TAG))
{
getAddr().addExtension(LOCKSTEP_WAIT_TIME_TAG, nl.item(i).getTextContent());
- _lockstepTime = true;
}
- if (tag.equals(PASS_BY_VALUE))
+ else if (tag.equals(PASS_BY_VALUE))
{
getAddr().addExtension(PASS_BY_VALUE, nl.item(i).getTextContent());
}
+ else if (tag.equals(TEMPORARY_EPR))
+ {
+ getAddr().addExtension(TEMPORARY_EPR, nl.item(i).getTextContent());
+ }
}
}
}
@@ -201,15 +199,7 @@
public void setLockstep(boolean lockstep)
{
- if (_lockstep)
- throw new IllegalStateException("Lockstep already set!");
-
- if (lockstep)
- getAddr().addExtension(LOCKSTEP_ENDPOINT_TAG, "true");
- else
- getAddr().addExtension(LOCKSTEP_ENDPOINT_TAG, "false");
-
- _lockstep = true;
+ getAddr().addExtension(LOCKSTEP_ENDPOINT_TAG, Boolean.toString(lockstep));
}
public long getLockstepWaitTime()
@@ -224,7 +214,7 @@
}
catch (Exception ex)
{
- ex.printStackTrace();
+ _logger.warn("Failed to parse lockstep wait time", ex);
}
}
@@ -233,12 +223,7 @@
public void setLockstepWaitTime(long waitTime)
{
- if (_lockstepTime)
- throw new IllegalStateException("Lockstep wait time already set!");
-
- getAddr().addExtension(LOCKSTEP_WAIT_TIME_TAG, "" + waitTime);
-
- _lockstepTime = true;
+ getAddr().addExtension(LOCKSTEP_WAIT_TIME_TAG, Long.toString(waitTime));
}
public boolean getPassByValue()
@@ -256,14 +241,21 @@
public void setPassByValue(boolean val)
{
- if (_passByValue)
- throw new IllegalStateException("Pass by value already set!");
-
- getAddr().addExtension(PASS_BY_VALUE, "" + val);
-
- _passByValue = true;
+ getAddr().addExtension(PASS_BY_VALUE, Boolean.toString(val));
}
+ public boolean isTemporaryEPR()
+ {
+ final String temporary = getAddr().getExtensionValue(TEMPORARY_EPR) ;
+
+ return Boolean.valueOf(temporary) ;
+ }
+
+ public void setTemporaryEPR(final boolean temporary)
+ {
+ getAddr().addExtension(TEMPORARY_EPR, Boolean.toString(temporary)) ;
+ }
+
public String toString()
{
return "InVMEpr [ " + super.getAddr().extendedToString() + " ]";
@@ -302,15 +294,6 @@
static
{
- try
- {
- _type = new URI("urn:jboss/esb/epr/type/invm");
- }
- catch (Exception ex)
- {
- ex.printStackTrace();
-
- throw new ExceptionInInitializerError(ex.toString());
- }
+ _type = URI.create("urn:jboss/esb/epr/type/invm");
}
}
\ No newline at end of file
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -34,7 +34,6 @@
import org.jboss.soa.esb.addressing.EPR;
import org.jboss.soa.esb.addressing.MalformedEPRException;
import org.jboss.soa.esb.addressing.PortReference;
-import org.jboss.soa.esb.addressing.eprs.InVMEpr;
import org.jboss.soa.esb.addressing.util.DefaultReplyTo;
import org.jboss.soa.esb.common.Configuration;
import org.jboss.soa.esb.common.Environment;
@@ -623,24 +622,12 @@
throw new MessageDeliverException("Caught unexpected throwable during send. Bailing-out!", t);
} finally {
- EPR replyToEpr = null;
-
- try {
- CourierUtil.cleanCourier(courier);
- // put back the old To since we will have changed it.
+ CourierUtil.cleanCourier(courier);
- replyToEpr = call.getReplyTo();
- call.setTo(currentToEpr);
- call.setReplyTo(currentReplyToEpr);
- } finally {
- // Need to do "special" cleanup for InVM replyTo couriers. Of course, this may
- // not be most "pure" to the courier architecture, but we can't use Courier.cleanup because it
- // is getting called multiple times between async delivery and pickup, which totally breaks
- // InVM delivery because it zaps the in-mem queue.
- if (replyToEpr instanceof InVMEpr) {
- CourierFactory.getInstance().removeInVMCourier((InVMEpr) replyToEpr);
- }
- }
+ // put back the old To since we will have changed it.
+
+ call.setTo(currentToEpr) ;
+ call.setReplyTo(currentReplyToEpr) ;
}
}
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/common/Environment.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/common/Environment.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/common/Environment.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -40,6 +40,7 @@
*/
public static final String DEFAULT_INVM_SCOPE = "jboss.esb.invm.scope.default";
+ public static final String INVM_EXPIRY_TIME = "org.jboss.soa.esb.invm.expiryTime";
public static final String SMTP_HOST = "org.jboss.soa.esb.mail.smtp.host";
public static final String SMTP_USERNAME = "org.jboss.soa.esb.mail.smtp.user";
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/couriers/CourierFactory.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/couriers/CourierFactory.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/couriers/CourierFactory.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -29,15 +29,12 @@
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.Map.Entry;
import org.apache.log4j.Logger;
import org.jboss.internal.soa.esb.couriers.TwoWayCourierImpl;
-import org.jboss.internal.soa.esb.couriers.InVMCourier;
import org.jboss.soa.esb.addressing.EPR;
import org.jboss.soa.esb.addressing.MalformedEPRException;
-import org.jboss.soa.esb.addressing.eprs.InVMEpr;
import org.jboss.soa.esb.lifecycle.LifecyclePriorities;
import org.jboss.soa.esb.lifecycle.LifecycleResource;
import org.jboss.soa.esb.lifecycle.LifecycleResourceException;
@@ -62,13 +59,6 @@
LifecyclePriorities.COURIER_PRIORITY) ;
/**
- * InVM Couiers.
- * <p/>
- * We handle this
- */
- private final Map<String, InVMCourier> inVMCouriers = new ConcurrentHashMap<String, InVMCourier>();
-
- /**
* Factory singleton instance.
*/
private static CourierFactory instance = new CourierFactory();
@@ -115,14 +105,8 @@
*/
public static Courier getCourier(EPR toEPR) throws CourierException, MalformedEPRException
{
- if(toEPR instanceof InVMEpr) {
- TwoWayCourierImpl courier = new TwoWayCourierImpl(CourierFactory.getInstance().getInVMCourier((InVMEpr) toEPR), null);
- courier.setToEpr(toEPR);
- return courier;
- } else {
- return getCourier(toEPR, null);
- }
- }
+ return getCourier(toEPR, null);
+ }
/**
* Obtain a courier which can perform pickups only and prime it with the "to address" (toEPR)
* and the "replyTo address" (replyToEPR).
@@ -133,40 +117,9 @@
*/
public static TwoWayCourier getPickupCourier(EPR replyToEPR) throws CourierException, MalformedEPRException
{
- if(replyToEPR instanceof InVMEpr) {
- TwoWayCourierImpl courier = new TwoWayCourierImpl(null, CourierFactory.getInstance().getInVMCourier((InVMEpr) replyToEPR));
- courier.setReplyToEpr(replyToEPR);
- return courier;
- } else {
- return getCourier(null, replyToEPR);
- }
- }
+ return getCourier(null, replyToEPR);
+ }
- public synchronized InVMCourier addInVMCourier(InVMEpr epr) {
- String address = epr.getAddr().getAddress();
- InVMCourier courier = new InVMCourier(epr);
-
- inVMCouriers.put(address, courier);
- courier.setActive(true);
-
- return courier;
- }
-
- public synchronized InVMCourier getInVMCourier(InVMEpr epr) {
- String address = epr.getAddr().getAddress();
-
- return inVMCouriers.get(address);
- }
-
- public synchronized void removeInVMCourier(InVMEpr epr) {
- String address = epr.getAddr().getAddress();
- InVMCourier courier = inVMCouriers.remove(address);
-
- if(courier != null) {
- courier.setActive(false);
- }
- }
-
/**
* Obtain a courier which can perform both deliveries and pickups and prime it with
* the "to address" (toEPR) and the "replyTo address" (replyToEPR).
Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/tests/InVMCourierUnitTest.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/tests/InVMCourierUnitTest.java 2009-08-04 14:52:02 UTC (rev 28793)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/tests/InVMCourierUnitTest.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -26,14 +26,12 @@
import junit.framework.JUnit4TestAdapter;
import junit.framework.TestCase;
import org.jboss.internal.soa.esb.couriers.InVMCourier;
-import org.jboss.internal.soa.esb.couriers.TwoWayCourierImpl;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTemporaryTransport;
+import org.jboss.internal.soa.esb.couriers.transport.InVMTransport;
import org.jboss.internal.soa.esb.services.registry.InVMRegistryInterceptor;
import org.jboss.internal.soa.esb.services.registry.MockRegistry;
-import org.jboss.soa.esb.addressing.MalformedEPRException;
import org.jboss.soa.esb.addressing.EPR;
import org.jboss.soa.esb.addressing.eprs.InVMEpr;
-import org.jboss.soa.esb.couriers.CourierException;
-import org.jboss.soa.esb.couriers.CourierFactory;
import org.jboss.soa.esb.couriers.CourierFactoryUtil;
import org.jboss.soa.esb.message.Message;
import org.jboss.soa.esb.message.format.MessageFactory;
@@ -42,7 +40,6 @@
import org.junit.Test;
import java.net.URI;
-import java.net.URISyntaxException;
public class InVMCourierUnitTest extends TestCase {
@@ -57,6 +54,300 @@
@Test
public void testUnthreadedDeliver() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid1"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ InVMCourier courier = new InVMCourier(epr);
+ Producer producer = new Producer(courier);
+ Consumer consumer = new Consumer(courier);
+
+ producer.run();
+ consumer.run();
+
+ Assert.assertEquals(true, consumer.valid());
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+ @Test
+ public void testThreadedDeliver() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid2"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ InVMCourier courier = new InVMCourier(epr);
+ Producer producer = new Producer(courier);
+ Consumer consumer = new Consumer(courier);
+
+ producer.start();
+ consumer.start();
+
+ consumer.join(TIMEOUT) ;
+ Assert.assertTrue("Consumer valid", consumer.valid()) ;
+
+ Assert.assertEquals(consumer.valid(), true);
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+ @Test
+ public void testDelayedThreadedDeliver() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid3"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ InVMCourier courier = new InVMCourier(epr);
+ Producer producer = new Producer(courier);
+ Consumer consumer = new Consumer(courier);
+
+ consumer.start();
+
+ try {
+ Thread.sleep(500);
+ }
+ catch (Exception ex) {
+ }
+
+ producer.start();
+
+ consumer.join(TIMEOUT) ;
+ Assert.assertTrue("Consumer valid", consumer.valid()) ;
+
+ Assert.assertEquals(consumer.valid(), true);
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+ @Test
+ public void testThreadedNullDeliver() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid4"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ InVMCourier courier = new InVMCourier(epr);
+ Consumer consumer = new Consumer(courier);
+
+ consumer.start();
+
+ try {
+ Thread.sleep(500);
+ }
+ catch (Exception ex) {
+ }
+
+ consumer.join(TIMEOUT) ;
+ Assert.assertFalse("Consumer terminated", consumer.isAlive()) ;
+ Assert.assertFalse("Consumer valid", consumer.valid()) ;
+
+ Assert.assertEquals(consumer.valid(), false);
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+ @Test
+ public void testLockstepDeliver() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid5?true#2000"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ InVMCourier courier = new InVMCourier(epr);
+ Producer producer = new Producer(courier);
+ Consumer consumer = new Consumer(courier);
+
+ consumer.start();
+ producer.start();
+
+ consumer.join(TIMEOUT) ;
+ Assert.assertTrue("Consumer valid", consumer.valid()) ;
+
+ Assert.assertEquals(consumer.valid(), true);
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+ @Test
+ public void testPassByValueDeliver() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid5/true?true#2000"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ InVMCourier courier = new InVMCourier(epr);
+ Producer producer = new Producer(courier);
+ Consumer consumer = new Consumer(courier);
+
+ consumer.start();
+ producer.start();
+
+ consumer.join(TIMEOUT) ;
+ Assert.assertTrue("Consumer valid", consumer.valid()) ;
+
+ Assert.assertEquals(consumer.valid(), true);
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+
+ @Test
+ public void testLockstepMultiProducerPerformance() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid5?true#2000"));
+ InVMTransport.getInstance().registerEPR("x", "y", epr) ;
+ try
+ {
+ int iters = 1000;
+ int numberOfProducers = 50;
+ Producer[] producer = new Producer[numberOfProducers];
+ InVMCourier courier = new InVMCourier(epr);
+ Consumer consumer = new Consumer(courier, iters * numberOfProducers,
+ false);
+ long stime = System.currentTimeMillis();
+
+ for (int i = 0; i < numberOfProducers; i++)
+ producer[i] = new Producer(courier, iters, false);
+
+ consumer.start();
+
+ for (int j = 0; j < numberOfProducers; j++)
+ producer[j].start();
+
+ consumer.join(TIMEOUT) ;
+ Assert.assertTrue("Consumer valid", consumer.valid()) ;
+
+ long ftime = System.currentTimeMillis();
+ double factor = 1000.00 / (ftime - stime);
+ double msgsPerSecond = iters * factor * numberOfProducers;
+
+ if (!consumer.valid())
+ System.err.println("Completed " + consumer.itersCompleted());
+
+ Assert.assertEquals(consumer.valid(), true);
+
+ System.err.println("Time for " + iters * numberOfProducers
+ + " messages is " + (ftime - stime) + " milliseconds.");
+ System.err.println("Messages per second: " + msgsPerSecond);
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterEPR("x", "y", epr) ;
+ }
+ }
+
+ @Test
+ public void test_JBESB_2108_01() throws ServiceNotFoundException, RegistryException {
+ InVMRegistryInterceptor reg = new InVMRegistryInterceptor();
+ try
+ {
+ reg.setRegistry(new MockRegistry()) ;
+
+ reg.registerEPR("x", "y", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
+
+ EPR eprFromReg = reg.findEPR("x", "y");
+ Assert.assertEquals("InVM EPR count", 1, InVMTransport.getInstance().findEPRs("x", "y").size()) ;
+
+ // Register the EPR again...
+ reg.registerEPR("x", "y", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
+ Assert.assertEquals("InVM EPR count", 2, InVMTransport.getInstance().findEPRs("x", "y").size()) ;
+
+ // Unregister it once...
+ reg.unRegisterEPR("x", "y", new InVMEpr(new EPR(URI.create("x://123"))));
+
+ // Should still be able to find it...
+ reg.findEPR("x", "y");
+ Assert.assertEquals("InVM EPR count", 1, InVMTransport.getInstance().findEPRs("x", "y").size()) ;
+
+ // Unregister it again...
+ reg.unRegisterEPR("x", "y", new InVMEpr(new EPR(URI.create("x://123"))));
+
+ Assert.assertEquals("InVM EPR count", 0, InVMTransport.getInstance().findEPRs("x", "y").size()) ;
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterService("x", "y") ;
+ }
+ }
+
+ @Test
+ public void test_JBESB_2108_02() throws ServiceNotFoundException, RegistryException {
+ // KEV
+ InVMRegistryInterceptor reg = new InVMRegistryInterceptor();
+ try
+ {
+ reg.setRegistry(new MockRegistry()) ;
+
+ reg.registerEPR("xy", "z", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
+ try {
+ reg.registerEPR("x", "yz", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
+ Assert.fail("Expected RegistryException");
+ } catch (RegistryException e) {} // expected
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterService("xy", "z") ;
+ InVMTransport.getInstance().unRegisterService("x", "yz") ;
+ }
+ }
+
+ @Test
+ public void test_JBESB_2108_03() throws ServiceNotFoundException, RegistryException {
+ InVMRegistryInterceptor reg = new InVMRegistryInterceptor();
+ try
+ {
+ EPR xy_z = new InVMEpr(new EPR(URI.create("x://123")));
+ EPR x_yz = new InVMEpr(new EPR(URI.create("x://1234")));
+ reg.setRegistry(new MockRegistry()) ;
+
+ // Register "xy:z"...
+ reg.registerEPR("xy", "z", "blah", xy_z, "blah");
+ Assert.assertEquals(xy_z, reg.findEPR("xy", "z"));
+ Assert.assertEquals("InVM EPR count", 1, InVMTransport.getInstance().findEPRs("xy", "z").size()) ;
+ Assert.assertEquals("InVM EPR count", 0, InVMTransport.getInstance().findEPRs("x", "yz").size()) ;
+
+ // Register "x:yz"...
+ reg.registerEPR("x", "yz", "blah", x_yz, "blah");
+ Assert.assertEquals(x_yz, reg.findEPR("x", "yz"));
+ Assert.assertEquals("InVM EPR count", 1, InVMTransport.getInstance().findEPRs("xy", "z").size()) ;
+ Assert.assertEquals("InVM EPR count", 1, InVMTransport.getInstance().findEPRs("x", "yz").size()) ;
+
+ // Unregister "xy:z"...
+ reg.unRegisterEPR("xy", "z", xy_z);
+ Assert.assertEquals("InVM EPR count", 0, InVMTransport.getInstance().findEPRs("xy", "z").size()) ;
+ Assert.assertEquals("InVM EPR count", 1, InVMTransport.getInstance().findEPRs("x", "yz").size()) ;
+
+ // Unregister "x:yz"...
+ reg.unRegisterEPR("x", "yz", x_yz);
+ Assert.assertEquals("InVM EPR count", 0, InVMTransport.getInstance().findEPRs("xy", "z").size()) ;
+ Assert.assertEquals("InVM EPR count", 0, InVMTransport.getInstance().findEPRs("x", "yz").size()) ;
+ }
+ finally
+ {
+ InVMTransport.getInstance().unRegisterService("x", "yz") ;
+ InVMTransport.getInstance().unRegisterService("xy", "z") ;
+ }
+ }
+
+ @Test
+ public void testTemporaryUnthreadedDeliver() throws Exception {
+ InVMEpr epr = new InVMEpr(new URI("invm://serviceid1"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
InVMCourier courier = new InVMCourier(epr);
Producer producer = new Producer(courier);
Consumer consumer = new Consumer(courier);
@@ -65,11 +356,16 @@
consumer.run();
Assert.assertEquals(true, consumer.valid());
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
}
@Test
- public void testThreadedDeliver() throws Exception {
+ public void testTemporaryThreadedDeliver() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid2"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
InVMCourier courier = new InVMCourier(epr);
Producer producer = new Producer(courier);
Consumer consumer = new Consumer(courier);
@@ -81,11 +377,16 @@
Assert.assertTrue("Consumer valid", consumer.valid()) ;
Assert.assertEquals(consumer.valid(), true);
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
}
@Test
- public void testDelayedThreadedDeliver() throws Exception {
+ public void testTemporaryDelayedThreadedDeliver() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid3"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
InVMCourier courier = new InVMCourier(epr);
Producer producer = new Producer(courier);
Consumer consumer = new Consumer(courier);
@@ -104,11 +405,17 @@
Assert.assertTrue("Consumer valid", consumer.valid()) ;
Assert.assertEquals(consumer.valid(), true);
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
}
@Test
- public void testThreadedNullDeliver() throws Exception {
+ public void testTemporaryThreadedNullDeliver() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid4"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
+
InVMCourier courier = new InVMCourier(epr);
Consumer consumer = new Consumer(courier);
@@ -125,11 +432,16 @@
Assert.assertFalse("Consumer valid", consumer.valid()) ;
Assert.assertEquals(consumer.valid(), false);
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
}
@Test
- public void testLockstepDeliver() throws Exception {
+ public void testTemporaryLockstepDeliver() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid5?true#2000"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
InVMCourier courier = new InVMCourier(epr);
Producer producer = new Producer(courier);
Consumer consumer = new Consumer(courier);
@@ -141,11 +453,16 @@
Assert.assertTrue("Consumer valid", consumer.valid()) ;
Assert.assertEquals(consumer.valid(), true);
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
}
@Test
- public void testPassByValueDeliver() throws Exception {
+ public void testTemporaryPassByValueDeliver() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid5/true?true#2000"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
InVMCourier courier = new InVMCourier(epr);
Producer producer = new Producer(courier);
Consumer consumer = new Consumer(courier);
@@ -157,27 +474,16 @@
Assert.assertTrue("Consumer valid", consumer.valid()) ;
Assert.assertEquals(consumer.valid(), true);
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
}
-
@Test
- public void test_single_Courier_Creation() throws URISyntaxException,
- MalformedEPRException, CourierException {
+ public void testTemporaryLockstepMultiProducerPerformance() throws Exception {
InVMEpr epr = new InVMEpr(new URI("invm://serviceid5?true#2000"));
- TwoWayCourierImpl courier1 = (TwoWayCourierImpl) CourierFactory
- .getInstance().getCourier(epr);
- TwoWayCourierImpl courier2 = (TwoWayCourierImpl) CourierFactory
- .getInstance().getCourier(epr);
-
- Assert.assertTrue(courier1.getDeliverCourier() == courier2
- .getDeliverCourier());
- Assert.assertTrue(courier1.getPickupCourier() == courier2
- .getPickupCourier());
- }
-
- @Test
- public void testLockstepMultiProducerPerformance() throws Exception {
- InVMEpr epr = new InVMEpr(new URI("invm://serviceid5?true#2000"));
+ epr.setTemporaryEPR(true) ;
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
int iters = 1000;
int numberOfProducers = 50;
Producer[] producer = new Producer[numberOfProducers];
@@ -205,128 +511,14 @@
System.err.println("Completed " + consumer.itersCompleted());
Assert.assertEquals(consumer.valid(), true);
+ Assert.assertEquals("Ordered entry size", 0, InVMTemporaryTransport.getInstance().getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, InVMTemporaryTransport.getInstance().getServiceIdToEntrySize()) ;
System.err.println("Time for " + iters * numberOfProducers
+ " messages is " + (ftime - stime) + " milliseconds.");
System.err.println("Messages per second: " + msgsPerSecond);
}
-
- @Test
- public void test_JBESB_2108_01() throws ServiceNotFoundException, RegistryException {
- InVMRegistryInterceptor reg = new InVMRegistryInterceptor();
- reg.setRegistry(new MockRegistry()) ;
-
- reg.registerEPR("x", "y", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
-
- EPR eprFromReg = reg.findEPR("x", "y");
- InVMCourier courier = CourierFactory.getInstance().getInVMCourier((InVMEpr) eprFromReg);
- Assert.assertNotNull(courier);
- Assert.assertTrue(courier.isActive());
-
- // Register the EPR again...
- reg.registerEPR("x", "y", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
-
- // Unregister it once...
- reg.unRegisterEPR("x", "y", new InVMEpr(new EPR(URI.create("x://123"))));
-
- // Should still be able to find it...
- reg.findEPR("x", "y");
- // Should still be able to get the courier and it should still be active...
- Assert.assertNotNull(CourierFactory.getInstance().getInVMCourier((InVMEpr) eprFromReg));
- Assert.assertTrue(courier.isActive());
-
- // Unregister it again...
- reg.unRegisterEPR("x", "y", new InVMEpr(new EPR(URI.create("x://123"))));
-
- // Should get an error trying to find it now...
- try {
- reg.findEPR("x", "y");
- Assert.fail("Expected ServiceNotFoundException.");
- } catch(ServiceNotFoundException e) {
- Assert.assertEquals("Registry entry [x][y] not found.", e.getMessage());
- }
-
- // Shouldn't be able to get the courier and it should be inactive...
- Assert.assertNull(CourierFactory.getInstance().getInVMCourier((InVMEpr) eprFromReg));
- Assert.assertFalse(courier.isActive());
- }
-
- @Test
- public void test_JBESB_2108_02() throws ServiceNotFoundException, RegistryException {
- InVMRegistryInterceptor reg = new InVMRegistryInterceptor();
- reg.setRegistry(new MockRegistry()) ;
-
- reg.registerEPR("xy", "z", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
- try {
- reg.registerEPR("x", "yz", "blah", new InVMEpr(new EPR(URI.create("x://123"))), "blah");
- Assert.fail("Expected RegistryException");
- } catch (RegistryException e) {
- Assert.assertEquals("Invalid attempt to register InVM EPR 'InVMEpr [ PortReference < <wsa:Address x://123/> > ]' for Service 'x:yz'. This InVM EPR is already registered for Service 'xy:z'.", e.getMessage());
- }
- }
-
- @Test
- public void test_JBESB_2108_03() throws ServiceNotFoundException, RegistryException {
- InVMRegistryInterceptor reg = new InVMRegistryInterceptor();
- EPR xy_z = new InVMEpr(new EPR(URI.create("x://123")));
- EPR x_yz = new InVMEpr(new EPR(URI.create("x://1234")));
- reg.setRegistry(new MockRegistry()) ;
-
- // Register "xy:z"...
- reg.registerEPR("xy", "z", "blah", xy_z, "blah");
- Assert.assertEquals(xy_z, reg.findEPR("xy", "z"));
- InVMCourier xy_z_courier = CourierFactory.getInstance().getInVMCourier((InVMEpr) xy_z);
- Assert.assertNotNull(xy_z_courier);
- Assert.assertTrue(xy_z_courier.isActive());
-
- // Register "x:yz"...
- reg.registerEPR("x", "yz", "blah", x_yz, "blah");
- Assert.assertEquals(x_yz, reg.findEPR("x", "yz"));
- InVMCourier x_yz_courier = CourierFactory.getInstance().getInVMCourier((InVMEpr) x_yz);
- Assert.assertNotNull(x_yz_courier);
- Assert.assertTrue(x_yz_courier.isActive());
-
- // Unregister "xy:z"...
- reg.unRegisterEPR("xy", "z", xy_z);
- // Shouldn't be able to get the courier for "xy_z" and should be inactive...
- Assert.assertNull(CourierFactory.getInstance().getInVMCourier((InVMEpr) xy_z));
- Assert.assertFalse(xy_z_courier.isActive());
- // Should still be able to get the courier for "x_yz" and should still be active...
- Assert.assertNotNull(CourierFactory.getInstance().getInVMCourier((InVMEpr) x_yz));
- Assert.assertTrue(x_yz_courier.isActive());
-
- // Unregister "x:yz"...
- reg.unRegisterEPR("x", "yz", x_yz);
- // Shouldn't be able to get the courier for "x_yz" and should be inactive...
- Assert.assertNull(CourierFactory.getInstance().getInVMCourier((InVMEpr) x_yz));
- Assert.assertFalse(x_yz_courier.isActive());
- }
-
- /*
- * test that WeakHashMap could be used to gc invmcouriers.
- *
- @Test
- public void testFactory () throws Exception
- {
- int numberOfProducersConsumers = 100;
-
- for (int i = 0; i < numberOfProducersConsumers; i++)
- {
- InVMEpr epr = new InVMEpr(new URI("invm://serviceid" + i
- + "test"));
-
- TwoWayCourierImpl courier = (TwoWayCourierImpl) InternalCourierFactory.getInstance().getCourier(epr);
-
- Assert.assertNotNull(courier.getDeliverCourier());
- Assert.assertNull(courier.getPickupCourier());
- }
-
- for (int j = 0; j < 10; j++)
- System.gc(); // not guaranteed to work first time.
-
- //Assert.assertEquals(CourierFactoryHelper.getInVMCourierInstanceCount(InternalCourierFactory.getInstance()), 0);
- }*/
-
+
public static final int ITERATIONS = 10;
public static final int TIMEOUT = 30000;
}
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransportUnitTest.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransportUnitTest.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransportUnitTest.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,188 @@
+package org.jboss.internal.soa.esb.couriers.transport;
+
+import java.net.URI;
+
+import org.jboss.soa.esb.addressing.eprs.InVMEpr;
+import org.jboss.soa.esb.common.Environment;
+import org.jboss.soa.esb.message.Message;
+import org.jboss.soa.esb.message.format.MessageFactory;
+import org.junit.Test;
+
+import junit.framework.Assert;
+import junit.framework.JUnit4TestAdapter;
+
+/**
+ * InVMTemporaryTransport unit tests.
+ *
+ * Most of the delivery functionality is covered by the InVM courier unit tests.
+ *
+ * @author <a href="mailto:kevin.conner at jboss.com">Kevin Conner</a>
+ */
+public class InVMTemporaryTransportUnitTest
+{
+ @Test
+ public void testExpiry()
+ throws Exception
+ {
+ final String currentExpiryTime = System.getProperty(Environment.INVM_EXPIRY_TIME) ;
+ try
+ {
+ final long expiryTime = 5000 ;
+ System.setProperty(Environment.INVM_EXPIRY_TIME, Long.toString(expiryTime)) ;
+ final InVMTemporaryTransport transport = new InVMTemporaryTransport() ;
+ transport.startReaper() ;
+ try
+ {
+ final InVMEpr epr = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr1")) ;
+ final Message message = MessageFactory.getInstance().getMessage() ;
+
+ Assert.assertEquals("Ordered entry size", 0, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, transport.getServiceIdToEntrySize()) ;
+
+ transport.deliver(epr, message) ;
+
+ Assert.assertEquals("Ordered entry size", 1, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 1, transport.getServiceIdToEntrySize()) ;
+
+ Thread.sleep(expiryTime*2) ;
+
+ Assert.assertEquals("Ordered entry size", 0, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, transport.getServiceIdToEntrySize()) ;
+ }
+ finally
+ {
+ transport.stopReaper() ;
+ }
+ }
+ finally
+ {
+ if (currentExpiryTime == null)
+ {
+ System.getProperties().remove(Environment.INVM_EXPIRY_TIME) ;
+ }
+ else
+ {
+ System.setProperty(Environment.INVM_EXPIRY_TIME, currentExpiryTime) ;
+ }
+ }
+ }
+
+ @Test
+ public void testMultipleExpiries()
+ throws Exception
+ {
+ final String currentExpiryTime = System.getProperty(Environment.INVM_EXPIRY_TIME) ;
+ try
+ {
+ final long expiryTime = 5000 ;
+ System.setProperty(Environment.INVM_EXPIRY_TIME, Long.toString(expiryTime)) ;
+ final InVMTemporaryTransport transport = new InVMTemporaryTransport() ;
+ transport.startReaper() ;
+ try
+ {
+ final InVMEpr epr = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr1")) ;
+ final Message message = MessageFactory.getInstance().getMessage() ;
+
+ Assert.assertEquals("Ordered entry size", 0, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, transport.getServiceIdToEntrySize()) ;
+
+ transport.deliver(epr, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr, message) ;
+
+ Assert.assertEquals("Ordered entry size", 5, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 1, transport.getServiceIdToEntrySize()) ;
+
+ Thread.sleep(expiryTime*2) ;
+
+ Assert.assertEquals("Ordered entry size", 0, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, transport.getServiceIdToEntrySize()) ;
+ }
+ finally
+ {
+ transport.stopReaper() ;
+ }
+ }
+ finally
+ {
+ if (currentExpiryTime == null)
+ {
+ System.getProperties().remove(Environment.INVM_EXPIRY_TIME) ;
+ }
+ else
+ {
+ System.setProperty(Environment.INVM_EXPIRY_TIME, currentExpiryTime) ;
+ }
+ }
+ }
+
+ @Test
+ public void testMultipleExpiriesMultipleEPRs()
+ throws Exception
+ {
+ final String currentExpiryTime = System.getProperty(Environment.INVM_EXPIRY_TIME) ;
+ try
+ {
+ final long expiryTime = 5000 ;
+ System.setProperty(Environment.INVM_EXPIRY_TIME, Long.toString(expiryTime)) ;
+ final InVMTemporaryTransport transport = new InVMTemporaryTransport() ;
+ transport.startReaper() ;
+ try
+ {
+ final InVMEpr epr1 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr1")) ;
+ final InVMEpr epr2 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr2")) ;
+ final InVMEpr epr3 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr3")) ;
+ final InVMEpr epr4 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr4")) ;
+ final InVMEpr epr5 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://epr5")) ;
+ final Message message = MessageFactory.getInstance().getMessage() ;
+
+ Assert.assertEquals("Ordered entry size", 0, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, transport.getServiceIdToEntrySize()) ;
+
+ transport.deliver(epr1, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr2, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr3, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr4, message) ;
+ Thread.sleep(100) ;
+ transport.deliver(epr5, message) ;
+
+ Assert.assertEquals("Ordered entry size", 5, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 5, transport.getServiceIdToEntrySize()) ;
+
+ Thread.sleep(expiryTime*2) ;
+
+ Assert.assertEquals("Ordered entry size", 0, transport.getOrderedEntriesSize()) ;
+ Assert.assertEquals("Service id size", 0, transport.getServiceIdToEntrySize()) ;
+ }
+ finally
+ {
+ transport.stopReaper() ;
+ }
+ }
+ finally
+ {
+ if (currentExpiryTime == null)
+ {
+ System.getProperties().remove(Environment.INVM_EXPIRY_TIME) ;
+ }
+ else
+ {
+ System.setProperty(Environment.INVM_EXPIRY_TIME, currentExpiryTime) ;
+ }
+ }
+ }
+
+ public static junit.framework.Test suite()
+ {
+ return new JUnit4TestAdapter(InVMTemporaryTransportUnitTest.class);
+ }
+}
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTemporaryTransportUnitTest.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
Added: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransportUnitTest.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransportUnitTest.java (rev 0)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransportUnitTest.java 2009-08-04 15:28:34 UTC (rev 28794)
@@ -0,0 +1,114 @@
+package org.jboss.internal.soa.esb.couriers.transport;
+
+import java.net.URI;
+
+import org.jboss.soa.esb.addressing.eprs.InVMEpr;
+import org.jboss.soa.esb.services.registry.ServiceNotFoundException;
+import org.junit.Test;
+
+import junit.framework.Assert;
+import junit.framework.JUnit4TestAdapter;
+
+/**
+ * InVMTransport unit tests.
+ *
+ * Most of the delivery functionality is covered by the InVM courier unit tests.
+ *
+ * @author <a href="mailto:kevin.conner at jboss.com">Kevin Conner</a>
+ */
+public class InVMTransportUnitTest
+{
+ private static final String TEST_CATEGORY1 = "test category1" ;
+ private static final String TEST_CATEGORY2 = "test category2" ;
+ private static final String TEST_NAME1 = "service name 1" ;
+ private static final String TEST_NAME2 = "service name 2" ;
+
+ @Test
+ public void testRegistration()
+ throws Exception
+ {
+ final InVMTransport transport = new InVMTransport() ;
+ final InVMEpr epr1 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://" + InVMEpr.createEncodedServiceId(TEST_CATEGORY1, TEST_NAME1))) ;
+ final InVMEpr epr2 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://" + InVMEpr.createEncodedServiceId(TEST_CATEGORY2, TEST_NAME2))) ;
+ final InVMEpr epr3 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://" + InVMEpr.createEncodedServiceId(TEST_CATEGORY2, TEST_NAME2))) ;
+ final InVMEpr epr4 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://" + InVMEpr.createEncodedServiceId(TEST_CATEGORY2, TEST_NAME2))) ;
+
+ Assert.assertEquals("All services count", 0, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY2).size()) ;
+
+ transport.registerEPR(TEST_CATEGORY1, TEST_NAME1, epr1) ;
+
+ Assert.assertEquals("All services count", 1, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 1, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY2).size()) ;
+
+ transport.registerEPR(TEST_CATEGORY2, TEST_NAME2, epr2) ;
+
+ Assert.assertEquals("All services count", 2, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 1, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 1, transport.findServices(TEST_CATEGORY2).size()) ;
+
+ transport.unRegisterEPR(TEST_CATEGORY1, TEST_NAME1, epr1) ;
+
+ Assert.assertEquals("All services count", 1, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 1, transport.findServices(TEST_CATEGORY2).size()) ;
+
+ transport.registerEPR(TEST_CATEGORY2, TEST_NAME2, epr3) ;
+ transport.registerEPR(TEST_CATEGORY2, TEST_NAME2, epr4) ;
+
+ Assert.assertEquals("All services count", 1, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 1, transport.findServices(TEST_CATEGORY2).size()) ;
+ Assert.assertEquals("EPR count", 3, transport.findEPRs(TEST_CATEGORY2, TEST_NAME2).size()) ;
+
+ transport.unRegisterEPR(TEST_CATEGORY2, TEST_NAME2, epr2) ;
+
+ Assert.assertEquals("All services count", 1, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 1, transport.findServices(TEST_CATEGORY2).size()) ;
+ Assert.assertEquals("EPR count", 2, transport.findEPRs(TEST_CATEGORY2, TEST_NAME2).size()) ;
+
+ transport.unRegisterService(TEST_CATEGORY2, TEST_NAME2) ;
+
+ Assert.assertEquals("All services count", 0, transport.findAllServices().size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY1).size()) ;
+ Assert.assertEquals("Category count", 0, transport.findServices(TEST_CATEGORY2).size()) ;
+ Assert.assertEquals("EPR count", 0, transport.findEPRs(TEST_CATEGORY2, TEST_NAME2).size()) ;
+ }
+
+ @Test
+ public void testMissingServices()
+ throws Exception
+ {
+ final InVMTransport transport = new InVMTransport() ;
+ final InVMEpr epr1 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://" + InVMEpr.createEncodedServiceId(TEST_CATEGORY1, TEST_NAME1))) ;
+ final InVMEpr epr2 = new InVMEpr(URI.create(InVMEpr.INVM_PROTOCOL + "://" + InVMEpr.createEncodedServiceId(TEST_CATEGORY1, TEST_NAME2))) ;
+
+ try
+ {
+ transport.unRegisterEPR(TEST_CATEGORY1, TEST_NAME1, epr1) ;
+ Assert.fail("Expected ServiceNotFoundException") ;
+ }
+ catch (final ServiceNotFoundException snfe) {} // expected
+
+ transport.unRegisterService(TEST_CATEGORY1, TEST_NAME1) ;
+
+ transport.registerEPR(TEST_CATEGORY1, TEST_NAME2, epr2) ;
+
+ try
+ {
+ transport.unRegisterEPR(TEST_CATEGORY1, TEST_NAME1, epr1) ;
+ Assert.fail("Expected ServiceNotFoundException") ;
+ }
+ catch (final ServiceNotFoundException snfe) {} // expected
+
+ transport.unRegisterService(TEST_CATEGORY1, TEST_NAME1) ;
+ }
+
+ public static junit.framework.Test suite()
+ {
+ return new JUnit4TestAdapter(InVMTransportUnitTest.class);
+ }
+}
Property changes on: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/tests/src/org/jboss/internal/soa/esb/couriers/transport/InVMTransportUnitTest.java
___________________________________________________________________
Name: svn:keywords
+ Rev Date
Name: svn:eol-style
+ native
More information about the jboss-svn-commits
mailing list