* A stream is a sequence of events and the way to interact with a stream is to * register actions. The {@code EventStream} interface is not to be implemented * and future versions of the JDK may prevent this completely. *
* To receive a notification when an event arrives, register an action using the * {@link #onEvent(Consumer)} method. To filter the stream for an event with a * specific name, use {@link #onEvent(String, Consumer)} method. *
* By default, the same {@code RecordedEvent} object can be used to * represent two or more distinct events. That object can be delivered * multiple times to the same action as well as to other actions. To use an * event object after the action is completed, the * {@link #setReuse(boolean)} method should be set to {@code false} so a * new object is allocated for each event. *
* Events are delivered in batches. To receive a notification when a batch is * complete, register an action using the {@link #onFlush(Runnable)} method. * This is an opportunity to aggregate or push data to external systems while * the Java Virtual Machine (JVM) is preparing the next batch. *
* Events within a batch are sorted chronologically by their end time. * Well-ordering of events is only maintained for events available to the JVM at * the point of flush, i.e. for the set of events delivered as a unit in a * single batch. Events delivered in a batch could therefore be out-of-order * compared to events delivered in a previous batch, but never out-of-order with * events within the same batch. If ordering is not a concern, sorting can be * disabled using the {@link #setOrdered(boolean)} method. *
* To dispatch events to registered actions, the stream must be started. To * start processing in the current thread, invoke the {@link #start()} method. * To process actions asynchronously in a separate thread, invoke the * {@link #startAsync()} method. To await completion of the stream, use the * awaitTermination {@link #awaitTermination()} or the * {@link #awaitTermination(Duration)} method. *
* When a stream ends it is automatically closed. To manually stop processing of * events, close the stream by invoking the {@link #close()} method. A stream * can also be automatically closed in exceptional circumstances, for example if * the JVM that is being monitored exits. To receive a notification in any of * these occasions, use the {@link #onClose(Runnable)} method to register an * action. *
* If an unexpected exception occurs in an action, it is possible to catch the * exception in an error handler. An error handler can be registered using the * {@link #onError(Runnable)} method. If no error handler is registered, the * default behavior is to print the exception and its backtrace to the standard * error stream. *
* The following example shows how an {@code EventStream} can be used to listen * to events on a JVM running Flight Recorder * *
*
* try (var es = EventStream.openRepository()) {
* es.onEvent("jdk.CPULoad", event -> {
* System.out.println("CPU Load " + event.getEndTime());
* System.out.println(" Machine total: " + 100 * event.getFloat("machineTotal") + "%");
* System.out.println(" JVM User: " + 100 * event.getFloat("jvmUser") + "%");
* System.out.println(" JVM System: " + 100 * event.getFloat("jvmSystem") + "%");
* System.out.println();
* });
* es.onEvent("jdk.GarbageCollection", event -> {
* System.out.println("Garbage collection: " + event.getLong("gcId"));
* System.out.println(" Cause: " + event.getString("cause"));
* System.out.println(" Total pause: " + event.getDuration("sumOfPauses"));
* System.out.println(" Longest pause: " + event.getDuration("longestPause"));
* System.out.println();
* });
* es.start();
* }
*
*
* * To start recording together with the stream, see {@link RecordingStream}. * * @since 14 */ public interface EventStream extends AutoCloseable { /** * Creates a stream from the repository of the current Java Virtual Machine * (JVM). *
* By default, the stream starts with the next event flushed by Flight * Recorder. * * @return an event stream, not {@code null} * * @throws IOException if a stream can't be opened, or an I/O error occurs * when trying to access the repository * * @throws SecurityException if a security manager exists and the caller * does not have * {@code FlightRecorderPermission("accessFlightRecorder")} */ public static EventStream openRepository() throws IOException { Utils.checkAccessFlightRecorder(); return new EventDirectoryStream(AccessController.getContext(), null, SecuritySupport.PRIVILIGED, false); } /** * Creates an event stream from a disk repository. *
* By default, the stream starts with the next event flushed by Flight * Recorder. * * @param directory location of the disk repository, not {@code null} * * @return an event stream, not {@code null} * * @throws IOException if a stream can't be opened, or an I/O error occurs * when trying to access the repository * * @throws SecurityException if a security manager exists and its * {@code checkRead} method denies read access to the directory, or * files in the directory. */ public static EventStream openRepository(Path directory) throws IOException { Objects.nonNull(directory); AccessControlContext acc = AccessController.getContext(); return new EventDirectoryStream(acc, directory, FileAccess.UNPRIVILIGED, false); } /** * Creates an event stream from a file. *
* By default, the stream starts with the first event in the file.
*
* @param file location of the file, not {@code null}
*
* @return an event stream, not {@code null}
*
* @throws IOException if the file can't be opened, or an I/O error occurs
* during reading
*
* @throws SecurityException if a security manager exists and its
* {@code checkRead} method denies read access to the file
*/
static EventStream openFile(Path file) throws IOException {
return new EventFileStream(AccessController.getContext(), file);
}
/**
* Registers an action to perform on all events in the stream.
*
* @param action an action to perform on each {@code RecordedEvent}, not
* {@code null}
*/
void onEvent(Consumer
* if an action is not registered, an exception stack trace is printed to
* standard error.
*
* Registering an action overrides the default behavior. If multiple actions
* have been registered, they are performed in the order of registration.
*
* If this method itself throws an exception, resulting behavior is
* undefined.
*
* @param action an action to perform if an exception occurs, not
* {@code null}
*/
void onError(Consumer
* If the stream is already closed, the action will be performed immediately
* in the current thread.
*
* @param action an action to perform after the stream is closed, not
* {@code null}
* @see #close()
*/
void onClose(Runnable action);
/**
* Releases all resources associated with this stream.
*
* Closing a previously closed stream has no effect.
*/
void close();
/**
* Unregisters an action.
*
* If the action has been registered multiple times, all instances are
* unregistered.
*
* @param action the action to unregister, not {@code null}
*
* @return {@code true} if the action was unregistered, {@code false}
* otherwise
*
* @see #onEvent(Consumer)
* @see #onEvent(String, Consumer)
* @see #onFlush(Runnable)
* @see #onClose(Runnable)
* @see #onError(Consumer)
*/
boolean remove(Object action);
/**
* Specifies that the event object in an {@link #onEvent(Consumer)} action
* can be reused.
*
* If reuse is set to {@code true), an action should not keep a reference
* to the event object after the action has completed.
*
* @param reuse {@code true} if an event object can be reused, {@code false}
* otherwise
*/
void setReuse(boolean reuse);
/**
* Specifies that events arrives in chronological order, sorted by the time
* they were committed to the stream.
*
* @param ordered if event objects arrive in chronological order to
* {@code #onEvent(Consumer)}
*/
void setOrdered(boolean ordered);
/**
* Specifies the start time of the stream.
*
* The start time must be set before starting the stream
*
* @param startTime the start time, not {@code null}
*
* @throws IllegalStateException if the stream is already started
*
* @see #start()
* @see #startAsync()
*/
void setStartTime(Instant startTime);
/**
* Specifies the end time of the stream.
*
* The end time must be set before starting the stream.
*
* At end time, the stream is closed.
*
* @param endTime the end time, not {@code null}
*
* @throws IllegalStateException if the stream is already started
*
* @see #start()
* @see #startAsync()
*/
void setEndTime(Instant endTime);
/**
* Start processing of actions.
*
* Actions are performed in the current thread.
*
* @throws IllegalStateException if the stream is already started or closed
*/
void start();
/**
* Start asynchronous processing of actions.
*
* Actions are performed in a single separate thread.
*
* @throws IllegalStateException if the stream is already started or closed
*/
void startAsync();
/**
* Blocks until all actions are completed, or the stream is closed, or the
* timeout occurs, or the current thread is interrupted, whichever happens
* first.
*
* @param timeout the maximum time to wait, not {@code null}
*
* @throws IllegalArgumentException if timeout is negative
* @throws InterruptedException if interrupted while waiting
*
* @see #start()
* @see #startAsync()
* @see Thread#interrupt()
*/
void awaitTermination(Duration timeout) throws InterruptedException;
/**
* Blocks until all actions are completed, or the stream is closed, or the
* current thread is interrupted, whichever happens first.
*
* @throws InterruptedException if interrupted while waiting
*
* @see #start()
* @see #startAsync()
* @see Thread#interrupt()
*/
void awaitTermination() throws InterruptedException;
}