--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventConsumer.java Wed Jul 03 22:51:44 2019 +0200
@@ -0,0 +1,393 @@
+/*
+ * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation. Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.jfr.consumer;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+import jdk.jfr.EventType;
+import jdk.jfr.internal.JVM;
+import jdk.jfr.internal.LogLevel;
+import jdk.jfr.internal.LogTag;
+import jdk.jfr.internal.Logger;
+import jdk.jfr.internal.LongMap;
+import jdk.jfr.internal.consumer.InternalEventFilter;
+
+abstract class EventConsumer implements Runnable {
+
+ public final static Instant NEXT_EVENT = Instant.now();
+
+ final static class EventDispatcher {
+ public final static EventDispatcher[] NO_DISPATCHERS = new EventDispatcher[0];
+
+ final private String eventName;
+ final Consumer<RecordedEvent> action;
+
+ public EventDispatcher(Consumer<RecordedEvent> action) {
+ this(null, action);
+ }
+
+ public EventDispatcher(String eventName, Consumer<RecordedEvent> action) {
+ this.eventName = eventName;
+ this.action = action;
+ }
+
+ public void offer(RecordedEvent event) {
+ action.accept(event);
+ }
+
+ public boolean accepts(EventType eventType) {
+ return (eventName == null || eventType.getName().equals(eventName));
+ }
+ }
+
+ private final static JVM jvm = JVM.getJVM();
+ private final static VarHandle closedHandle;
+ private final static VarHandle consumersHandle;
+ private final static VarHandle dispatcherHandle;
+ private final static VarHandle flushActionsHandle;
+ private final static VarHandle closeActionsHandle;
+ private final static VarHandle orderedHandle;
+ private final static VarHandle reuseHandle;
+ private final static VarHandle startTimeHandle;
+ static {
+ try {
+ MethodHandles.Lookup l = MethodHandles.lookup();
+ closedHandle = l.findVarHandle(EventConsumer.class, "closed", boolean.class);
+ consumersHandle = l.findVarHandle(EventConsumer.class, "consumers", EventDispatcher[].class);
+ dispatcherHandle = l.findVarHandle(EventConsumer.class, "dispatcher", LongMap.class);
+ flushActionsHandle = l.findVarHandle(EventConsumer.class, "flushActions", Runnable[].class);
+ closeActionsHandle = l.findVarHandle(EventConsumer.class, "closeActions", Runnable[].class);
+ orderedHandle = l.findVarHandle(EventConsumer.class, "ordered", boolean.class);
+ reuseHandle = l.findVarHandle(EventConsumer.class, "reuse", boolean.class);
+ startTimeHandle = l.findVarHandle(EventConsumer.class, "startTime", Instant.class);
+ } catch (ReflectiveOperationException e) {
+ throw new InternalError(e);
+ }
+ }
+ // set by VarHandle
+ private boolean closed;
+ // set by VarHandle
+ private EventDispatcher[] consumers = new EventDispatcher[0];
+ // set by VarHandle
+ private LongMap<EventDispatcher[]> dispatcher = new LongMap<>();
+ // set by VarHandle
+ private Runnable[] flushActions = new Runnable[0];
+ // set by VarHandle
+ private Runnable[] closeActions = new Runnable[0];
+
+ private final AccessControlContext accessControlContext;
+ protected volatile InternalEventFilter eventFilter = InternalEventFilter.ACCEPT_ALL;
+
+ private boolean started;
+ private Thread thread;
+
+ protected long startNanos;
+ protected boolean ordered = true;
+ protected boolean reuse = true;
+ Instant startTime;
+
+ public EventConsumer(AccessControlContext acc) throws IOException {
+ this.accessControlContext = acc;
+ }
+
+ public void run() {
+ doPriviliged(() -> execute());
+ }
+
+ void doPriviliged(Runnable r) {
+ AccessController.doPrivileged(new PrivilegedAction<Void>() {
+ @Override
+ public Void run() {
+ r.run();
+ return null;
+ }
+ }, accessControlContext);
+ }
+
+ private void execute() {
+ jvm.exclude(Thread.currentThread());
+ try {
+ updateStartNanos();
+ process();
+ } catch (IOException e) {
+ if (!isClosed()) {
+ logException(e);
+ }
+ } catch (Exception e) {
+ logException(e);
+ } finally {
+ Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Execution of stream ended.");
+ }
+ }
+
+ // User setting overrides default
+ private void updateStartNanos() {
+ if (startTime != null) {
+ try {
+ setStartNanos(startTime.toEpochMilli() * 1_000_000L);
+ } catch (ArithmeticException ae) {
+ setStartNanos(Long.MAX_VALUE);
+ }
+ }
+ }
+
+ private void logException(Exception e) {
+ e.printStackTrace(); // for debugging purposes, remove before
+ // integration
+ Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Unexpected error processing stream. " + e.getMessage());
+ }
+
+ public abstract void process() throws IOException;
+
+ public synchronized boolean remove(Object action) {
+ boolean remove = false;
+ Runnable[] updatedFlushActions = removeAction(flushActions, action);
+ if (updatedFlushActions != null) {
+ flushActionsHandle.setVolatile(this, updatedFlushActions);
+ remove = true;
+ }
+ Runnable[] updatedCloseActions = removeAction(closeActions, action);
+ if (updatedCloseActions != null) {
+ closeActionsHandle.setVolatile(this, updatedCloseActions);
+ remove = true;
+ }
+
+ boolean removeConsumer = false;
+ List<EventDispatcher> list = new ArrayList<>();
+ for (int i = 0; i < consumers.length; i++) {
+ if (consumers[i].action != action) {
+ list.add(consumers[i]);
+ } else {
+ removeConsumer = true;
+ remove = true;
+ }
+ }
+ if (removeConsumer) {
+ EventDispatcher[] array = list.toArray(new EventDispatcher[list.size()]);
+ eventFilter = buildFilter(array);
+ consumersHandle.setVolatile(this, array);
+ dispatcherHandle.setVolatile(this, new LongMap<>()); // will reset
+ // dispatch
+ }
+ return remove;
+ }
+
+ public void dispatch(RecordedEvent e) {
+ if (e.endTime < startNanos) {
+ return;
+ }
+
+ EventDispatcher[] consumerDispatch = dispatcher.get(e.getEventType().getId());
+ if (consumerDispatch == null) {
+ consumerDispatch = EventDispatcher.NO_DISPATCHERS;
+ for (EventDispatcher ec : consumers.clone()) {
+ if (ec.accepts(e.getEventType())) {
+ consumerDispatch = merge(consumerDispatch, ec);
+ }
+ }
+ dispatcher.put(e.getEventType().getId(), consumerDispatch);
+ }
+ for (int i = 0; i < consumerDispatch.length; i++) {
+ try {
+ consumerDispatch[i].offer(e);
+ } catch (Exception exception) {
+ // Is this a reasonable behavior for an exception?
+ // Error will abort the stream.
+ }
+ }
+
+ }
+
+ public void onEvent(Consumer<RecordedEvent> action) {
+ add(new EventDispatcher(action));
+ }
+
+ public void onEvent(String eventName, Consumer<RecordedEvent> action) {
+ add(new EventDispatcher(eventName, action));
+ }
+
+ InternalEventFilter buildFilter(EventDispatcher[] dispatchers) {
+ InternalEventFilter ef = new InternalEventFilter();
+ for (EventDispatcher ed : dispatchers) {
+ String name = ed.eventName;
+ if (name == null) {
+ return InternalEventFilter.ACCEPT_ALL;
+ }
+ ef.setThreshold(name, 0);
+ }
+ return ef.threadSafe();
+ }
+
+ private synchronized void add(EventDispatcher e) {
+ EventDispatcher[] dispatchers = merge(consumers, e);
+ eventFilter = buildFilter(dispatchers);
+ consumersHandle.setVolatile(this, dispatchers);
+ dispatcherHandle.setVolatile(this, new LongMap<>()); // will reset
+ }
+
+ public synchronized void onFlush(Runnable action) {
+ flushActionsHandle.setVolatile(this, addAction(flushActions, action));
+ }
+
+ public synchronized void addCloseAction(Runnable action) {
+ closeActionsHandle.setVolatile(this, addAction(closeActions, action));
+ }
+
+ public void setClosed(boolean closed) {
+ closedHandle.setVolatile(this, closed);
+ }
+
+ final public boolean isClosed() {
+ return closed;
+ }
+
+ public void runCloseActions() {
+
+ Runnable[] cas = this.closeActions;
+ for (int i = 0; i < cas.length; i++) {
+ cas[i].run();
+ }
+ }
+
+ public void runFlushActions() {
+ Runnable[] fas = this.flushActions;
+ for (int i = 0; i < fas.length; i++) {
+ fas[i].run();
+ }
+ }
+
+ public void startAsync(long startNanos) {
+ if (started) {
+ throw new IllegalStateException("Event stream can only be started once");
+ }
+ started = true;
+ setStartNanos(startNanos);
+ thread = new Thread(this);
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ public void start(long startNanos) {
+ synchronized (this) {
+ if (started) {
+ throw new IllegalStateException("Event stream can only be started once");
+ }
+ started = true;
+ setStartNanos(startNanos);
+ }
+ run();
+ }
+
+ public void awaitTermination(Duration timeout) {
+ Objects.requireNonNull(timeout);
+ Thread t = null;
+ synchronized (this) {
+ t = thread;
+ }
+ if (t != null && t != Thread.currentThread()) {
+ try {
+ t.join(timeout.toMillis());
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+ }
+
+ public void awaitTermination() {
+ awaitTermination(Duration.ofMillis(0));
+ }
+
+ private void setStartNanos(long startNanos) {
+ this.startNanos = startNanos;
+ }
+
+ protected static EventDispatcher[] merge(EventDispatcher[] current, EventDispatcher add) {
+ EventDispatcher[] array = new EventDispatcher[current.length + 1];
+ System.arraycopy(current, 0, array, 0, current.length);
+ array[current.length] = add;
+ return array;
+ }
+
+ private static Runnable[] removeAction(Runnable[] array, Object action) {
+ if (array.length == 0) {
+ return null;
+ }
+ boolean remove = false;
+ List<Runnable> list = new ArrayList<>();
+ for (int i = 0; i < array.length; i++) {
+ if (array[i] != action) {
+ list.add(array[i]);
+ } else {
+ remove = true;
+ }
+ }
+ if (remove) {
+ return list.toArray(new Runnable[list.size()]);
+ }
+ return null;
+ }
+
+ private static Runnable[] addAction(Runnable[] array, Runnable action) {
+ ArrayList<Runnable> a = new ArrayList<>();
+ a.addAll(Arrays.asList(array));
+ a.add(action);
+ return a.toArray(new Runnable[0]);
+ }
+
+ abstract public void close();
+
+ public void setReuse(boolean reuse) {
+ reuseHandle.setVolatile(this, reuse);
+ }
+
+ public void setOrdered(boolean ordered) {
+ orderedHandle.setVolatile(this, ordered);
+ }
+
+ public void setStartTime(Instant startTime) {
+ Objects.nonNull(startTime);
+ if (started) {
+ throw new IllegalStateException("Stream is already started");
+ }
+ if (startTime.isBefore(Instant.EPOCH)) {
+ startTime = Instant.EPOCH;
+ }
+ startTimeHandle.setVolatile(this, startTime);
+ }
+
+}
\ No newline at end of file
--- a/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventDirectoryStream.java Thu Jun 27 10:41:01 2019 +0200
+++ b/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventDirectoryStream.java Wed Jul 03 22:51:44 2019 +0200
@@ -35,7 +35,6 @@
import java.util.Objects;
import java.util.function.Consumer;
-import jdk.jfr.internal.consumer.EventConsumer;
import jdk.jfr.internal.consumer.RecordingInput;
import jdk.jfr.internal.consumer.RepositoryFiles;
@@ -46,23 +45,32 @@
*/
final class EventDirectoryStream implements EventStream {
- static final class ParserConsumer extends EventConsumer {
+ static final class DirectoryConsumer extends EventConsumer {
private static final Comparator<? super RecordedEvent> END_TIME = (e1, e2) -> Long.compare(e1.endTime, e2.endTime);
private static final int DEFAULT_ARRAY_SIZE = 10_000;
private final RepositoryFiles repositoryFiles;
private ChunkParser chunkParser;
private RecordedEvent[] sortedList;
+ protected long chunkStartNanos;
- public ParserConsumer(AccessControlContext acc, Path p) throws IOException {
+ public DirectoryConsumer(AccessControlContext acc, Path p) throws IOException {
super(acc);
repositoryFiles = new RepositoryFiles(p);
}
@Override
public void process() throws IOException {
- Path path = repositoryFiles.nextPath(startNanos);
- startNanos = repositoryFiles.getTimestamp(path) + 1;
+ chunkStartNanos = startNanos;
+ Path path;
+ if (startTime == EventConsumer.NEXT_EVENT) {
+ // TODO: Need to skip forward to the next event
+ // For now, use the last chunk.
+ path = repositoryFiles.lastPath();
+ } else {
+ path = repositoryFiles.nextPath(chunkStartNanos);
+ }
+ chunkStartNanos = repositoryFiles.getTimestamp(path) + 1;
try (RecordingInput input = new RecordingInput(path.toFile())) {
chunkParser = new ChunkParser(input, this.reuse);
while (!isClosed()) {
@@ -81,11 +89,11 @@
runFlushActions();
}
- path = repositoryFiles.nextPath(startNanos);
+ path = repositoryFiles.nextPath(chunkStartNanos);
if (path == null) {
return; // stream closed
}
- startNanos = repositoryFiles.getTimestamp(path) + 1;
+ chunkStartNanos = repositoryFiles.getTimestamp(path) + 1;
input.setFile(path);
chunkParser = chunkParser.newChunkParser();
}
@@ -147,8 +155,9 @@
private final EventConsumer eventConsumer;
- public EventDirectoryStream(AccessControlContext acc, Path p) throws IOException {
- eventConsumer = new ParserConsumer(acc, p);
+ public EventDirectoryStream(AccessControlContext acc, Path p, Instant startTime) throws IOException {
+ eventConsumer = new DirectoryConsumer(acc, p);
+ eventConsumer.startTime = startTime;
}
public void close() {
@@ -223,4 +232,9 @@
public void setOrdered(boolean ordered) {
eventConsumer.setOrdered(ordered);
}
+
+ @Override
+ public void setStartTime(Instant startTime) {
+ eventConsumer.setStartTime(startTime);
+ }
}
--- a/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventFileStream.java Thu Jun 27 10:41:01 2019 +0200
+++ b/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventFileStream.java Wed Jul 03 22:51:44 2019 +0200
@@ -30,12 +30,12 @@
import java.security.AccessControlContext;
import java.security.AccessController;
import java.time.Duration;
+import java.time.Instant;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Objects;
import java.util.function.Consumer;
-import jdk.jfr.internal.consumer.EventConsumer;
import jdk.jfr.internal.consumer.RecordingInput;
/**
@@ -44,7 +44,7 @@
*/
final class EventFileStream implements EventStream {
- private final static class FileEventConsumer extends EventConsumer {
+ private final static class FileConsumer extends EventConsumer {
private static final Comparator<? super RecordedEvent> END_TIME = (e1, e2) -> Long.compare(e1.endTime, e2.endTime);
private static final int DEFAULT_ARRAY_SIZE = 100_000;
private final RecordingInput input;
@@ -53,7 +53,7 @@
private RecordedEvent[] sortedList;
private boolean ordered;
- public FileEventConsumer(AccessControlContext acc, RecordingInput input) throws IOException {
+ public FileConsumer(AccessControlContext acc, RecordingInput input) throws IOException {
super(acc);
this.input = input;
}
@@ -129,15 +129,17 @@
public void close() {
}
+
+
}
private final RecordingInput input;
- private final FileEventConsumer eventConsumer;
+ private final FileConsumer eventConsumer;
- public EventFileStream(Path path) throws IOException {
+ public EventFileStream(Path path, Instant from, Instant to) throws IOException {
Objects.requireNonNull(path);
input = new RecordingInput(path.toFile());
- eventConsumer = new FileEventConsumer(AccessController.getContext(), input);
+ eventConsumer = new FileConsumer(AccessController.getContext(), input);
}
@Override
@@ -212,4 +214,9 @@
public void setOrdered(boolean ordered) {
eventConsumer.setOrdered(ordered);
}
+
+ @Override
+ public void setStartTime(Instant startTime) {
+ eventConsumer.setStartTime(startTime);
+ }
}
--- a/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventStream.java Thu Jun 27 10:41:01 2019 +0200
+++ b/src/jdk.jfr/share/classes/jdk/jfr/consumer/EventStream.java Wed Jul 03 22:51:44 2019 +0200
@@ -38,49 +38,36 @@
public interface EventStream extends AutoCloseable {
/**
- * Creates a stream starting from the next written event in a disk
- * repository.
+ * Creates a stream from a disk repository.
+ * <p>
+ * By default, the stream will start with the next event that is 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
+ * during reading
*/
public static EventStream openRepository(Path directory) throws IOException {
- return new EventDirectoryStream(AccessController.getContext(), directory);
+ return new EventDirectoryStream(AccessController.getContext(), directory, EventConsumer.NEXT_EVENT);
}
/**
- * Creates an event stream starting from the first event in a file.
+ * Creates an event stream from a file.
+ * <p>
+ * By default, the stream will start 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 a stream can't be opened,or an I/O error occurs
+ * @throws IOException if a stream can't be opened, or an I/O error occurs
* during reading
*/
public static EventStream openFile(Path file) throws IOException {
- return new EventFileStream(file);
- }
-
- /**
- * Creates an event stream starting start time and end time in a file.
- *
- * @param file location of the file, not {@code null}
- *
- * @param the start start time for the stream, or {@code null} to get data
- * from the beginning of the
- *
- * @param the end end time for the stream, or {@code null} to get data until
- * the end.
- *
- * @throws IllegalArgumentException if {@code end} happens before
- * {@code start}
- *
- * @throws IOException if a stream can't be opened,or an I/O error occurs
- * during reading
- */
- public static EventStream openFile(Path file, Instant from, Instant to) throws IOException {
- throw new UnsupportedOperationException("Not yet implemented");
- // return new EventFileStream(file);
+ return new EventFileStream(file, null, null);
}
/**
@@ -95,6 +82,7 @@
* Performs an action on all events in the stream with a specified name.
*
* @param eventName the name of the event, not {@code null}
+ *
* @param action an action to be performed on each {@code RecordedEvent}
* that matches the event name, not {@code null}
*/
@@ -110,6 +98,9 @@
/**
* Performs an action when the event stream is closed.
+ * <p>
+ * If the stream is already closed, the action will be executed immediately
+ * in the current thread.
*
* @param action an action to be performed after the stream has been closed,
* not {@code null}
@@ -128,6 +119,7 @@
* removed.
*
* @param action the action to remove, not {@code null}
+ *
* @return {@code true} if the action was removed, {@code false} otherwise
*
* @see #onClose(Runnable)
@@ -138,35 +130,40 @@
boolean remove(Object action);
/**
- * Hint that the event object in an {@link #onEvent(Consumer)} action
- * may be reused.
+ * Specifies that the event object in an {@link #onEvent(Consumer)} action
+ * is to be reused.
* <p>
* If reuse is set to {@code true), a callback should not keep a reference
* to the event object after the callback from {@code onEvent} has returned.
- * <p>
- * By default reuse is set to {@code true}
*
- * @param resuse if event objects can be reused between calls to {@code #onEvent(Consumer)}
+ * @param resuse if event objects can be reused between calls to
+ * {@code #onEvent(Consumer)}
*
*/
public void setReuse(boolean reuse);
/**
- * Orders events in chronological order aft the end timestamp
+ * Specifies that events arrives in chronological order, sorted by the time
+ * they were committed to the event stream.
*
- * TODO: WHAT ABOUT EVENTS THAT OCCUR WAY OUT OF ORDER
- * <p>
- * By default ordered is set to {@code true}
- *
- * @param ordered if event objects arrive in chronological order to {@code #onEvent(Consumer)}
+ * @param ordered if event objects arrive in chronological order to
+ * {@code #onEvent(Consumer)}
*/
public void setOrdered(boolean ordered);
/**
- * Starts processing events in the stream.
+ * Specifies start time of the event stream.
+ *
+ * @param startTime the start time, not {@code null}
+ *
+ * @throws IllegalStateException if the stream has already been started
+ */
+ public void setStartTime(Instant startTime);
+
+ /**
+ * Start processing events in the stream.
* <p>
- * All actions will performed on this stream will happen in the current
- * thread.
+ * All actions performed on this stream will happen in the current thread.
*
* @throws IllegalStateException if the stream is already started or if it
* has been closed
@@ -174,11 +171,11 @@
void start();
/**
- * Starts processing events in the stream asynchronously.
+ * Start processing events in the stream asynchronously.
* <p>
* All actions on this stream will be performed in a separate thread.
*
- * @throws IllegalStateException if the stream is already started or if it
+ * @throws IllegalStateException if the stream is already started, or if it
* has been closed
*/
void startAsync();
--- a/src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordingStream.java Thu Jun 27 10:41:01 2019 +0200
+++ b/src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordingStream.java Wed Jul 03 22:51:44 2019 +0200
@@ -29,6 +29,7 @@
import java.security.AccessControlContext;
import java.security.AccessController;
import java.time.Duration;
+import java.time.Instant;
import java.util.Map;
import java.util.function.Consumer;
@@ -42,8 +43,8 @@
import jdk.jfr.internal.Utils;
/**
- * An event stream produces events from a file, directory or a running JVM (Java
- * Virtual Machine).
+ * An recording stream produces events from a running JVM (Java Virtual
+ * Machine).
*/
public class RecordingStream implements AutoCloseable, EventStream {
@@ -81,7 +82,7 @@
this.recording = new Recording();
this.recording.setFlushInterval(Duration.ofMillis(1000));
try {
- this.stream = new EventDirectoryStream(acc, null);
+ this.stream = new EventDirectoryStream(acc, null, null);
} catch (IOException ioe) {
throw new IllegalStateException(ioe.getMessage());
}
@@ -156,7 +157,8 @@
* The following example shows how to merge settings.
*
* <pre>
- * {@code
+ * {
+ * @code
* Map<String, String> settings = recording.getSettings();
* settings.putAll(additionalSettings);
* recordingStream.setSettings(settings);
@@ -214,8 +216,9 @@
public EventSettings disable(Class<? extends Event> eventClass) {
return recording.disable(eventClass);
}
+
/**
- * Determines how far back data is kept for the stream if the stream can't
+ * Determines how far back data is kept for the stream, if the stream can't
* keep up.
* <p>
* To control the amount of recording data stored on disk, the maximum
@@ -314,7 +317,7 @@
* Determines how often events are made available for streaming.
*
* @param interval the interval at which events are made available to the
- * stream
+ * stream, no {@code null}
*
* @throws IllegalArgumentException if <code>interval</code> is negative
*
@@ -338,4 +341,9 @@
public void setOrdered(boolean ordered) {
stream.setOrdered(ordered);
}
+
+ @Override
+ public void setStartTime(Instant startTime) {
+ stream.setStartTime(startTime);
+ }
}
--- a/src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/RepositoryFiles.java Thu Jun 27 10:41:01 2019 +0200
+++ b/src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/RepositoryFiles.java Wed Jul 03 22:51:44 2019 +0200
@@ -34,6 +34,8 @@
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
@@ -45,7 +47,7 @@
public final class RepositoryFiles {
private final Path repostory;
- private final SortedMap<Long, Path> pathSet = new TreeMap<>();
+ private final NavigableMap<Long, Path> pathSet = new TreeMap<>();
private final Map<Path, Long> pathLookup = new HashMap<>();
private volatile boolean closed;
@@ -57,8 +59,18 @@
return pathLookup.get(p);
}
+ public Path lastPath() {
+ return nextPath(-1);
+ }
+
public Path nextPath(long startTimeNanos) {
while (!closed) {
+ if (startTimeNanos == -1) {
+ Entry<Long, Path> e = pathSet.lastEntry();
+ if (e != null) {
+ return e.getValue();
+ }
+ }
SortedMap<Long, Path> after = pathSet.tailMap(startTimeNanos);
if (!after.isEmpty()) {
Path path = after.get(after.firstKey());
@@ -147,4 +159,6 @@
pathSet.notify();
}
}
+
+
}
\ No newline at end of file