@Test public void testBatching() throws IOException, InterruptedException { runner.setProperty(ListenRELP.MAX_BATCH_SIZE, "5"); final List<RELPFrame> frames = new ArrayList<>(); frames.add(OPEN_FRAME); frames.add(SYSLOG_FRAME); frames.add(SYSLOG_FRAME); frames.add(SYSLOG_FRAME); frames.add(CLOSE_FRAME); run(frames, 1, 3, null); final List<ProvenanceEventRecord> events = runner.getProvenanceEvents(); Assert.assertNotNull(events); Assert.assertEquals(1, events.size()); final ProvenanceEventRecord event = events.get(0); Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType()); Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("relp")); final List<MockFlowFile> mockFlowFiles = runner.getFlowFilesForRelationship(ListenRELP.REL_SUCCESS); Assert.assertEquals(1, mockFlowFiles.size()); final MockFlowFile mockFlowFile = mockFlowFiles.get(0); Assert.assertEquals(SYSLOG_FRAME.getCommand(), mockFlowFile.getAttribute(ListenRELP.RELPAttributes.COMMAND.key())); Assert.assertTrue(!StringUtils.isBlank(mockFlowFile.getAttribute(ListenRELP.RELPAttributes.PORT.key()))); Assert.assertTrue(!StringUtils.isBlank(mockFlowFile.getAttribute(ListenRELP.RELPAttributes.SENDER.key()))); }
@Override protected String getTransitUri(FlowFileEventBatch batch) { final String sender = batch.getEvents().get(0).getSender(); final String senderHost = sender.startsWith("/") && sender.length() > 1 ? sender.substring(1) : sender; final String transitUri = new StringBuilder().append("relp").append(": .append(port).toString(); return transitUri; }
ListenRELP extends AbstractListenEventBatchingProcessor<RELPEvent> { @Override protected String getTransitUri(FlowFileEventBatch batch) { final String sender = batch.getEvents().get(0).getSender(); final String senderHost = sender.startsWith("/") && sender.length() > 1 ? sender.substring(1) : sender; final String transitUri = new StringBuilder().append("relp").append(": .append(port).toString(); return transitUri; } }
ListenRELP extends AbstractListenEventBatchingProcessor<RELPEvent> { @Override protected String getTransitUri(FlowFileEventBatch batch) { final String sender = batch.getEvents().get(0).getSender(); final String senderHost = sender.startsWith("/") && sender.length() > 1 ? sender.substring(1) : sender; final String transitUri = new StringBuilder().append("relp").append(": .append(port).toString(); return transitUri; }  }
ListenRELP extends AbstractListenEventBatchingProcessor<RELPEvent> { @Override protected String getTransitUri(FlowFileEventBatch batch) { final String sender = batch.getEvents().get(0).getSender(); final String senderHost = sender.startsWith("/") && sender.length() > 1 ? sender.substring(1) : sender; final String transitUri = new StringBuilder().append("relp").append(": .append(port).toString(); return transitUri; }  @Override @OnScheduled void onScheduled(ProcessContext context); }
ListenRELP extends AbstractListenEventBatchingProcessor<RELPEvent> { @Override protected String getTransitUri(FlowFileEventBatch batch) { final String sender = batch.getEvents().get(0).getSender(); final String senderHost = sender.startsWith("/") && sender.length() > 1 ? sender.substring(1) : sender; final String transitUri = new StringBuilder().append("relp").append(": .append(port).toString(); return transitUri; }  @Override @OnScheduled void onScheduled(ProcessContext context); static final PropertyDescriptor SSL_CONTEXT_SERVICE; }