@Test public void testErrorQueue() throws IOException { final List<ListenSyslog.RawSyslogEvent> msgs = new ArrayList<>(); msgs.add(new ListenSyslog.RawSyslogEvent(VALID_MESSAGE.getBytes(), "sender-01")); msgs.add(new ListenSyslog.RawSyslogEvent(VALID_MESSAGE.getBytes(), "sender-01")); final AtomicInteger getMessageAttempts = new AtomicInteger(0); msgs.add(new ListenSyslog.RawSyslogEvent(VALID_MESSAGE.getBytes(), "sender-01") { @Override public byte[] getData() { final int attempts = getMessageAttempts.incrementAndGet(); if (attempts == 1) { throw new FlowFileAccessException("Unit test failure"); } else { return VALID_MESSAGE.getBytes(); } } }); final CannedMessageProcessor proc = new CannedMessageProcessor(msgs); final TestRunner runner = TestRunners.newTestRunner(proc); runner.setProperty(ListenSyslog.MAX_BATCH_SIZE, "5"); runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.UDP_VALUE.getValue()); runner.setProperty(ListenSyslog.PORT, "0"); runner.setProperty(ListenSyslog.PARSE_MESSAGES, "false"); runner.run(); assertEquals(1, proc.getErrorQueueSize()); runner.assertAllFlowFilesTransferred(ListenSyslog.REL_SUCCESS, 1); runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0).assertContentEquals(VALID_MESSAGE + "\n" + VALID_MESSAGE); runner.clearTransferState(); runner.run(); runner.assertAllFlowFilesTransferred(ListenSyslog.REL_SUCCESS, 1); runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0).assertContentEquals(VALID_MESSAGE); }
protected int getErrorQueueSize() { return errorEvents.size(); }
ListenSyslog extends AbstractSyslogProcessor { protected int getErrorQueueSize() { return errorEvents.size(); } }
ListenSyslog extends AbstractSyslogProcessor { protected int getErrorQueueSize() { return errorEvents.size(); }  }
ListenSyslog extends AbstractSyslogProcessor { protected int getErrorQueueSize() { return errorEvents.size(); }  @Override Set<Relationship> getRelationships(); @Override List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue); @OnScheduled void onScheduled(final ProcessContext context); @OnUnscheduled void onUnscheduled(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
ListenSyslog extends AbstractSyslogProcessor { protected int getErrorQueueSize() { return errorEvents.size(); }  @Override Set<Relationship> getRelationships(); @Override List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue); @OnScheduled void onScheduled(final ProcessContext context); @OnUnscheduled void onUnscheduled(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); static final PropertyDescriptor MAX_MESSAGE_QUEUE_SIZE; static final PropertyDescriptor RECV_BUFFER_SIZE; static final PropertyDescriptor MAX_SOCKET_BUFFER_SIZE; static final PropertyDescriptor MAX_CONNECTIONS; static final PropertyDescriptor MAX_BATCH_SIZE; static final PropertyDescriptor MESSAGE_DELIMITER; static final PropertyDescriptor PARSE_MESSAGES; static final PropertyDescriptor SSL_CONTEXT_SERVICE; static final Relationship REL_SUCCESS; static final Relationship REL_INVALID; }