@Test public void testHandlerIsNotExecutedWithIndexingRequestWithNoProducts() { String json = "{" + " \"channel\" : 1.0," + " \"site\" : \"amazon\"," + " \"items\" : [ ]"+ "}"; processor.processRequest(configuation, ByteBuffer.wrap(json.getBytes())); assertEquals(0,handler.getNumberOfCalls()); }
@Override public IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data) { if(canIndex) { try { boolean published = ringBuffer.tryPublishEvent(requestConverter.createConverter(config,data)); return published ? IndexRequestPublishingStatus.PUBLISHED : IndexRequestPublishingStatus.NO_SPACE_AVALIABLE; } catch(InvalidIndexingRequestException e) { log.warn("Invalid json content, unable to process request: {}", e.getMessage()); if(log.isDebugEnabled()) { if(canOutputRequestData && data.hasArray()) { log.debug("content requested to be indexed: {}", Arrays.toString(data.array())); } } return IndexRequestPublishingStatus.FAILED; } } else { log.error("The indexing processor has been shutdown, and cannot accept requests."); return IndexRequestPublishingStatus.PUBLISHER_SHUTDOWN; } }
DisruptorBasedRelatedItemIndexRequestProcessor implements RelatedItemIndexRequestProcessor { @Override public IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data) { if(canIndex) { try { boolean published = ringBuffer.tryPublishEvent(requestConverter.createConverter(config,data)); return published ? IndexRequestPublishingStatus.PUBLISHED : IndexRequestPublishingStatus.NO_SPACE_AVALIABLE; } catch(InvalidIndexingRequestException e) { log.warn("Invalid json content, unable to process request: {}", e.getMessage()); if(log.isDebugEnabled()) { if(canOutputRequestData && data.hasArray()) { log.debug("content requested to be indexed: {}", Arrays.toString(data.array())); } } return IndexRequestPublishingStatus.FAILED; } } else { log.error("The indexing processor has been shutdown, and cannot accept requests."); return IndexRequestPublishingStatus.PUBLISHER_SHUTDOWN; } } }
DisruptorBasedRelatedItemIndexRequestProcessor implements RelatedItemIndexRequestProcessor { @Override public IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data) { if(canIndex) { try { boolean published = ringBuffer.tryPublishEvent(requestConverter.createConverter(config,data)); return published ? IndexRequestPublishingStatus.PUBLISHED : IndexRequestPublishingStatus.NO_SPACE_AVALIABLE; } catch(InvalidIndexingRequestException e) { log.warn("Invalid json content, unable to process request: {}", e.getMessage()); if(log.isDebugEnabled()) { if(canOutputRequestData && data.hasArray()) { log.debug("content requested to be indexed: {}", Arrays.toString(data.array())); } } return IndexRequestPublishingStatus.FAILED; } } else { log.error("The indexing processor has been shutdown, and cannot accept requests."); return IndexRequestPublishingStatus.PUBLISHER_SHUTDOWN; } } DisruptorBasedRelatedItemIndexRequestProcessor(Configuration configuration,
                                                          IndexingRequestConverterFactory requestConverter,
                                                          EventFactory<RelatedItemIndexingMessage> indexingMessageFactory,
                                                          RelatedItemIndexingMessageEventHandler eventHandler
    ); }
DisruptorBasedRelatedItemIndexRequestProcessor implements RelatedItemIndexRequestProcessor { @Override public IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data) { if(canIndex) { try { boolean published = ringBuffer.tryPublishEvent(requestConverter.createConverter(config,data)); return published ? IndexRequestPublishingStatus.PUBLISHED : IndexRequestPublishingStatus.NO_SPACE_AVALIABLE; } catch(InvalidIndexingRequestException e) { log.warn("Invalid json content, unable to process request: {}", e.getMessage()); if(log.isDebugEnabled()) { if(canOutputRequestData && data.hasArray()) { log.debug("content requested to be indexed: {}", Arrays.toString(data.array())); } } return IndexRequestPublishingStatus.FAILED; } } else { log.error("The indexing processor has been shutdown, and cannot accept requests."); return IndexRequestPublishingStatus.PUBLISHER_SHUTDOWN; } } DisruptorBasedRelatedItemIndexRequestProcessor(Configuration configuration,
                                                          IndexingRequestConverterFactory requestConverter,
                                                          EventFactory<RelatedItemIndexingMessage> indexingMessageFactory,
                                                          RelatedItemIndexingMessageEventHandler eventHandler
    ); @Override IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data); @PreDestroy void shutdown(); }
DisruptorBasedRelatedItemIndexRequestProcessor implements RelatedItemIndexRequestProcessor { @Override public IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data) { if(canIndex) { try { boolean published = ringBuffer.tryPublishEvent(requestConverter.createConverter(config,data)); return published ? IndexRequestPublishingStatus.PUBLISHED : IndexRequestPublishingStatus.NO_SPACE_AVALIABLE; } catch(InvalidIndexingRequestException e) { log.warn("Invalid json content, unable to process request: {}", e.getMessage()); if(log.isDebugEnabled()) { if(canOutputRequestData && data.hasArray()) { log.debug("content requested to be indexed: {}", Arrays.toString(data.array())); } } return IndexRequestPublishingStatus.FAILED; } } else { log.error("The indexing processor has been shutdown, and cannot accept requests."); return IndexRequestPublishingStatus.PUBLISHER_SHUTDOWN; } } DisruptorBasedRelatedItemIndexRequestProcessor(Configuration configuration,
                                                          IndexingRequestConverterFactory requestConverter,
                                                          EventFactory<RelatedItemIndexingMessage> indexingMessageFactory,
                                                          RelatedItemIndexingMessageEventHandler eventHandler
    ); @Override IndexRequestPublishingStatus processRequest(Configuration config, ByteBuffer data); @PreDestroy void shutdown();  }