jdk-sandbox: comparison src/jdk.jfr/share/classes/jdk/jfr/consumer/AbstractEventStream.java

equal deleted inserted replaced

-:74a38c0b5054
+:aa7b1ea52413
 public static final class StreamConfiguration {
 private static final Runnable[] NO_ACTIONS = new Runnable[0];
 private Runnable[] flushActions = NO_ACTIONS;
 private Runnable[] closeActions = NO_ACTIONS;
+private Runnable[] errorActions = NO_ACTIONS;
 private EventDispatcher[] dispatchers = NO_DISPATCHERS;
 private InternalEventFilter eventFilter = InternalEventFilter.ACCEPT_ALL;
+private LongMap<EventDispatcher[]> dispatcherLookup = new LongMap<>();
+private boolean changedConfiguration = false;
 private boolean closed = false;
 private boolean reuse = true;
 private boolean ordered = true;
 private Instant startTime = null;
 private Instant endTime = null;
 private boolean started = false;
 private long startNanos = 0;
 private long endNanos = Long.MAX_VALUE;
-private LongMap<EventDispatcher[]> dispatcherLookup = new LongMap<>();
-private boolean changed = false;
 public StreamConfiguration(StreamConfiguration configuration) {
 this.flushActions = configuration.flushActions;
 this.closeActions = configuration.closeActions;
 this.dispatchers = configuration.dispatchers;
 final public StreamConfiguration addCloseAction(Runnable action) {
 closeActions = add(closeActions, action);
 return this;
 }
+public StreamConfiguration addErrorAction(Runnable action) {
+errorActions = add(errorActions, action);
+return this;
+}
 final public StreamConfiguration setClosed(boolean closed) {
 this.closed = closed;
-changed = true;
+changedConfiguration = true;
 return this;
 }
 final public boolean isClosed() {
 return closed;
 }
 EventDispatcher[] result = list.toArray(new EventDispatcher[0]);
 if (modified) {
 eventFilter = buildFilter(result);
 dispatcherLookup = new LongMap<>();
-changed = true;
+changedConfiguration = true;
 }
 return result;
 }
 private <T> T[] remove(T[] array, Object action) {
 List<T> list = new ArrayList<>(array.length);
 for (int i = 0; i < array.length; i++) {
 if (array[i] != action) {
 list.add(array[i]);
 } else {
-changed = true;
+changedConfiguration = true;
 }
 }
 return list.toArray(array);
 }
 private <T> T[] add(T[] array, T object) {
 List<T> list = new ArrayList<>(Arrays.asList(array));
 list.add(object);
-changed = true;
+changedConfiguration = true;
 return list.toArray(array);
 }
 private static InternalEventFilter buildFilter(EventDispatcher[] dispatchers) {
 InternalEventFilter ef = new InternalEventFilter();
 return ef;
 }
 final public StreamConfiguration setReuse(boolean reuse) {
 this.reuse = reuse;
-changed = true;
+changedConfiguration = true;
 return this;
 }
 final public StreamConfiguration setOrdered(boolean ordered) {
 this.ordered = ordered;
-changed = true;
+changedConfiguration = true;
 return this;
 }
 public StreamConfiguration setEndTime(Instant endTime) {
 this.endTime = endTime;
 this.endNanos = Utils.timeToNanos(endTime);
-changed = true;
+changedConfiguration = true;
 return this;
 }
 final public StreamConfiguration setStartTime(Instant startTime) {
 this.startTime = startTime;
 this.startNanos = Utils.timeToNanos(startTime);
-changed = true;
+changedConfiguration = true;
 return this;
 }
 final public Instant getStartTime() {
 return startTime;
 return started;
 }
 final public StreamConfiguration setStartNanos(long startNanos) {
 this.startNanos = startNanos;
-changed = true;
+changedConfiguration = true;
 return this;
 }
 final public void setStarted(boolean started) {
 this.started = started;
-changed = true;
+changedConfiguration = true;
 }
 final public boolean hasChanged() {
-return changed;
+return changedConfiguration;
 }
 final public boolean getReuse() {
 return reuse;
 }
 }
 private EventDispatcher[] getDispatchers() {
 return dispatchers;
 }
 }
 final static class EventDispatcher {
 final static EventDispatcher[] NO_DISPATCHERS = new EventDispatcher[0];
 private final AccessControlContext accessControlContext;
 private final Thread thread;
 private final boolean active;
 protected final Runnable flushOperation = () -> runFlushActions();
-// Updated by updateConfiguration()
+// Modified by updateConfiguration()
 protected volatile StreamConfiguration configuration = new StreamConfiguration();
 // Cache the last event type and dispatch.
 private EventType lastEventType;
 private EventDispatcher[] lastEventDispatch;
 public Void run() {
 execute();
 return null;
 }
 }, accessControlContext);
 }
 private void execute() {
 JVM.getJVM().exclude(Thread.currentThread());
 try {
 process();
-} catch (IOException e) {
-if (!isClosed()) {
-logException(e);
-}
 } catch (Exception e) {
-logException(e);
+defaultErrorHandler(e);
 } finally {
 Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Execution of stream ended.");
 }
 }
