1 /*

     2  * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.jfr.consumer;

    27 

    28 import java.io.IOException;

    29 import java.security.AccessControlContext;

    30 import java.time.Duration;

    31 import java.time.Instant;

    32 import java.util.Objects;

    33 import java.util.function.Consumer;

    34 

    35 import jdk.jfr.EventType;

    36 

    37 final class EventDirectoryStream implements EventStream {

    38 

    39     public final static class EventConsumer {

    40         final private String eventName;

    41         final Consumer<RecordedEvent> action;

    42 

    43         EventConsumer(String eventName, Consumer<RecordedEvent> eventConsumer) {

    44             this.eventName = eventName;

    45             this.action = eventConsumer;

    46         }

    47 

    48         public void offer(RecordedEvent event) {

    49             action.accept(event);

    50         }

    51 

    52         public boolean accepts(EventType eventType) {

    53             return (eventName == null || eventType.getName().equals(eventName));

    54         }

    55     }

    56 

    57     private final EventRunner eventRunner;

    58     private Thread thread;

    59     private boolean started;

    60 

    61     public EventDirectoryStream(AccessControlContext acc) throws IOException {

    62         eventRunner = new EventRunner(acc);

    63     }

    64 

    65     public void close() {

    66         synchronized (eventRunner) {

    67             eventRunner.close();

    68         }

    69     }

    70 

    71     public synchronized void onFlush(Runnable action) {

    72         Objects.requireNonNull(action);

    73         synchronized (eventRunner) {

    74             this.eventRunner.addFlush(action);

    75         }

    76     }

    77 

    78     void start(long startNanos) {

    79         synchronized (eventRunner) {

    80             if (started) {

    81                 throw new IllegalStateException("Event stream can only be started once");

    82             }

    83             started = true;

    84             eventRunner.setStartNanos(startNanos);

    85         }

    86         eventRunner.run();

    87     }

    88 

    89     @Override

    90     public void start() {

    91         start(Instant.now().toEpochMilli() * 1000*1000L);

    92     }

    93 

    94     @Override

    95     public void startAsync() {

    96         startAsync(Instant.now().toEpochMilli() * 1000*1000L);

    97     }

    98 

    99     void startAsync(long startNanos) {

   100         synchronized (eventRunner) {

   101             eventRunner.setStartNanos(startNanos);

   102             thread = new Thread(eventRunner);

   103             thread.setDaemon(true);

   104             thread.start();

   105         }

   106     }

   107 

   108     public void addEventConsumer(EventConsumer action) {

   109         Objects.requireNonNull(action);

   110         synchronized (eventRunner) {

   111             eventRunner.add(action);

   112         }

   113     }

   114 

   115 

   116 

   117     @Override

   118     public void onEvent(Consumer<RecordedEvent> action) {

   119         Objects.requireNonNull(action);

   120         synchronized (eventRunner) {

   121             eventRunner.add(new EventConsumer(null, action));

   122         }

   123     }

   124 

   125     @Override

   126     public void onEvent(String eventName, Consumer<RecordedEvent> action) {

   127         Objects.requireNonNull(eventName);

   128         Objects.requireNonNull(action);

   129         synchronized (eventRunner) {

   130             eventRunner.add(new EventConsumer(eventName, action));

   131         }

   132     }

   133 

   134     @Override

   135     public void onClose(Runnable action) {

   136         Objects.requireNonNull(action);

   137         synchronized (eventRunner) {

   138             eventRunner.addCloseAction(action);

   139         }

   140     }

   141 

   142     @Override

   143     public boolean remove(Object action) {

   144         Objects.requireNonNull(action);

   145         synchronized (eventRunner) {

   146             return eventRunner.remove(action);

   147         }

   148     }

   149 

   150     @Override

   151     public void awaitTermination(Duration timeout) {

   152         Objects.requireNonNull(timeout);

   153         Thread t = null;

   154         synchronized (eventRunner) {

   155             t = thread;

   156         }

   157         if (t != null && t != Thread.currentThread()) {

   158             try {

   159                 t.join(timeout.toMillis());

   160             } catch (InterruptedException e) {

   161                 // ignore

   162             }

   163         }

   164     }

   165 

   166     @Override

   167     public void awaitTermination() {

   168         awaitTermination(Duration.ofMillis(0));

   169     }

   170 

   171 

   172 }