src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventConsumer.java
author egahlin
Fri, 24 May 2019 19:39:31 +0200
branchJEP-349-branch
changeset 57372 50ca040843ea
child 57373 400db63e4937
permissions -rw-r--r--
Prepare infrastructure for multiple implementations of EventStream
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
57372
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     1
/*
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     2
 * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     3
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     4
 *
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     5
 * This code is free software; you can redistribute it and/or modify it
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     6
 * under the terms of the GNU General Public License version 2 only, as
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     7
 * published by the Free Software Foundation.  Oracle designates this
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     8
 * particular file as subject to the "Classpath" exception as provided
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
     9
 * by Oracle in the LICENSE file that accompanied this code.
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    10
 *
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    11
 * This code is distributed in the hope that it will be useful, but WITHOUT
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    12
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    13
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    14
 * version 2 for more details (a copy is included in the LICENSE file that
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    15
 * accompanied this code).
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    16
 *
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    17
 * You should have received a copy of the GNU General Public License version
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    18
 * 2 along with this work; if not, write to the Free Software Foundation,
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    19
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    20
 *
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    21
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    22
 * or visit www.oracle.com if you need additional information or have any
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    23
 * questions.
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    24
 */
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    25
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    26
package jdk.jfr.internal.consumer;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    27
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    28
import java.io.IOException;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    29
import java.lang.invoke.MethodHandles;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    30
import java.lang.invoke.VarHandle;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    31
import java.security.AccessControlContext;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    32
import java.security.AccessController;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    33
import java.security.PrivilegedAction;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    34
import java.time.Duration;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    35
import java.util.ArrayList;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    36
import java.util.Arrays;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    37
import java.util.List;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    38
import java.util.Objects;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    39
import java.util.function.Consumer;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    40
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    41
import jdk.jfr.EventType;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    42
import jdk.jfr.consumer.LongMap;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    43
import jdk.jfr.consumer.RecordedEvent;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    44
import jdk.jfr.internal.JVM;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    45
import jdk.jfr.internal.LogLevel;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    46
import jdk.jfr.internal.LogTag;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    47
import jdk.jfr.internal.Logger;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    48
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    49
abstract public class EventConsumer implements Runnable {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    50
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    51
    private final static class EventDispatcher {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    52
        public final static EventDispatcher[] NO_DISPATCHERS = new EventDispatcher[0];
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    53
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    54
        final private String eventName;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    55
        final Consumer<RecordedEvent> action;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    56
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    57
        public EventDispatcher(Consumer<RecordedEvent> action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    58
            this(null, action);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    59
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    60
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    61
        public EventDispatcher(String eventName, Consumer<RecordedEvent> action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    62
            this.eventName = eventName;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    63
            this.action = action;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    64
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    65
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    66
        public void offer(RecordedEvent event) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    67
            action.accept(event);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    68
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    69
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    70
        public boolean accepts(EventType eventType) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    71
            return (eventName == null || eventType.getName().equals(eventName));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    72
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    73
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    74
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    75
    private final static JVM jvm = JVM.getJVM();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    76
    private final static VarHandle closedHandle;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    77
    private final static VarHandle consumersHandle;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    78
    private final static VarHandle dispatcherHandle;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    79
    private final static VarHandle flushActionsHandle;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    80
    private final static VarHandle closeActionsHandle;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    81
    static {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    82
        try {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    83
            MethodHandles.Lookup l = MethodHandles.lookup();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    84
            closedHandle = l.findVarHandle(EventConsumer.class, "closed", boolean.class);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    85
            consumersHandle = l.findVarHandle(EventConsumer.class, "consumers", EventDispatcher[].class);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    86
            dispatcherHandle = l.findVarHandle(EventConsumer.class, "dispatcher", LongMap.class);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    87
            flushActionsHandle = l.findVarHandle(EventConsumer.class, "flushActions", Runnable[].class);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    88
            closeActionsHandle = l.findVarHandle(EventConsumer.class, "closeActions", Runnable[].class);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    89
        } catch (ReflectiveOperationException e) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    90
            throw new InternalError(e);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    91
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    92
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    93
    // set by VarHandle
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    94
    private boolean closed;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    95
    // set by VarHandle
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    96
    private EventDispatcher[] consumers = new EventDispatcher[0];
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    97
    // set by VarHandle
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    98
    private LongMap<EventDispatcher[]> dispatcher = new LongMap<>();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
    99
    // set by VarHandle
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   100
    private Runnable[] flushActions = new Runnable[0];
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   101
    // set by VarHandle
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   102
    private Runnable[] closeActions = new Runnable[0];
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   103
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   104
    protected InternalEventFilter eventFilter = InternalEventFilter.ACCEPT_ALL;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   105
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   106
    private final AccessControlContext accessControlContext;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   107
    private boolean started;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   108
    private Thread thread;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   109
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   110
    protected long startNanos;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   111
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   112
    public EventConsumer(AccessControlContext acc) throws IOException {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   113
        this.accessControlContext = acc;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   114
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   115
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   116
    public void run() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   117
        doPriviliged(() -> execute());
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   118
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   119
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   120
    void doPriviliged(Runnable r) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   121
        AccessController.doPrivileged(new PrivilegedAction<Void>() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   122
            @Override
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   123
            public Void run() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   124
                r.run();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   125
                return null;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   126
            }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   127
        }, accessControlContext);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   128
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   129
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   130
    private void execute() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   131
        jvm.exclude(Thread.currentThread());
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   132
        try {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   133
            process();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   134
        } catch (Throwable e) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   135
            e.printStackTrace();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   136
            Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Unexpectedexception iterating consumer.");
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   137
        } finally {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   138
            Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Execution of stream ended.");
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   139
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   140
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   141
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   142
    public abstract void process() throws Exception;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   143
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   144
    public synchronized boolean remove(Object action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   145
        boolean remove = false;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   146
        Runnable[] updatedFlushActions = removeAction(flushActions, action);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   147
        if (updatedFlushActions != null) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   148
            flushActionsHandle.setVolatile(this, updatedFlushActions);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   149
            remove = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   150
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   151
        Runnable[] updatedCloseActions = removeAction(closeActions, action);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   152
        if (updatedCloseActions != null) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   153
            closeActionsHandle.setVolatile(this, updatedCloseActions);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   154
            remove = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   155
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   156
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   157
        boolean removeConsumer = false;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   158
        List<EventDispatcher> list = new ArrayList<>();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   159
        for (int i = 0; i < consumers.length; i++) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   160
            if (consumers[i].action != action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   161
                list.add(consumers[i]);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   162
            } else {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   163
                removeConsumer = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   164
                remove = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   165
            }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   166
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   167
        if (removeConsumer) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   168
            EventDispatcher[] array = list.toArray(new EventDispatcher[list.size()]);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   169
            consumersHandle.setVolatile(this, array);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   170
            dispatcherHandle.setVolatile(this, new LongMap<>()); // will reset
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   171
                                                                 // dispatch
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   172
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   173
        return remove;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   174
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   175
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   176
    public void dispatch(RecordedEvent e) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   177
        EventDispatcher[] consumerDispatch = dispatcher.get(e.getEventType().getId());
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   178
        if (consumerDispatch == null) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   179
            consumerDispatch = EventDispatcher.NO_DISPATCHERS;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   180
            for (EventDispatcher ec : consumers.clone()) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   181
                if (ec.accepts(e.getEventType())) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   182
                    consumerDispatch = merge(consumerDispatch, ec);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   183
                }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   184
            }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   185
            dispatcher.put(e.getEventType().getId(), consumerDispatch);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   186
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   187
        for (int i = 0; i < consumerDispatch.length; i++) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   188
            consumerDispatch[i].offer(e);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   189
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   190
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   191
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   192
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   193
    public void onEvent(Consumer<RecordedEvent> action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   194
        add(new EventDispatcher(action));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   195
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   196
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   197
    public void onEvent(String eventName, Consumer<RecordedEvent> action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   198
        add(new EventDispatcher(eventName, action));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   199
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   200
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   201
    private synchronized void add(EventDispatcher e) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   202
        consumersHandle.setVolatile(this, merge(consumers, e));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   203
        dispatcherHandle.setVolatile(this, new LongMap<>()); // will reset
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   204
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   205
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   206
    public synchronized void onFlush(Runnable action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   207
        flushActionsHandle.setVolatile(this, addAction(flushActions, action));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   208
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   209
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   210
    public synchronized void addCloseAction(Runnable action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   211
        closeActionsHandle.setVolatile(this, addAction(closeActions, action));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   212
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   213
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   214
    public void setClosed(boolean closed) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   215
        closedHandle.setVolatile(this, closed);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   216
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   217
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   218
    final public boolean isClosed() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   219
        return closed;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   220
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   221
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   222
    public void runCloseActions() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   223
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   224
        Runnable[] cas = this.closeActions;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   225
        for (int i = 0; i < cas.length; i++) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   226
            cas[i].run();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   227
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   228
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   229
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   230
    public void runFlushActions() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   231
        Runnable[] fas = this.flushActions;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   232
        for (int i = 0; i < fas.length; i++) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   233
            fas[i].run();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   234
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   235
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   236
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   237
    public synchronized void startAsync(long startNanos) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   238
        if (started) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   239
            throw new IllegalStateException("Event stream can only be started once");
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   240
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   241
        started = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   242
        setStartNanos(startNanos);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   243
        thread = new Thread(this);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   244
        thread.setDaemon(true);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   245
        thread.start();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   246
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   247
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   248
    public void start(long startNanos) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   249
        synchronized (this) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   250
            if (started) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   251
                throw new IllegalStateException("Event stream can only be started once");
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   252
            }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   253
            started = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   254
            setStartNanos(startNanos);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   255
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   256
        run();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   257
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   258
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   259
    public void awaitTermination(Duration timeout) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   260
        Objects.requireNonNull(timeout);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   261
        Thread t = null;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   262
        synchronized (this) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   263
            t = thread;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   264
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   265
        if (t != null && t != Thread.currentThread()) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   266
            try {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   267
                t.join(timeout.toMillis());
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   268
            } catch (InterruptedException e) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   269
                // ignore
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   270
            }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   271
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   272
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   273
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   274
    public void awaitTermination() {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   275
        awaitTermination(Duration.ofMillis(0));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   276
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   277
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   278
    private void setStartNanos(long startNanos) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   279
        this.startNanos = startNanos;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   280
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   281
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   282
    protected static EventDispatcher[] merge(EventDispatcher[] current, EventDispatcher add) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   283
        EventDispatcher[] array = new EventDispatcher[current.length + 1];
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   284
        System.arraycopy(current, 0, array, 0, current.length);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   285
        array[current.length] = add;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   286
        return array;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   287
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   288
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   289
    private static Runnable[] removeAction(Runnable[] array, Object action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   290
        if (array.length == 0) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   291
            return null;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   292
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   293
        boolean remove = false;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   294
        List<Runnable> list = new ArrayList<>();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   295
        for (int i = 0; i < array.length; i++) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   296
            if (array[i] != action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   297
                list.add(array[i]);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   298
            } else {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   299
                remove = true;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   300
            }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   301
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   302
        if (remove) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   303
            return list.toArray(new Runnable[list.size()]);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   304
        }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   305
        return null;
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   306
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   307
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   308
    private static Runnable[] addAction(Runnable[] array, Runnable action) {
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   309
        ArrayList<Runnable> a = new ArrayList<>();
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   310
        a.addAll(Arrays.asList(array));
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   311
        a.add(action);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   312
        return a.toArray(new Runnable[0]);
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   313
    }
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   314
50ca040843ea Prepare infrastructure for multiple implementations of EventStream
egahlin
parents:
diff changeset
   315
}