stash eventbus changes

Author: CRZbulabula

iotdb-core/confignode/pom.xml

@@ -116,6 +116,10 @@
             <groupId>commons-io</groupId>
             <artifactId>commons-io</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.timecho.ratis</groupId>
             <artifactId>ratis-common</artifactId>

iotdb-core/node-commons/src/main/java/com/google/common/base/Objects.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.google.common.base;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import javax.annotation.CheckForNull;
+
+import java.util.Arrays;
+
+/**
+ * Helper functions that can operate on any {@code Object}.
+ *
+ * <p>See the Guava User Guide on <a
+ * href="https://github.com/google/guava/wiki/CommonObjectUtilitiesExplained">writing {@code Object}
+ * methods with {@code Objects}</a>.
+ *
+ * @author Laurence Gonsalves
+ * @since 2.0
+ */
+@ElementTypesAreNonnullByDefault
+public final class Objects extends ExtraObjectsMethodsForWeb {
+  private Objects() {}
+
+  /**
+   * Determines whether two possibly-null objects are equal. Returns:
+   *
+   * <ul>
+   *   <li>{@code true} if {@code a} and {@code b} are both null.
+   *   <li>{@code true} if {@code a} and {@code b} are both non-null and they are equal according to
+   *       {@link Object#equals(Object)}.
+   *   <li>{@code false} in all other situations.
+   * </ul>
+   *
+   * <p>This assumes that any non-null objects passed to this function conform to the {@code
+   * equals()} contract.
+   *
+   * <p><b>Note for Java 7 and later:</b> This method should be treated as deprecated; use {@link
+   * java.util.Objects#equals} instead.
+   */
+  public static boolean equal(@CheckForNull Object a, @CheckForNull Object b) {
+    return a == b || (a != null && a.equals(b));
+  }
+
+  /**
+   * Generates a hash code for multiple values. The hash code is generated by calling {@link
+   * Arrays#hashCode(Object[])}. Note that array arguments to this method, with the exception of a
+   * single Object array, do not get any special handling; their hash codes are based on identity
+   * and not contents.
+   *
+   * <p>This is useful for implementing {@link Object#hashCode()}. For example, in an object that
+   * has three properties, {@code x}, {@code y}, and {@code z}, one could write:
+   *
+   * <pre>{@code
+   * public int hashCode() {
+   *   return Objects.hashCode(getX(), getY(), getZ());
+   * }
+   * }</pre>
+   *
+   * <p><b>Warning:</b> When a single object is supplied, the returned hash code does not equal the
+   * hash code of that object.
+   *
+   * <p><b>Note for Java 7 and later:</b> This method should be treated as deprecated; use {@link
+   * java.util.Objects#hash} instead.
+   */
+  public static int hashCode(@CheckForNull @Nullable Object... objects) {
+    return Arrays.hashCode(objects);
+  }
+}

iotdb-core/node-commons/src/main/java/com/google/common/eventbus/DeadEvent.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.google.common.eventbus;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Wraps an event that was posted, but which had no subscribers and thus could not be delivered.
+ *
+ * <p>Registering a DeadEvent subscriber is useful for debugging or logging, as it can detect
+ * misconfigurations in a system's event distribution.
+ *
+ * @author Cliff Biffle
+ * @since 10.0
+ */
+@ElementTypesAreNonnullByDefault
+public class DeadEvent {
+
+  private final Object source;
+  private final Object event;
+
+  /**
+   * Creates a new DeadEvent.
+   *
+   * @param source object broadcasting the DeadEvent (generally the {@link EventBus}).
+   * @param event the event that could not be delivered.
+   */
+  public DeadEvent(Object source, Object event) {
+    this.source = checkNotNull(source);
+    this.event = checkNotNull(event);
+  }
+
+  /**
+   * Returns the object that originated this event (<em>not</em> the object that originated the
+   * wrapped event). This is generally an {@link EventBus}.
+   *
+   * @return the source of this event.
+   */
+  public Object getSource() {
+    return source;
+  }
+
+  /**
+   * Returns the wrapped, 'dead' event, which the system was unable to deliver to any registered
+   * subscriber.
+   *
+   * @return the 'dead' event that could not be delivered.
+   */
+  public Object getEvent() {
+    return event;
+  }
+
+  @Override
+  public String toString() {
+    return "DeadEvent{" + "source=" + source + ", event=" + event + '}';
+  }
+}

iotdb-core/node-commons/src/main/java/com/google/common/eventbus/Dispatcher.java

