@Test public void testSendingManyItemsExceedingBatchSize() { CountDownLatch latch = new CountDownLatch(78); repo.handlers.get(0).setEndOfBatchCountDownLatch(latch); repo.handlers.get(1).setEndOfBatchCountDownLatch(latch); try { for(int i=0;i<25;i++) { handler.onEvent(getMessage(),i,false); } handler.onEvent(getMessage(),26,true); } catch(Exception e) { fail(); } try { boolean countedDown = latch.await(10000, TimeUnit.MILLISECONDS); assertTrue("Storage Repository not called in required time",countedDown); } catch (InterruptedException e) { fail("Timed out waiting for messages to be sent through the ring buffer"); e.printStackTrace(); } assertEquals(51,repo.handlers.get(0).getNumberOfCalls()); assertTrue(repo.handlers.get(0).getNumberOfEndOfBatchCalls()>0); assertEquals(27,repo.handlers.get(1).getNumberOfCalls()); assertTrue(repo.handlers.get(1).getNumberOfEndOfBatchCalls()>0); }
@Override public void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch) throws Exception { try { if(request.isValidMessage()) { if(request.getRelatedItems().getNumberOfRelatedItems()==0) { log.info("indexing message not valid. ignoring. No related products"); return; } for(RelatedItem p : converter.convertFrom(request)) batchMessages.add(p); if(endOfBatch || batchMessages.size()>=batchSize) { log.debug("handing off request to indexing processor"); try { disruptors[nextDisruptor[COUNTER_POS]++ & mask].publishEvents(BatchCopyingRelatedItemIndexMessageTranslator.INSTANCE,batchMessages.toArray(new RelatedItem[batchMessages.size()])); } finally { batchMessages.clear(); } } } else { log.info("indexing message not valid, and will be ignored. Potentially contained {} related products", request.getRelatedItems().getNumberOfRelatedItems()); } } finally { request.setValidMessage(false); } }
RoundRobinRelatedItemIndexingMessageEventHandler implements RelatedItemIndexingMessageEventHandler { @Override public void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch) throws Exception { try { if(request.isValidMessage()) { if(request.getRelatedItems().getNumberOfRelatedItems()==0) { log.info("indexing message not valid. ignoring. No related products"); return; } for(RelatedItem p : converter.convertFrom(request)) batchMessages.add(p); if(endOfBatch || batchMessages.size()>=batchSize) { log.debug("handing off request to indexing processor"); try { disruptors[nextDisruptor[COUNTER_POS]++ & mask].publishEvents(BatchCopyingRelatedItemIndexMessageTranslator.INSTANCE,batchMessages.toArray(new RelatedItem[batchMessages.size()])); } finally { batchMessages.clear(); } } } else { log.info("indexing message not valid, and will be ignored. Potentially contained {} related products", request.getRelatedItems().getNumberOfRelatedItems()); } } finally { request.setValidMessage(false); } } }
RoundRobinRelatedItemIndexingMessageEventHandler implements RelatedItemIndexingMessageEventHandler { @Override public void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch) throws Exception { try { if(request.isValidMessage()) { if(request.getRelatedItems().getNumberOfRelatedItems()==0) { log.info("indexing message not valid. ignoring. No related products"); return; } for(RelatedItem p : converter.convertFrom(request)) batchMessages.add(p); if(endOfBatch || batchMessages.size()>=batchSize) { log.debug("handing off request to indexing processor"); try { disruptors[nextDisruptor[COUNTER_POS]++ & mask].publishEvents(BatchCopyingRelatedItemIndexMessageTranslator.INSTANCE,batchMessages.toArray(new RelatedItem[batchMessages.size()])); } finally { batchMessages.clear(); } } } else { log.info("indexing message not valid, and will be ignored. Potentially contained {} related products", request.getRelatedItems().getNumberOfRelatedItems()); } } finally { request.setValidMessage(false); } } RoundRobinRelatedItemIndexingMessageEventHandler(final Configuration configuration,
                                                            RelatedItemIndexingMessageConverter converter,
                                                            RelatedItemReferenceMessageFactory messageFactory,
                                                            RelatedItemReferenceEventHandlerFactory relatedItemIndexingEventHandlerFactory
    ); }
RoundRobinRelatedItemIndexingMessageEventHandler implements RelatedItemIndexingMessageEventHandler { @Override public void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch) throws Exception { try { if(request.isValidMessage()) { if(request.getRelatedItems().getNumberOfRelatedItems()==0) { log.info("indexing message not valid. ignoring. No related products"); return; } for(RelatedItem p : converter.convertFrom(request)) batchMessages.add(p); if(endOfBatch || batchMessages.size()>=batchSize) { log.debug("handing off request to indexing processor"); try { disruptors[nextDisruptor[COUNTER_POS]++ & mask].publishEvents(BatchCopyingRelatedItemIndexMessageTranslator.INSTANCE,batchMessages.toArray(new RelatedItem[batchMessages.size()])); } finally { batchMessages.clear(); } } } else { log.info("indexing message not valid, and will be ignored. Potentially contained {} related products", request.getRelatedItems().getNumberOfRelatedItems()); } } finally { request.setValidMessage(false); } } RoundRobinRelatedItemIndexingMessageEventHandler(final Configuration configuration,
                                                            RelatedItemIndexingMessageConverter converter,
                                                            RelatedItemReferenceMessageFactory messageFactory,
                                                            RelatedItemReferenceEventHandlerFactory relatedItemIndexingEventHandlerFactory
    ); @Override void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch); void shutdown(); }
RoundRobinRelatedItemIndexingMessageEventHandler implements RelatedItemIndexingMessageEventHandler { @Override public void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch) throws Exception { try { if(request.isValidMessage()) { if(request.getRelatedItems().getNumberOfRelatedItems()==0) { log.info("indexing message not valid. ignoring. No related products"); return; } for(RelatedItem p : converter.convertFrom(request)) batchMessages.add(p); if(endOfBatch || batchMessages.size()>=batchSize) { log.debug("handing off request to indexing processor"); try { disruptors[nextDisruptor[COUNTER_POS]++ & mask].publishEvents(BatchCopyingRelatedItemIndexMessageTranslator.INSTANCE,batchMessages.toArray(new RelatedItem[batchMessages.size()])); } finally { batchMessages.clear(); } } } else { log.info("indexing message not valid, and will be ignored. Potentially contained {} related products", request.getRelatedItems().getNumberOfRelatedItems()); } } finally { request.setValidMessage(false); } } RoundRobinRelatedItemIndexingMessageEventHandler(final Configuration configuration,
                                                            RelatedItemIndexingMessageConverter converter,
                                                            RelatedItemReferenceMessageFactory messageFactory,
                                                            RelatedItemReferenceEventHandlerFactory relatedItemIndexingEventHandlerFactory
    ); @Override void onEvent(RelatedItemIndexingMessage request, long l, boolean endOfBatch); void shutdown();  }