[grid] Wire EventBus into Distributor, SessionMap, and LocalNode

java/server/src/org/openqa/selenium/events/zeromq/BUCK

@@ -1,14 +1,17 @@
 java_library(
     name = "zeromq",
     srcs = glob(["*.java"]),
+    exported_deps = [
+        "//java/server/src/org/openqa/selenium/events:events",
+        "//third_party/java/zeromq:jeromq",
+    ],
     deps = [
         "//java/client/src/org/openqa/selenium/remote:remote",
         "//java/server/src/org/openqa/selenium/concurrent:concurrent",
-        "//java/server/src/org/openqa/selenium/events:events",
         "//third_party/java/guava:guava",
-        "//third_party/java/zeromq:jeromq",
     ],
     visibility = [
-        "//java/server/test/org/openqa/selenium/events/...",
+        "//java/server/src/org/openqa/selenium/grid/server:server",
+        "//java/server/test/org/openqa/selenium/...",
     ],
 )

java/server/src/org/openqa/selenium/grid/commands/Hub.java

@@ -23,6 +23,7 @@
 import com.beust.jcommander.ParameterException;
 
 import org.openqa.selenium.cli.CliCommand;
+import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.config.AnnotatedConfig;
 import org.openqa.selenium.grid.config.CompoundConfig;
 import org.openqa.selenium.grid.config.ConcatenatingConfig;
@@ -35,6 +36,7 @@
 import org.openqa.selenium.grid.server.BaseServer;
 import org.openqa.selenium.grid.server.BaseServerFlags;
 import org.openqa.selenium.grid.server.BaseServerOptions;
+import org.openqa.selenium.grid.server.EventBusConfig;
 import org.openqa.selenium.grid.server.HelpFlags;
 import org.openqa.selenium.grid.log.LoggingOptions;
 import org.openqa.selenium.grid.server.Server;
