Allow fractional message rates (useful for testing massive fanout).

Author: Simon MacMullen

test/src/com/rabbitmq/examples/MulticastMain.java

@@ -49,7 +49,7 @@ public static void main(String[] args) {
             String exchangeName  = strArg(cmd, 'e', exchangeType);
             String queueName     = strArg(cmd, 'u', "");
             int samplingInterval = intArg(cmd, 'i', 1);
-            int rateLimit        = intArg(cmd, 'r', 0);
+            float rateLimit      = floatArg(cmd, 'r', 0.0f);
             int producerCount    = intArg(cmd, 'x', 1);
             int consumerCount    = intArg(cmd, 'y', 1);
             int producerTxSize   = intArg(cmd, 'm', 0);
@@ -161,6 +161,10 @@ private static int intArg(CommandLine cmd, char opt, int def) {
         return Integer.parseInt(cmd.getOptionValue(opt, Integer.toString(def)));
     }
 
+    private static float floatArg(CommandLine cmd, char opt, float def) {
+        return Float.parseFloat(cmd.getOptionValue(opt, Float.toString(def)));
+    }
+
     private static List<?> lstArg(CommandLine cmd, char opt) {
         String[] vals = cmd.getOptionValues('f');
         if (vals == null) {

test/src/com/rabbitmq/examples/perf/MulticastParams.java

@@ -32,7 +32,7 @@ public class MulticastParams {
     private int minMsgSize = 0;
 
     private int timeLimit = 0;
-    private int rateLimit = 0;
+    private float rateLimit = 0;
     private int producerMsgCount = 0;
     private int consumerMsgCount = 0;
 
@@ -59,7 +59,7 @@ public void setQueueName(String queueName) {
         this.queueName = queueName;
     }
 
-    public void setRateLimit(int rateLimit) {
+    public void setRateLimit(float rateLimit) {
         this.rateLimit = rateLimit;
     }
 

test/src/com/rabbitmq/examples/perf/Producer.java

@@ -41,7 +41,7 @@ public class Producer implements Runnable, ReturnListener,
     private boolean immediate;
     private boolean persistent;
     private int     txSize;
-    private int     rateLimit;
+    private float   rateLimit;
     private int     msgLimit;
     private long    timeLimit;
 
@@ -59,7 +59,7 @@ public class Producer implements Runnable, ReturnListener,
 
     public Producer(Channel channel, String exchangeName, String id,
                     List<?> flags, int txSize,
-                    int rateLimit, int msgLimit, int minMsgSize, int timeLimit,
+                    float rateLimit, int msgLimit, int minMsgSize, int timeLimit,
                     long confirm, Stats stats)
         throws IOException {
 
@@ -172,8 +172,8 @@ private void delay(long now)
         //10 ms have elapsed, we have sent 200 messages
         //the 200 msgs we have actually sent should have taken us
         //200 * 1000 / 5000 = 40 ms. So we pause for 40ms - 10ms
-        long pause = rateLimit == 0 ?
-            0 : (msgCount * 1000L / rateLimit - elapsed);
+        long pause = (long) (rateLimit == 0.0f ?
+            0.0f : (msgCount * 1000.0 / rateLimit - elapsed));
         if (pause > 0) {
             Thread.sleep(pause);
         }