svn commit: r594151 - /incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp

[cctrieloff]

Author: cctrieloff
Date: Mon Nov 12 06:02:14 2007
New Revision: 594151

URL: http://svn.apache.org/viewvc?rev=594151&view=rev
Log:

-- added sync request
-- make queue durable if -- durable yes specified


Modified:
    incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp

Modified: incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp?rev=594151&r1=594150&r2=594151&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp Mon Nov 12 06:02:14 
2007
@@ -24,6 +24,7 @@
 #include "qpid/client/Session_0_10.h"
 #include "qpid/client/SubscriptionManager.h"
 #include "qpid/client/Connection.h"
+#include "qpid/client/Completion.h"
 #include "qpid/client/Message.h"
 #include "qpid/sys/Time.h"
 
@@ -173,15 +174,22 @@
             sprintf(msgBuf, "%d", i);
             session.messageTransfer(arg::destination=exchange(),
                                     arg::content=msg);
-            if (!opts.summary && (i%10000)==0) cout << "." << flush;
+            if (!opts.summary && (i%10000)==0){
+                 cout << "." << flush;
+                 session.execution().sendSyncRequest();
+            }
         }
+        session.execution().sendSyncRequest();
+        
+        //Completion compl;
         if (!opts.summary) cout << " done." << endl;
         msg.setData("done");    // Send done messages.
         if (mode==SHARED)
             for (int i = 0; i < opts.consumers; ++i)
-                session.messageTransfer(arg::destination=exchange(), 
arg::content=msg);
+                 session.messageTransfer(arg::destination=exchange(), 
arg::content=msg);
         else
             session.messageTransfer(arg::destination=exchange(), 
arg::content=msg);
+        session.execution().sendSyncRequest();
         AbsTime end=now();
 
         // Report
@@ -191,6 +199,16 @@
                  << "publish count:" << opts.count << endl
                  << "publish secs:" << secs(start,end) << endl
                  << "publish rate:" << publish_rate << endl;
+        
+        
+
+        // Report
+//        end=now(); //compl.wait();  (wait for publish confirm of write if 
durable)
+//        publish_rate=(opts.count)/secs(start,end);
+//        if (!opts.summary)
+//            cout << endl
+//                 << "synced secs:" << secs(start,end) << endl
+//                 << "synced rate:" << publish_rate << endl;
 
         double consume_rate = 0; // Average rate for consumers.
         //  Wait for consumer(s) to finish.
@@ -247,7 +265,8 @@
             consumeQueue=session.getId().str(); // Unique name.
             session.queueDeclare(arg::queue=consumeQueue,
                                  arg::exclusive=true,
-                                 arg::autoDelete=true);
+                                 arg::autoDelete=true,
+                                 arg::durable=opts.durable);
             session.queueBind(arg::queue=consumeQueue,
                               arg::exchange=exchange(),
                               arg::routingKey="perftest");
@@ -263,8 +282,9 @@
         Message msg;
         while ((msg=consume.pop()).getData() != "done") {
             ++consumed;
+
         }
-        msg.acknowledge();      // Ack all outstanding messages.
+        msg.acknowledge();      // Ack all outstanding messages -- ??
         AbsTime end=now();
 
         // Report to publisher.