@@ -93,12 +95,22 @@ public Executable configure(String... args) {
 
       LoggingOptions loggingOptions = new LoggingOptions(config);
       loggingOptions.configureLogging();
+
       DistributedTracer tracer = loggingOptions.getTracer();
       GlobalDistributedTracer.setInstance(tracer);
 
-      SessionMap sessions = new LocalSessionMap(tracer);
-      Distributor distributor = new LocalDistributor(tracer, HttpClient.Factory.createDefault());
-      Router router = new Router(tracer, sessions, distributor);
+      EventBusConfig events = new EventBusConfig(config);
+      EventBus bus = events.getEventBus();
+
+      SessionMap sessions = new LocalSessionMap(tracer, bus);
+
+      HttpClient.Factory clientFactory = HttpClient.Factory.createDefault();
+
+      Distributor distributor = new LocalDistributor(
+          tracer,
+          bus,
+          clientFactory);
+      Router router = new Router(tracer, clientFactory, sessions, distributor);
 
       Server<?> server = new BaseServer<>(
           new BaseServerOptions(config));

java/server/src/org/openqa/selenium/grid/commands/Standalone.java

@@ -24,6 +24,7 @@
 
 import org.openqa.selenium.WebDriverException;
 import org.openqa.selenium.cli.CliCommand;
+import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.config.AnnotatedConfig;
 import org.openqa.selenium.grid.config.CompoundConfig;
 import org.openqa.selenium.grid.config.ConcatenatingConfig;
@@ -38,6 +39,8 @@
 import org.openqa.selenium.grid.server.BaseServer;
 import org.openqa.selenium.grid.server.BaseServerFlags;
 import org.openqa.selenium.grid.server.BaseServerOptions;
+import org.openqa.selenium.grid.server.EventBusConfig;
+import org.openqa.selenium.grid.server.EventBusFlags;
 import org.openqa.selenium.grid.server.HelpFlags;
 import org.openqa.selenium.grid.server.Server;
 import org.openqa.selenium.grid.server.W3CCommandHandler;
@@ -70,12 +73,14 @@ public String getDescription() {
   public Executable configure(String... args) {
     HelpFlags help = new HelpFlags();
     BaseServerFlags baseFlags = new BaseServerFlags(4444);
+    EventBusFlags eventFlags = new EventBusFlags(4443);
     NodeFlags nodeFlags = new NodeFlags();
 
     JCommander commander = JCommander.newBuilder()
         .programName("standalone")
         .addObject(baseFlags)
         .addObject(help)
+        .addObject(eventFlags)
         .addObject(nodeFlags)
         .build();
 
@@ -96,7 +101,8 @@ public Executable configure(String... args) {
           new EnvConfig(),
           new ConcatenatingConfig("selenium", '.', System.getProperties()),
           new AnnotatedConfig(help),
-          new AnnotatedConfig(baseFlags));
+          new AnnotatedConfig(baseFlags),
+          new AnnotatedConfig(eventFlags));
 
       LoggingOptions loggingOptions = new LoggingOptions(config);
       loggingOptions.configureLogging();
@@ -106,11 +112,14 @@ public Executable configure(String... args) {
       DistributedTracer tracer = loggingOptions.getTracer();
       GlobalDistributedTracer.setInstance(tracer);
 
-      HttpClient.Factory httpClientFactory = HttpClient.Factory.createDefault();
+      EventBusConfig events = new EventBusConfig(config);
+      EventBus bus = events.getEventBus();
 
-      SessionMap sessions = new LocalSessionMap(tracer);
-      Distributor distributor = new LocalDistributor(tracer, httpClientFactory);
-      Router router = new Router(tracer, sessions, distributor);
+      HttpClient.Factory clientFactory = HttpClient.Factory.createDefault();
+
+      SessionMap sessions = new LocalSessionMap(tracer, bus);
+      Distributor distributor = new LocalDistributor(tracer, bus, clientFactory);
+      Router router = new Router(tracer, clientFactory, sessions, distributor);
 
       String hostName;
       try {
@@ -128,9 +137,14 @@ public Executable configure(String... args) {
         throw new RuntimeException(e);
       }
 
-      LocalNode.Builder node = LocalNode.builder(tracer, httpClientFactory, localhost, sessions)
+      LocalNode.Builder node = LocalNode.builder(
+          tracer,
+          bus,
+          clientFactory,
+          localhost,
+          sessions)
           .maximumConcurrentSessions(Runtime.getRuntime().availableProcessors() * 3);
-      nodeFlags.configure(httpClientFactory, node);
+      nodeFlags.configure(clientFactory, node);
 
       distributor.add(node.build());
 

java/server/src/org/openqa/selenium/grid/distributor/httpd/DistributorServer.java

@@ -23,6 +23,7 @@
 import com.beust.jcommander.ParameterException;
 
 import org.openqa.selenium.cli.CliCommand;
+import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.config.AnnotatedConfig;
 import org.openqa.selenium.grid.config.CompoundConfig;
 import org.openqa.selenium.grid.config.ConcatenatingConfig;
@@ -34,6 +35,7 @@
 import org.openqa.selenium.grid.server.BaseServer;
 import org.openqa.selenium.grid.server.BaseServerFlags;
 import org.openqa.selenium.grid.server.BaseServerOptions;
+import org.openqa.selenium.grid.server.EventBusConfig;
 import org.openqa.selenium.grid.server.HelpFlags;
 import org.openqa.selenium.grid.server.Server;
 import org.openqa.selenium.grid.server.W3CCommandHandler;
@@ -89,10 +91,17 @@ public Executable configure(String... args) {
 
       LoggingOptions loggingOptions = new LoggingOptions(config);
       loggingOptions.configureLogging();
+
       DistributedTracer tracer = loggingOptions.getTracer();
       GlobalDistributedTracer.setInstance(tracer);
 
-      Distributor distributor = new LocalDistributor(tracer, HttpClient.Factory.createDefault());
+      EventBusConfig events = new EventBusConfig(config);
+      EventBus bus = events.getEventBus();
+
+      Distributor distributor = new LocalDistributor(
+          tracer,
+          bus,
+          HttpClient.Factory.createDefault());
 
       BaseServerOptions serverOptions = new BaseServerOptions(config);
 

java/server/src/org/openqa/selenium/grid/distributor/local/BUCK

@@ -8,6 +8,7 @@ java_library(
     deps = [
         "//java/client/src/org/openqa/selenium/remote/tracing:tracing",
         "//java/server/src/org/openqa/selenium/concurrent:concurrent",
+        "//java/server/src/org/openqa/selenium/events:events",
         "//java/server/src/org/openqa/selenium/grid/web:web",
         "//java/server/src/org/openqa/selenium/grid/node:node",
         "//java/server/src/org/openqa/selenium/grid/node/remote:remote",

java/server/src/org/openqa/selenium/grid/distributor/local/LocalDistributor.java

@@ -27,6 +27,7 @@
 import org.openqa.selenium.Capabilities;
 import org.openqa.selenium.SessionNotCreatedException;
 import org.openqa.selenium.concurrent.Regularly;
+import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.data.Session;
 import org.openqa.selenium.grid.distributor.Distributor;
 import org.openqa.selenium.grid.distributor.DistributorStatus;
@@ -64,12 +65,17 @@ public class LocalDistributor extends Distributor {
   private final ReadWriteLock lock = new ReentrantReadWriteLock(/* fair */ true);
   private final Set<Host> hosts = new HashSet<>();
   private final DistributedTracer tracer;
+  private final EventBus bus;
   private final Regularly hostChecker = new Regularly("distributor host checker");
   private final Map<UUID, Collection<Runnable>> allChecks = new ConcurrentHashMap<>();
 
-  public LocalDistributor(DistributedTracer tracer, HttpClient.Factory httpClientFactory) {
+  public LocalDistributor(
+      DistributedTracer tracer,
+      EventBus bus,
+      HttpClient.Factory httpClientFactory) {
     super(tracer, httpClientFactory);
     this.tracer = Objects.requireNonNull(tracer);
+    this.bus = Objects.requireNonNull(bus);
   }
 
   @Override

java/server/src/org/openqa/selenium/grid/node/httpd/NodeServer.java

@@ -24,6 +24,7 @@
 
 import org.openqa.selenium.cli.CliCommand;
 import org.openqa.selenium.concurrent.Regularly;
+import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.config.AnnotatedConfig;
 import org.openqa.selenium.grid.config.CompoundConfig;
 import org.openqa.selenium.grid.config.ConcatenatingConfig;
@@ -38,6 +39,7 @@
 import org.openqa.selenium.grid.server.BaseServer;
 import org.openqa.selenium.grid.server.BaseServerFlags;
 import org.openqa.selenium.grid.server.BaseServerOptions;
+import org.openqa.selenium.grid.server.EventBusConfig;
 import org.openqa.selenium.grid.server.HelpFlags;
 import org.openqa.selenium.grid.server.Server;
 import org.openqa.selenium.grid.server.W3CCommandHandler;
@@ -105,8 +107,13 @@ public Executable configure(String... args) {
 
       LoggingOptions loggingOptions = new LoggingOptions(config);
       loggingOptions.configureLogging();
+
       DistributedTracer tracer = loggingOptions.getTracer();
       GlobalDistributedTracer.setInstance(tracer);
+
+      EventBusConfig events = new EventBusConfig(config);
+      EventBus bus = events.getEventBus();
+
       HttpClient.Factory httpClientFactory = HttpClient.Factory.createDefault();
 
       SessionMapOptions sessionsOptions = new SessionMapOptions(config);
@@ -117,6 +124,7 @@ public Executable configure(String... args) {
 
       LocalNode.Builder builder = LocalNode.builder(
           tracer,
+          bus,
           httpClientFactory,
           serverOptions.getExternalUri(),
           sessions);

java/server/src/org/openqa/selenium/grid/node/local/BUCK

@@ -5,6 +5,7 @@ java_library(
     deps = [
         "//java/client/src/org/openqa/selenium/remote/tracing:tracing",
         "//java/server/src/org/openqa/selenium/concurrent:concurrent",
+        "//java/server/src/org/openqa/selenium/events:events",
         "//java/server/src/org/openqa/selenium/grid/config:config",
         "//java/server/src/org/openqa/selenium/grid/node:node",
         "//java/server/src/org/openqa/selenium/grid/sessionmap:sessionmap",

java/server/src/org/openqa/selenium/grid/node/local/LocalNode.java

@@ -30,6 +30,7 @@
 import org.openqa.selenium.NoSuchSessionException;
 import org.openqa.selenium.UnsupportedCommandException;
 import org.openqa.selenium.concurrent.Regularly;
+import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.grid.component.HealthCheck;
 import org.openqa.selenium.grid.data.Session;
 import org.openqa.selenium.grid.node.Node;
@@ -58,6 +59,7 @@
 
 public class LocalNode extends Node {
 
+  private final EventBus bus;
   private final URI externalUri;
   private final HealthCheck healthCheck;
   private final int maxSessionCount;
@@ -67,6 +69,7 @@ public class LocalNode extends Node {
 
   private LocalNode(
       DistributedTracer tracer,
+      EventBus bus,
       URI uri,
       HealthCheck healthCheck,
       int maxSessionCount,
@@ -79,6 +82,7 @@ private LocalNode(
         maxSessionCount > 0,
         "Only a positive number of sessions can be run: " + maxSessionCount);
 
+    this.bus = Objects.requireNonNull(bus);
     this.externalUri = Objects.requireNonNull(uri);
     this.healthCheck = Objects.requireNonNull(healthCheck);
     this.maxSessionCount = Math.min(maxSessionCount, factories.size());
@@ -268,15 +272,17 @@ private Map<String, Object> toJson() {
 
   public static Builder builder(
       DistributedTracer tracer,
+      EventBus bus,
       HttpClient.Factory httpClientFactory,
       URI uri,
       SessionMap sessions) {
-    return new Builder(tracer, httpClientFactory, uri, sessions);
+    return new Builder(tracer, bus, httpClientFactory, uri, sessions);
   }
 
   public static class Builder {
 
     private final DistributedTracer tracer;
+    private final EventBus bus;
     private final HttpClient.Factory httpClientFactory;
     private final URI uri;
     private final SessionMap sessions;
@@ -288,10 +294,12 @@ public static class Builder {
 
     public Builder(
         DistributedTracer tracer,
+        EventBus bus,
         HttpClient.Factory httpClientFactory,
         URI uri,
         SessionMap sessions) {
       this.tracer = Objects.requireNonNull(tracer);
+      this.bus = Objects.requireNonNull(bus);
       this.httpClientFactory = Objects.requireNonNull(httpClientFactory);
       this.uri = Objects.requireNonNull(uri);
       this.sessions = Objects.requireNonNull(sessions);
@@ -329,6 +337,7 @@ public LocalNode build() {
 
       return new LocalNode(
           tracer,
+          bus,
           uri,
           check,
           maxCount,

java/server/src/org/openqa/selenium/grid/router/Router.java

@@ -27,6 +27,7 @@
 import org.openqa.selenium.grid.web.Routes;
 import org.openqa.selenium.injector.Injector;
 import org.openqa.selenium.json.Json;
+import org.openqa.selenium.remote.http.HttpClient;
 import org.openqa.selenium.remote.http.HttpRequest;
 import org.openqa.selenium.remote.http.HttpResponse;
 import org.openqa.selenium.remote.tracing.DistributedTracer;
@@ -43,9 +44,14 @@ public class Router implements Predicate<HttpRequest>, CommandHandler {
   private final Injector injector;
   private final Routes routes;
 
-  public Router(DistributedTracer tracer, SessionMap sessions, Distributor distributor) {
+  public Router(
+      DistributedTracer tracer,
+      HttpClient.Factory clientFactory,
+      SessionMap sessions,
+      Distributor distributor) {
     injector = Injector.builder()
         .register(tracer)
+        .register(clientFactory)
         .register(sessions)
         .register(distributor)
         .register(new Json())

java/server/src/org/openqa/selenium/grid/router/httpd/RouterServer.java

@@ -104,19 +104,22 @@ public Executable configure(String... args) {
 
       SessionMapOptions sessionsOptions = new SessionMapOptions(config);
       URL sessionMapUrl = sessionsOptions.getSessionMapUri().toURL();
+
+      HttpClient.Factory clientFactory = HttpClient.Factory.createDefault();
+
       SessionMap sessions = new RemoteSessionMap(
-          HttpClient.Factory.createDefault().createClient(sessionMapUrl));
+          clientFactory.createClient(sessionMapUrl));
 
       BaseServerOptions serverOptions = new BaseServerOptions(config);
 
       DistributorOptions distributorOptions = new DistributorOptions(config);
       URL distributorUrl = distributorOptions.getDistributorUri().toURL();
       Distributor distributor = new RemoteDistributor(
           tracer,
-          HttpClient.Factory.createDefault(),
+          clientFactory,
           distributorUrl);
 
-      Router router = new Router(tracer, sessions, distributor);
+      Router router = new Router(tracer, clientFactory, sessions, distributor);
 
       Server<?> server = new BaseServer<>(serverOptions);
       server.addRoute(Routes.matching(router).using(router).decorateWith(W3CCommandHandler.class));

java/server/src/org/openqa/selenium/grid/server/BUCK

@@ -5,13 +5,15 @@ java_library(
         "//third_party/java/servlet:javax.servlet-api",
     ],
     exported_deps = [
+        "//java/server/src/org/openqa/selenium/events:events",
         "//java/server/src/org/openqa/selenium/grid/component:component",
         "//java/server/src/org/openqa/selenium/grid/log:log",
         "//java/server/src/org/openqa/selenium/grid/web:web",
     ],
     deps = [
         "//java/client/src/org/openqa/selenium/remote:remote",
         "//java/client/src/org/openqa/selenium/remote/tracing:tracing",
+        "//java/server/src/org/openqa/selenium/events/zeromq:zeromq",
         "//java/server/src/org/openqa/selenium/injector:injector",
         "//java/server/src/org/openqa/selenium/grid/config:config",
         "//third_party/java/beust:jcommander",