[hornetq-commits] JBoss hornetq SVN: r9636 - branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example.

[do-not-reply at jboss.org]

Author: clebert.suconic at jboss.com
Date: 2010-09-02 23:15:25 -0400 (Thu, 02 Sep 2010)
New Revision: 9636

Modified:
   branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Receiver.java
   branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Sender.java
   branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/TXRestartSoak.java
Log:
tweak on soak test

Modified: branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Receiver.java
===================================================================
--- branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Receiver.java	2010-09-03 02:07:52 UTC (rev 9635)
+++ branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Receiver.java	2010-09-03 03:15:25 UTC (rev 9636)
@@ -13,6 +13,9 @@
 
 package org.hornetq.jms.example;
 
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.Queue;
@@ -33,6 +36,8 @@
    
    private Queue queue;
    
+   private final Semaphore sem = new Semaphore(0);
+   
    private final String queueJNDI;
    
    protected volatile long msgs = 0;
@@ -66,6 +71,10 @@
             
             for (int i = 0 ; i < 1000; i++)
             {
+               if (!sem.tryAcquire(1, 5, TimeUnit.SECONDS))
+               {
+                  break;
+               }
                Message msg = cons.receive(5000);
                if (msg == null)
                {
@@ -133,6 +142,14 @@
       return "Receiver::" + this.queueJNDI + ", msgs=" + msgs + ", pending=" + pendingMsgs;
    }
 
+   /**
+    * @param pendingMsgs2
+    */
+   public void messageProduced(int pendingMsgs2)
+   {
+      sem.release(pendingMsgs2);
+   }
+
    // Package protected ---------------------------------------------
 
    // Protected -----------------------------------------------------

Modified: branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Sender.java
===================================================================
--- branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Sender.java	2010-09-03 02:07:52 UTC (rev 9635)
+++ branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/Sender.java	2010-09-03 03:15:25 UTC (rev 9636)
@@ -37,17 +37,20 @@
    protected Queue queue;
    
    protected long msgs = TXRestartSoak.MIN_MESSAGES_ON_QUEUE;
-   protected long pendingMsgs = 0;
+   protected int pendingMsgs = 0;
    
+   protected final Receiver[] receivers;
    
+   
    // Static --------------------------------------------------------
 
    // Constructors --------------------------------------------------
 
    // Public --------------------------------------------------------
    
-   public Sender()
+   public Sender(final Receiver[] receivers)
    {
+      this.receivers = receivers;
    }
 
    @Override
@@ -90,6 +93,11 @@
    protected void onCommit()
    {
       this.msgs += pendingMsgs;
+      for (Receiver rec : receivers)
+      {
+         rec.messageProduced(pendingMsgs);
+      }
+      
       pendingMsgs = 0;
       System.out.println("commit on sender msgs = " + msgs );
    }

Modified: branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/TXRestartSoak.java
===================================================================
--- branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/TXRestartSoak.java	2010-09-03 02:07:52 UTC (rev 9635)
+++ branches/Branch_2_1/examples/soak/tx-restarts/src/org/hornetq/jms/example/TXRestartSoak.java	2010-09-03 03:15:25 UTC (rev 9636)
@@ -104,13 +104,13 @@
          
          session.commit();
          
-         Sender send = new Sender();
+         Receiver rec1 = new Receiver("/queue/diverted1");
+         Receiver rec2 = new Receiver("/queue/diverted2");
          
+         Sender send = new Sender(new Receiver[]{rec1, rec2});
+         
          send.start();
-         
-         Receiver rec1 = new Receiver("/queue/diverted1");
          rec1.start();
-         Receiver rec2 = new Receiver("/queue/diverted2");
          rec2.start();