@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.google.common.eventbus;
+
+import com.google.common.collect.Queues;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Handler for dispatching events to subscribers, providing different event ordering guarantees that
+ * make sense for different situations.
+ *
+ * <p><b>Note:</b> The dispatcher is orthogonal to the subscriber's {@code Executor}. The dispatcher
+ * controls the order in which events are dispatched, while the executor controls how (i.e. on which
+ * thread) the subscriber is actually called when an event is dispatched to it.
+ *
+ * @author Colin Decker
+ */
+@ElementTypesAreNonnullByDefault
+abstract class Dispatcher {
+
+  /**
+   * Returns a dispatcher that queues events that are posted reentrantly on a thread that is already
+   * dispatching an event, guaranteeing that all events posted on a single thread are dispatched to
+   * all subscribers in the order they are posted.
+   *
+   * <p>When all subscribers are dispatched to using a <i>direct</i> executor (which dispatches on
+   * the same thread that posts the event), this yields a breadth-first dispatch order on each
+   * thread. That is, all subscribers to a single event A will be called before any subscribers to
+   * any events B and C that are posted to the event bus by the subscribers to A.
+   */
+  static Dispatcher perThreadDispatchQueue() {
+    return new PerThreadQueuedDispatcher();
+  }
+
+  /**
+   * Returns a dispatcher that queues events that are posted in a single global queue. This behavior
+   * matches the original behavior of AsyncEventBus exactly, but is otherwise not especially useful.
+   * For async dispatch, an {@linkplain #immediate() immediate} dispatcher should generally be
+   * preferable.
+   */
+  static Dispatcher legacyAsync() {
+    return new LegacyAsyncDispatcher();
+  }
+
+  /**
+   * Returns a dispatcher that dispatches events to subscribers immediately as they're posted
+   * without using an intermediate queue to change the dispatch order. This is effectively a
+   * depth-first dispatch order, vs. breadth-first when using a queue.
+   */
+  static Dispatcher immediate() {
+    return ImmediateDispatcher.INSTANCE;
+  }
+
+  /** Dispatches the given {@code event} to the given {@code subscribers}. */
+  abstract void dispatch(Object event, Iterator<Subscriber> subscribers);
+
+  /** Implementation of a {@link #perThreadDispatchQueue()} dispatcher. */
+  private static final class PerThreadQueuedDispatcher extends Dispatcher {
+
+    // This dispatcher matches the original dispatch behavior of EventBus.
+
+    /** Per-thread queue of events to dispatch. */
+    private final ThreadLocal<Queue<Event>> queue =
+        new ThreadLocal<Queue<Event>>() {
+          @Override
+          protected Queue<Event> initialValue() {
+            return Queues.newArrayDeque();
+          }
+        };
+
+    /** Per-thread dispatch state, used to avoid reentrant event dispatching. */
+    private final ThreadLocal<Boolean> dispatching =
+        new ThreadLocal<Boolean>() {
+          @Override
+          protected Boolean initialValue() {
+            return false;
+          }
+        };
+
+    @Override
+    void dispatch(Object event, Iterator<Subscriber> subscribers) {
+      checkNotNull(event);
+      checkNotNull(subscribers);
+      Queue<Event> queueForThread = queue.get();
+      queueForThread.offer(new Event(event, subscribers));
+
+      if (!dispatching.get()) {
+        dispatching.set(true);
+        try {
+          Event nextEvent;
+          while ((nextEvent = queueForThread.poll()) != null) {
+            while (nextEvent.subscribers.hasNext()) {
+              nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
+            }
+          }
+        } finally {
+          dispatching.remove();
+          queue.remove();
+        }
+      }
+    }
+
+    private static final class Event {
+      private final Object event;
+      private final Iterator<Subscriber> subscribers;
+
+      private Event(Object event, Iterator<Subscriber> subscribers) {
+        this.event = event;
+        this.subscribers = subscribers;
+      }
+    }
+  }
+
+  /** Implementation of a {@link #legacyAsync()} dispatcher. */
+  private static final class LegacyAsyncDispatcher extends Dispatcher {
+
+    // This dispatcher matches the original dispatch behavior of AsyncEventBus.
+    //
+    // We can't really make any guarantees about the overall dispatch order for this dispatcher in
+    // a multithreaded environment for a couple of reasons:
+    //
+    // 1. Subscribers to events posted on different threads can be interleaved with each other
+    //    freely. (A event on one thread, B event on another could yield any of
+    //    [a1, a2, a3, b1, b2], [a1, b2, a2, a3, b2], [a1, b2, b3, a2, a3], etc.)
+    // 2. It's possible for subscribers to actually be dispatched to in a different order than they
+    //    were added to the queue. It's easily possible for one thread to take the head of the
+    //    queue, immediately followed by another thread taking the next element in the queue. That
+    //    second thread can then dispatch to the subscriber it took before the first thread does.
+    //
+    // All this makes me really wonder if there's any value in queueing here at all. A dispatcher
+    // that simply loops through the subscribers and dispatches the event to each would actually
+    // probably provide a stronger order guarantee, though that order would obviously be different
+    // in some cases.
+
+    /** Global event queue. */
+    private final ConcurrentLinkedQueue<EventWithSubscriber> queue =
+        Queues.newConcurrentLinkedQueue();
+
+    @Override
+    void dispatch(Object event, Iterator<Subscriber> subscribers) {
+      checkNotNull(event);
+      while (subscribers.hasNext()) {
+        queue.add(new EventWithSubscriber(event, subscribers.next()));
+      }
+
+      EventWithSubscriber e;
+      while ((e = queue.poll()) != null) {
+        e.subscriber.dispatchEvent(e.event);
+      }
+    }
+
+    private static final class EventWithSubscriber {
+      private final Object event;
+      private final Subscriber subscriber;
+
+      private EventWithSubscriber(Object event, Subscriber subscriber) {
+        this.event = event;
+        this.subscriber = subscriber;
+      }
+    }
+  }
+
+  /** Implementation of {@link #immediate()}. */
+  private static final class ImmediateDispatcher extends Dispatcher {
+    private static final ImmediateDispatcher INSTANCE = new ImmediateDispatcher();
+
+    @Override
+    void dispatch(Object event, Iterator<Subscriber> subscribers) {
+      checkNotNull(event);
+      while (subscribers.hasNext()) {
+        subscribers.next().dispatchEvent(event);
+      }
+    }
+  }
+}