-private void logException(Exception e) {
+public abstract void process() throws Exception;
-// FIXME: e.printStackTrace(); for debugging purposes,
-// remove before before integration
-e.printStackTrace();
-Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Unexpected error processing stream. " + e.getMessage());
-}
-public abstract void process() throws IOException;
 protected final void clearLastDispatch() {
 lastEventDispatch = null;
 lastEventType = null;
 }
 }
 for (int i = 0; i < ret.length; i++) {
 try {
 ret[i].offer(event);
 } catch (Exception e) {
-logException(e);
+handleError(e);
 }
 }
+}
+protected final void handleError(Throwable e) {
+StreamConfiguration c = configuration;
+if (c.errorActions.length == 0) {
+defaultErrorHandler(e);
+return;
+}
+for (Runnable r : c.errorActions) {
+r.run();
+}
+}
+protected final void defaultErrorHandler(Throwable e) {
+e.printStackTrace();
 }
 public final void runCloseActions() {
 Runnable[] cas = configuration.getCloseActions();
 for (int i = 0; i < cas.length; i++) {
 try {
 cas[i].run();
 } catch (Exception e) {
-logException(e);
+handleError(e);
 }
 }
 }
 public final void runFlushActions() {
 Runnable[] fas = configuration.getFlushActions();
 for (int i = 0; i < fas.length; i++) {
 try {
 fas[i].run();
 } catch (Exception e) {
-logException(e);
+handleError(e);
 }
 }
 }
 // Purpose of synchronizing the following methods is
 // to serialize changes to the configuration, so only one
 // thread at a time can change the configuration.
 public final synchronized void addCloseAction(Runnable action) {
 updateConfiguration(new StreamConfiguration(configuration).addCloseAction(action));
 }
+public final synchronized void addErrorAction(Runnable action) {
+updateConfiguration(new StreamConfiguration(configuration).addErrorAction(action));
+}
 public final synchronized void setClosed(boolean closed) {
 updateConfiguration(new StreamConfiguration(configuration).setClosed(closed));
 }
 public final synchronized void setReuse(boolean reuse) {
 startTime = Instant.EPOCH;
 }
 updateConfiguration(new StreamConfiguration(configuration).setStartTime(startTime));
 }
-public final void setEndTime(Instant endTime) {
+public final synchronized void setEndTime(Instant endTime) {
 if (configuration.isStarted()) {
 throw new IllegalStateException("Stream is already started");
 }
 updateConfiguration(new StreamConfiguration(configuration).setEndTime(endTime));
 }
 protected boolean updateConfiguration(StreamConfiguration newConfiguration) {
-// Changes to the configuration must happen one at a time, so make
+if (!Thread.holdsLock(this)) {
-// sure that we have the monitor
+throw new InternalError("Modification of configuration without proper lock");
-Thread.holdsLock(this);
+}
 if (newConfiguration.hasChanged()) {
 // Publish objects held by configuration object
 VarHandle.releaseFence();
 configuration = newConfiguration;
 return true;
 public final void start(long startNanos) {
 startInternal(startNanos);
 run();
 }
-private void startInternal(long startNanos) {
+private synchronized void startInternal(long startNanos) {
-synchronized (this) {
+if (configuration.isStarted()) {
-if (configuration.isStarted()) {
+throw new IllegalStateException("Event stream can only be started once");
-throw new IllegalStateException("Event stream can only be started once");
+}
-}
+StreamConfiguration c = new StreamConfiguration(configuration);
-StreamConfiguration c = new StreamConfiguration(configuration);
+if (active) {
-if (active) {
+c.setStartNanos(startNanos);
-c.setStartNanos(startNanos);
+}
-}
+c.setStarted(true);
-c.setStarted(true);
+updateConfiguration(c);
-updateConfiguration(c);
-}
 }
 public final void awaitTermination(Duration timeout) {
 Objects.requireNonNull(timeout);
 if (thread != Thread.currentThread()) {

branch	JEP-349-branch
changeset 57971	aa7b1ea52413
parent 57948	59ad17a19e24
child 57985	be121cbf3284