try {
outputStream.flush();
outputStream.close();
shouldRotate = false;
} catch (IOException e) {
throw new EventDeliveryException("Unable to rotate file "
+ pathController.getCurrentFile() + " while delivering event", e);
}
outputStream = null;
pathController.rotate();
}
}
if (outputStream == null) {
try {
logger.debug("Opening output stream for file {}",
pathController.getCurrentFile());
outputStream = new BufferedOutputStream(new FileOutputStream(
pathController.getCurrentFile()));
} catch (IOException e) {
throw new EventDeliveryException("Failed to open file "
+ pathController.getCurrentFile() + " while delivering event", e);
}
}
Channel channel = getChannel();
Transaction transaction = channel.getTransaction();
Event event = null;
Status result = Status.READY;
try {
transaction.begin();
event = channel.take();
if (event != null) {
byte[] bytes = formatter.format(event);
outputStream.write(bytes);
/*
* FIXME: Feature: Rotate on size and time by checking bytes written and
* setting shouldRotate = true if we're past a threshold.
*/
counterGroup.addAndGet("sink.bytesWritten", (long) bytes.length);
/*
* FIXME: Feature: Control flush interval based on time or number of
* events. For now, we're super-conservative and flush on each write.
*/
outputStream.flush();
} else {
// No events found, request back-off semantics from runner
result = Status.BACKOFF;
}
transaction.commit();
} catch (Exception ex) {
transaction.rollback();
throw new EventDeliveryException("Failed to process event: " + event, ex);
} finally {
transaction.close();
}
return result;