@Test public void testCallingOnEventThatEachExecutorIsCalledInRoundRobinFashion() throws Exception { Configuration config = new SystemPropertiesConfiguration(); RelatedItemSearchExecutor executor1 = mock(RelatedItemSearchExecutor.class); RelatedItemSearchExecutor executor2 = mock(RelatedItemSearchExecutor.class); RelatedItemSearchExecutor executor3 = mock(RelatedItemSearchExecutor.class); RelatedItemSearchExecutor executor4 = mock(RelatedItemSearchExecutor.class); RelatedItemSearchResultsToResponseGateway gateway = mock(RelatedItemSearchResultsToResponseGateway.class); handler = new RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler(gateway,new RelatedItemSearchExecutor[]{executor1,executor2,executor3,executor4}); RelatedItemSearchRequest r1 = new RelatedItemSearchRequest(config); r1.getSearchRequest().setLookupKey(new SipHashSearchRequestLookupKey("1")); handler.onEvent(r1,1,true); r1.getSearchRequest().setLookupKey(new SipHashSearchRequestLookupKey("2")); handler.onEvent(r1,1,true); r1.getSearchRequest().setLookupKey(new SipHashSearchRequestLookupKey("3")); handler.onEvent(r1,1,true); r1.getSearchRequest().setLookupKey(new SipHashSearchRequestLookupKey("4")); handler.onEvent(r1,1,true); verify(executor1,times(1)).executeSearch(any(RelatedItemSearch.class)); verify(executor2,times(1)).executeSearch(any(RelatedItemSearch.class)); verify(executor3,times(1)).executeSearch(any(RelatedItemSearch.class)); verify(executor4,times(1)).executeSearch(any(RelatedItemSearch.class)); }
@Override public void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch) throws Exception { handleRequest(event,searchRequestExecutor[this.currentIndex++ & mask]); }
RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler implements RelatedContentSearchRequestProcessorHandler { @Override public void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch) throws Exception { handleRequest(event,searchRequestExecutor[this.currentIndex++ & mask]); } }
RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler implements RelatedContentSearchRequestProcessorHandler { @Override public void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch) throws Exception { handleRequest(event,searchRequestExecutor[this.currentIndex++ & mask]); } RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler(RelatedItemSearchResultsToResponseGateway contextStorage,
                                                                               RelatedItemSearchExecutor[] searchExecutor); }
RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler implements RelatedContentSearchRequestProcessorHandler { @Override public void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch) throws Exception { handleRequest(event,searchRequestExecutor[this.currentIndex++ & mask]); } RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler(RelatedItemSearchResultsToResponseGateway contextStorage,
                                                                               RelatedItemSearchExecutor[] searchExecutor); @Override void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch); void handleRequest(RelatedItemSearchRequest searchRequest, RelatedItemSearchExecutor searchExecutor); void shutdown(); }
RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler implements RelatedContentSearchRequestProcessorHandler { @Override public void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch) throws Exception { handleRequest(event,searchRequestExecutor[this.currentIndex++ & mask]); } RoundRobinDisruptorBasedRelatedContentSearchRequestProcessorHandler(RelatedItemSearchResultsToResponseGateway contextStorage,
                                                                               RelatedItemSearchExecutor[] searchExecutor); @Override void onEvent(RelatedItemSearchRequest event, long sequence, boolean endOfBatch); void handleRequest(RelatedItemSearchRequest searchRequest, RelatedItemSearchExecutor searchExecutor); void shutdown();  }