/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.bufferserver.policy;

import com.malhartech.bufferserver.PhysicalNode;
import com.malhartech.bufferserver.util.SerializedData;
import java.util.Set;

/**
 *
 * Distributes to downstream nodes in a roundrobin fashion. A round robin load balancing policy<p>
 * <br>
 * A round robin load balaning policy. Does not take into account busy/load of a downstream physical node. Extends the base class {@link com.malhartech.bufferserver.policy.AbstractPolicy}<br>
 * <br>
 * @author chetan
 */
public class RoundRobin extends AbstractPolicy
{

  int index;

  /**
   * Constructor
   */
  public RoundRobin()
  {
    index = 0;
  }

  /**
   * 
   * @param nodes Set of downstream {@link com.malhartech.bufferserver.PhysicalNode}s
   * @param data Opaque {@link com.malhartech.bufferserver.util.SerializedData} to be send 
   */
  @Override
  public void distribute(Set<PhysicalNode> nodes, SerializedData data)
  {
    index %= nodes.size();
    int count = index++;
    /*
     * May need to look at accessing nodes as arrays, so that iteration can be avoided
     * This matters if say there are 1000+ partitions(?) and this may happen in a Big Message
     * application
     * 
     */
    for (PhysicalNode node : nodes) {
      if (count-- == 0) {
        node.send(data);
        break;
      }
    }
  }
}

@@ -548,16 +548,19 @@
  * %3Cbr%3E%0A
+ *%0A
  * @auth
@@ -569,16 +569,16 @@
  chetan%0A
-
  */%0Apubl
@@ -622,17 +622,16 @@
 olicy%0A%7B%0A
-%0A
   int in
@@ -717,25 +717,24 @@
 %0A%0A  /**%0A   *
- 
 %0A   * @param
@@ -898,17 +898,16 @@
  be send
- 
 %0A   */%0A 
@@ -1002,13 +1002,15 @@
   in
-dex %25
+t size 
 = no
@@ -1021,16 +1021,59 @@
 size();%0A
+    if (size %3E 0) %7B%0A      index %25= size;%0A  
     int 
@@ -1089,16 +1089,18 @@
 ndex++;%0A
+  
     /*%0A 
@@ -1103,16 +1103,18 @@
 /*%0A     
+  
 * May ne
@@ -1187,16 +1187,18 @@
 avoided%0A
+  
      * T
@@ -1288,16 +1288,18 @@
 ge%0A     
+ 
+ 
 * applic
@@ -1313,19 +1313,22 @@
     
-* %0A
+  *%0A  
      */%0A
     
@@ -1323,16 +1323,18 @@
      */%0A
+  
     for 
@@ -1351,17 +1351,16 @@
 ode node
- 
 : nodes)
@@ -1364,24 +1364,26 @@
 es) %7B%0A      
+ 
+ 
 if (count-- 
@@ -1398,16 +1398,18 @@
         
+  
 node.sen
@@ -1417,16 +1417,18 @@
 (data);%0A
+  
         
@@ -1434,16 +1434,26 @@
  break;%0A
+        %7D%0A
       %7D%0A
@@ -1458,14 +1458,15 @@
 %7D%0A    %7D%0A
-
   %7D%0A
+%0A
 %7D%0A

fix the divide by zero error.