MapBasedMessageIdTracker implements MessageIdTracker { public int getNextMessageId() { int result = Message.NONE; boolean wrapped = false; final long now = ClockUtil.nanoRealtime(); synchronized (messageIds) { counter = (counter & 0xffff) % range; int startIdx = counter; while (result < 0 && !wrapped) { int idx = counter++ % range; Long earliestUsage = messageIds.get(idx); if (earliestUsage == null || (earliestUsage - now) <= 0) { result = idx + min; messageIds.put(idx, now + exchangeLifetimeNanos); } wrapped = (counter % range) == startIdx; } } return result; } MapBasedMessageIdTracker(int initialMid, int minMid, int maxMid, NetworkConfig config); int getNextMessageId();  }
@Test public void testGetNextMessageIdFailsIfAllMidsAreInUse() throws Exception { NetworkConfig config = network.createStandardTestConfig(); MapBasedMessageIdTracker tracker = new MapBasedMessageIdTracker(INITIAL_MID, 0, TOTAL_NO_OF_MIDS, config); for (int i = 0; i < TOTAL_NO_OF_MIDS; i++) { tracker.getNextMessageId(); } int mid = tracker.getNextMessageId(); assertThat(mid, is(-1)); }
@Test public void testGetNextMessageIdFailsIfAllMidsInRangeAreInUse() throws Exception { NetworkConfig config = network.createStandardTestConfig(); final int minMid = 1024; final int maxMid = 2048; final int rangeMid = maxMid - minMid; MapBasedMessageIdTracker tracker = new MapBasedMessageIdTracker(INITIAL_MID + minMid, minMid, maxMid, config); for (int i = 0; i < rangeMid; i++) { int mid = tracker.getNextMessageId(); assertThat(mid, is(inRange(minMid, maxMid))); } int mid = tracker.getNextMessageId(); assertThat(mid, is(-1)); }
@Test public void testGetNextMessageIdReusesIdAfterExchangeLifetime() throws Exception { int exchangeLifetime = 100; NetworkConfig config = network.createStandardTestConfig(); config.setInt(NetworkConfig.Keys.EXCHANGE_LIFETIME, exchangeLifetime); final MapBasedMessageIdTracker tracker = new MapBasedMessageIdTracker(INITIAL_MID, 0, TOTAL_NO_OF_MIDS, config); int firstMid = tracker.getNextMessageId(); for (int i = 1; i < TOTAL_NO_OF_MIDS; i++) { tracker.getNextMessageId(); } exchangeLifetime += (exchangeLifetime >> 1); final AtomicInteger mid = new AtomicInteger(-1); TestConditionTools.waitForCondition(exchangeLifetime, 10, TimeUnit.MILLISECONDS, new TestCondition() { @Override public boolean isFulFilled() throws IllegalStateException { mid.set(tracker.getNextMessageId()); return 0 <= mid.get(); } }); assertThat(mid.get(), is(firstMid)); }