WeightedRoundRobinMultiplexer implements RpcMultiplexer { public int getAndAdvanceCurrentIndex() { int idx = this.getCurrentIndex(); this.advanceIndex(); return idx; } WeightedRoundRobinMultiplexer(int aNumQueues, String ns,
    Configuration conf); int getAndAdvanceCurrentIndex(); static final String IPC_CALLQUEUE_WRRMUX_WEIGHTS_KEY; static final Logger LOG; }
@Test public void testDefaultPattern() { mux = new WeightedRoundRobinMultiplexer(1, "", new Configuration()); for(int i = 0; i < 10; i++) { assertEquals(mux.getAndAdvanceCurrentIndex(), 0); } mux = new WeightedRoundRobinMultiplexer(2, "", new Configuration()); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); mux = new WeightedRoundRobinMultiplexer(3, "", new Configuration()); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 2); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); mux = new WeightedRoundRobinMultiplexer(4, "", new Configuration()); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 2); assertEquals(mux.getAndAdvanceCurrentIndex(), 2); assertEquals(mux.getAndAdvanceCurrentIndex(), 3); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); }
@Test public void testCustomPattern() { Configuration conf = new Configuration(); conf.setStrings("test.custom." + IPC_CALLQUEUE_WRRMUX_WEIGHTS_KEY, "1", "1"); mux = new WeightedRoundRobinMultiplexer(2, "test.custom", conf); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); conf.setStrings("test.custom." + IPC_CALLQUEUE_WRRMUX_WEIGHTS_KEY, "1", "3", "2"); mux = new WeightedRoundRobinMultiplexer(3, "test.custom", conf); for(int i = 0; i < 5; i++) { assertEquals(mux.getAndAdvanceCurrentIndex(), 0); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 1); assertEquals(mux.getAndAdvanceCurrentIndex(), 2); assertEquals(mux.getAndAdvanceCurrentIndex(), 2); } }