[jboss-svn-commits] JBL Code SVN: r17100 - labs/jbossesb/workspace/bramley/product/rosetta/src/org/jboss/internal/soa/esb/couriers.

[jboss-svn-commits at lists.jboss.org]

Author: mark.little at jboss.com
Date: 2007-12-07 12:38:56 -0500 (Fri, 07 Dec 2007)
New Revision: 17100

Modified:
   labs/jbossesb/workspace/bramley/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java
Log:
added back receiverOnly and comment on lock-step.

Modified: labs/jbossesb/workspace/bramley/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java
===================================================================
--- labs/jbossesb/workspace/bramley/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java	2007-12-07 17:26:49 UTC (rev 17099)
+++ labs/jbossesb/workspace/bramley/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java	2007-12-07 17:38:56 UTC (rev 17100)
@@ -32,91 +32,143 @@
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
-public class InVMCourier implements PickUpOnlyCourier, DeliverOnlyCourier {
+public class InVMCourier implements PickUpOnlyCourier, DeliverOnlyCourier
+{
 
-    /**
-     * package protected constructor - Objects of this class should only be
-     * instantiated by internal implementations
-     *
-     * @param epr
-     */
-    public InVMCourier(InVMEpr epr) throws CourierException,
-            MalformedEPRException
-    {
-        this.epr = epr;
-        if (epr.getLockstep()) {
-            deliveryTimeout = epr.getLockstepWaitTime();
-        } else {
-            deliveryTimeout = 0;
-        }
-    }
+	/**
+	 * Objects of this class should only be instantiated by internal
+	 * implementations
+	 * 
+	 * @param epr
+	 */
 
-    /**
-     * package the ESB message into the queue
-     *
-     * @param message Message - the message to deliverAsync
-     * @return boolean - the result of the delivery
-     * @throws CourierException -
-     *                          if problems were encountered
-     */
-    public boolean deliver(Message message) throws CourierException, MalformedEPRException
-    {
-        if (message == null) {
-            return false;
-        }
+	public InVMCourier(InVMEpr epr) throws CourierException,
+			MalformedEPRException
+	{
+		this(epr, false);
+	}
 
-        synchronized (messageQueue) {
-            messageQueue.add(message);
+	public InVMCourier(InVMEpr epr, boolean receiverOnly) throws CourierException,
+			MalformedEPRException
+	{
+		this.epr = epr;
+		
+		_receiverOnly = receiverOnly;
 
-            // Notify 1 waiting pickup thread of the delivery...
-            messageQueue.notify();
+		/*
+		 * 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".
+		 */
 
-            if (deliveryTimeout > 0) {
-                try {
-                    // Wait on notification from the pickup thread...
-                    messageQueue.wait(deliveryTimeout);
-                } catch (InterruptedException e) {
-                    _logger.warn("Timeout expired while waiting on message pickup on InVM queue '" + epr.getAddr().getAddress() + "'.", e);
-                }
-            }
-        }
+		if (epr.getLockstep())
+		{
+			deliveryTimeout = epr.getLockstepWaitTime();
+		}
+		else
+		{
+			deliveryTimeout = 0;
+		}
+	}
 
-        return true;
-    }
+	/**
+	 * package the ESB message into the queue
+	 * 
+	 * @param message
+	 *            Message - the message to deliverAsync
+	 * @return boolean - the result of the delivery
+	 * @throws CourierException -
+	 *             if problems were encountered
+	 */
+	public boolean deliver(Message message) throws CourierException,
+			MalformedEPRException
+	{
+		if (_receiverOnly)
+			throw new CourierException("This is a pickUp-only Courier");
+		
+		if (message == null)
+		{
+			return false;
+		}
 
-    public Message pickup(long millis) throws CourierException, CourierTimeoutException
-    {
-        Message message = null;
+		synchronized (messageQueue)
+		{
+			messageQueue.add(message);
 
-        millis = Math.max(millis, 100);
-        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()) {
-                message = messageQueue.remove();
-            }
+			// Notify 1 waiting pickup thread of the delivery...
+			messageQueue.notify();
 
-            // Notify 1 waiting delivery thread of the pickup...
-            messageQueue.notify();
-        }
+			if (deliveryTimeout > 0)
+			{
+				try
+				{
+					// Wait on notification from the pickup thread...
+					messageQueue.wait(deliveryTimeout);
+				}
+				catch (InterruptedException e)
+				{
+					_logger.warn(
+							"Timeout expired while waiting on message pickup on InVM queue '"
+									+ epr.getAddr().getAddress() + "'.", e);
+				}
+			}
+		}
 
-        return message;
-    }
+		return true;
+	}
 
-    public void cleanup()
-    {
-    }
+	public Message pickup(long millis) throws CourierException,
+			CourierTimeoutException
+	{
+		Message message = null;
 
-    protected InVMEpr epr;
+		millis = Math.max(millis, 100); // TODO magic number
+		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())
+			{
+				message = messageQueue.remove();
+			}
 
-    protected static Logger _logger = Logger.getLogger(InVMCourier.class);
+			// Notify 1 waiting delivery thread of the pickup...
+			messageQueue.notify();
+		}
 
-    private Queue<Message> messageQueue = new ConcurrentLinkedQueue<Message>();
+		return message;
+	}
 
-    private long deliveryTimeout = 0;
+	public void cleanup()
+	{
+	}
+
+	protected InVMEpr epr;
+
+	protected static Logger _logger = Logger.getLogger(InVMCourier.class);
+
+	private Queue<Message> messageQueue = new ConcurrentLinkedQueue<Message>();
+
+	private long deliveryTimeout = 0;
+
+	private boolean _receiverOnly;
 }