support for Event and AdminEvent subscriptions

graphql/src/main/java/net/brianlevine/keycloak/graphql/GraphQLController.java

@@ -7,35 +7,47 @@
 import graphql.schema.idl.SchemaPrinter;
 import io.leangen.graphql.GraphQLRuntime;
 import io.leangen.graphql.GraphQLSchemaGenerator;
+import io.vertx.core.Vertx;
 import jakarta.ws.rs.core.Request;
 import jakarta.ws.rs.core.HttpHeaders;
 import net.brianlevine.keycloak.graphql.queries.ErrorQuery;
 import net.brianlevine.keycloak.graphql.queries.RealmQuery;
 import net.brianlevine.keycloak.graphql.queries.UserQuery;
+import net.brianlevine.keycloak.graphql.subscriptions.EventsSubscription;
 import net.brianlevine.keycloak.graphql.util.OverrideTypeInfoGenerator;
+import org.keycloak.events.Event;
 import org.keycloak.models.KeycloakSession;
 
 import java.util.HashMap;
 import java.util.Map;
 
 public class GraphQLController {
 
-    private GraphQL graphQL;
+    private static GraphQL graphQL;
 
     public GraphQLController() {
     }
 
-    private GraphQL getSchema() {
+    public static GraphQL getSchema() {
+        return getSchema(false);
+    }
+
+    public static GraphQL getSchema(boolean reset) {
+
+        if (reset) {
+            graphQL = null;
+        }
 
         if (graphQL == null) {
             RealmQuery realmQuery = new RealmQuery();
             ErrorQuery errorQuery = new ErrorQuery();
             UserQuery userQuery = new UserQuery();
+            EventsSubscription testSubscription = new EventsSubscription();
 
             //Schema generated from query classes
             GraphQLSchema schema = new GraphQLSchemaGenerator()
-                    .withBasePackages("net.brianlevine.keycloak.graphql")
-                    .withOperationsFromSingletons(realmQuery, errorQuery, userQuery)
+                    .withBasePackages("net.brianlevine.keycloak.graphql", "org.keycloak.events", "org.keycloak.events.admin")
+                    .withOperationsFromSingletons(realmQuery, errorQuery, userQuery, testSubscription)
                     .withRelayConnectionCheckRelaxed()
                     .withTypeInfoGenerator(new OverrideTypeInfoGenerator().withHierarchicalNames(false))
                     .generate();
@@ -51,11 +63,20 @@ private GraphQL getSchema() {
         return graphQL;
     }
 
-    public Map<String, Object> executeQuery(String query, String operationName, KeycloakSession session, Request request, HttpHeaders headers, Map<String, Object> variables) {
+    public Map<String, Object> executeQuery(
+            String query,
+            String operationName,
+            KeycloakSession session,
+            Request request,
+            HttpHeaders headers,
+            Vertx vertx,
+            Map<String, Object> variables) {
+
         Map<String, Object> ctx = new HashMap<>();
         ctx.put("keycloak.session", session);
         ctx.put("request", request);
         ctx.put("headers", headers);
+        ctx.put("vertx", vertx);
 
         ExecutionResult executionResult = getSchema().execute(ExecutionInput.newExecutionInput()
                 .query(query)

graphql/src/main/java/net/brianlevine/keycloak/graphql/KeycloakGraphQLDataFetcherExceptionHandler.java

@@ -27,7 +27,7 @@ public DataFetcherExceptionHandlerResult onException(DataFetcherExceptionHandler
         return DataFetcherExceptionHandlerResult.newResult().error(error).build();
     }
 
-    @Override
+
     protected void logException(ExceptionWhileDataFetching error, Throwable exception) {
         LOGGER.error(error.getMessage(), exception);
     }

graphql/src/main/java/net/brianlevine/keycloak/graphql/SubscriptionServer.java

@@ -0,0 +1,76 @@
+package net.brianlevine.keycloak.graphql;
+
+
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.http.HttpServerOptions;
+import io.vertx.ext.web.Router;
+import io.vertx.ext.web.handler.graphql.ApolloWSHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+public class SubscriptionServer extends AbstractVerticle {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionServer.class);
+  private static final int DEFAULT_PORT = 8081;
+
+  @Override
+  public void start() {
+
+    String sPort = System.getenv("SUBSCRIPTION_PORT");
+    int port = (sPort != null) ? Integer.parseInt(sPort) : DEFAULT_PORT;
+
+    LOGGER.info("Starting SubscriptionServer on port {}...", port);
+
+    Router router = Router.router(vertx);
+    Map<String, Object> context = new HashMap<>();
+    context.put("vertex", getVertx());
+
+    //noinspection deprecation
+    ApolloWSHandler h = ApolloWSHandler.create(GraphQLController.getSchema()).beforeExecute((a) -> {
+      a.builder().graphQLContext(context);
+    });
+    router.route("/graphql").handler(h);
+
+    HttpServerOptions httpServerOptions = new HttpServerOptions()
+            .addWebSocketSubProtocol("graphql-ws")
+            .setLogActivity(true);
+
+    vertx.createHttpServer(httpServerOptions)
+      .requestHandler(router)
+      .listen(port, "0.0.0.0")
+            .onComplete(
+                    (e) -> LOGGER.info("SubscriptionServer started on port {}", e.actualPort()),
+                    (t) -> LOGGER.error("SubscriptionServer FAILED to start: ", t)
+            ).onFailure((e) -> LOGGER.error("SubscriptionServer FAILED to start: ", e));
+  }
+
+//  private GraphQL createGraphQL() {
+//    String schema = vertx.fileSystem().readFileBlocking("links.graphqls").toString();
+//
+//    SchemaParser schemaParser = new SchemaParser();
+//    TypeDefinitionRegistry typeDefinitionRegistry = schemaParser.parse(schema);
+//
+//    RuntimeWiring runtimeWiring = newRuntimeWiring()
+//            .type("Subscription", builder -> builder.dataFetcher("links", this::linksFetcher))
+//            .build();
+//
+//    SchemaGenerator schemaGenerator = new SchemaGenerator();
+//    GraphQLSchema graphQLSchema = schemaGenerator.makeExecutableSchema(typeDefinitionRegistry, runtimeWiring);
+//
+//    return GraphQL.newGraphQL(graphQLSchema)
+//            .build();
+//  }
+
+
+
+
+//  private Publisher<Link> linksFetcher(DataFetchingEnvironment env) {
+//    return Flowable.interval(1, TimeUnit.SECONDS) // Ticks
+//      .zipWith(Flowable.fromIterable(links), (tick, link) -> link) // Emit link on each tick
+//      .observeOn(RxHelper.scheduler(context)); // Observe on the verticle context thread
+//  }
+}

graphql/src/main/java/net/brianlevine/keycloak/graphql/events/MulticastEventListenerProvider.java

@@ -0,0 +1,62 @@
+package net.brianlevine.keycloak.graphql.events;
+
+import io.reactivex.rxjava3.processors.MulticastProcessor;
+import org.jboss.logging.Logger;
+import org.keycloak.events.Event;
+import org.keycloak.events.EventListenerProvider;
+import org.keycloak.events.admin.AdminEvent;
+
+/**
+ * An EventListenerProvider that sends Keycloak Events and AdminEvents to
+ * a reactive MulticaseProcessor which can be subscribed to to receive those events.
+ * This is used to implement the server side for Event and AdminEvent GraphQL subscriptions.
+ */
+public class MulticastEventListenerProvider
+        implements EventListenerProvider {
+
+    private static final Logger LOGGER = Logger.getLogger(MulticastEventListenerProvider.class);
+
+    private final MulticastProcessor<Event> eventMulticastProcessor;
+    private final MulticastProcessor<AdminEvent> adminEventMulticastProcessor;
+
+    public MulticastEventListenerProvider(MulticastProcessor<Event> eventMulticastProcessor, MulticastProcessor<AdminEvent> adminEventMulticastProcessor) {
+        this.eventMulticastProcessor = eventMulticastProcessor;
+        this.adminEventMulticastProcessor = adminEventMulticastProcessor;
+    }
+
+    @Override
+    public void onEvent(Event event) {
+        sendEvent(event);
+    }
+
+
+    @Override
+    public void onEvent(AdminEvent adminEvent, boolean b) {
+        sendEvent(adminEvent);
+    }
+
+    private void sendEvent(Event e) {
+        if (eventMulticastProcessor.hasSubscribers()) {
+            boolean res = eventMulticastProcessor.offer(e);
+
+            if (!res) {
+                LOGGER.warn("multicastProcessor.offer() returned false");
+            }
+        }
+    }
+
+    private void sendEvent(AdminEvent e) {
+        if (adminEventMulticastProcessor.hasSubscribers()) {
+            boolean res = adminEventMulticastProcessor.offer(e);
+
+            if (!res) {
+                LOGGER.warn("multicastProcessor.offer() returned false");
+            }
+        }
+    }
+
+
+    @Override
+    public void close() {
+    }
+}

graphql/src/main/java/net/brianlevine/keycloak/graphql/events/MulticastEventListenerProviderFactory.java

@@ -0,0 +1,57 @@
+package net.brianlevine.keycloak.graphql.events;
+
+import com.google.auto.service.AutoService;
+import io.reactivex.rxjava3.processors.MulticastProcessor;
+
+import org.keycloak.Config;
+import org.keycloak.events.Event;
+import org.keycloak.events.EventListenerProvider;
+import org.keycloak.events.EventListenerProviderFactory;
+import org.keycloak.events.admin.AdminEvent;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.KeycloakSessionFactory;
+
+/**
+ * Creates MulticastEventListenerProviders for Keycloak Events and AdminEvents
+ */
+@AutoService(EventListenerProviderFactory.class)
+public class MulticastEventListenerProviderFactory implements EventListenerProviderFactory {
+    private static MulticastProcessor<Event> eventMulticastProcessor;
+    private static MulticastProcessor<AdminEvent> adminEventMulticastProcessor;
+
+    public static MulticastProcessor<Event> getEventMulticastProcessor() {
+        return eventMulticastProcessor;
+    }
+
+    public static MulticastProcessor<AdminEvent> getAdminEventMulticastProcessor() {
+        return adminEventMulticastProcessor;
+    }
+
+    @Override
+    public EventListenerProvider create(KeycloakSession keycloakSession) {
+        return new MulticastEventListenerProvider(eventMulticastProcessor, adminEventMulticastProcessor);
+    }
+
+    @Override
+    public void init(Config.Scope scope) {
+        eventMulticastProcessor = MulticastProcessor.create(false);
+        eventMulticastProcessor.startUnbounded();
+        adminEventMulticastProcessor = MulticastProcessor.create(false);
+        adminEventMulticastProcessor.startUnbounded();
+    }
+
+    @Override
+    public void postInit(KeycloakSessionFactory keycloakSessionFactory) {
+
+    }
+
+    @Override
+    public void close() {
+
+    }
+
+    @Override
+    public String getId() {
+        return "multicast-event-listener";
+    }
+}

graphql/src/main/java/net/brianlevine/keycloak/graphql/queries/RealmQuery.java

@@ -19,10 +19,10 @@
 import org.keycloak.services.resources.admin.AdminAuth;
 import org.keycloak.services.resources.admin.permissions.AdminPermissions;
 
-
 import java.util.List;
 import java.util.Objects;
 
+
 public class RealmQuery {
 
     @GraphQLQuery(name = "realms", description = "Return a collection of realms that are viewable by the caller.")

graphql/src/main/java/net/brianlevine/keycloak/graphql/rest/GraphQLResourceProvider.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import io.vertx.core.Vertx;
 import jakarta.ws.rs.*;
 import jakarta.ws.rs.core.*;
 
@@ -28,10 +29,12 @@ public class GraphQLResourceProvider implements RealmResourceProvider {
 
 	private final KeycloakSession session;
 	private final GraphQLController graphql;
+	private final Vertx vertx;
 
-	public GraphQLResourceProvider(KeycloakSession session, GraphQLController graphql) {
+	public GraphQLResourceProvider(KeycloakSession session, GraphQLController graphql, Vertx vertx) {
 		this.session = session;
 		this.graphql = graphql;
+		this.vertx = vertx;
 
 		LOGGER.debug("Created GraphQLResourceProvider");
 	}
@@ -62,6 +65,7 @@ public Response postGraphQL(Map<String, Object> body, @Context Request request,
 				session,
 				request,
 				headers,
+				vertx,
 				variables != null ? (Map<String, Object>) variables : Collections.emptyMap());
 
 		ObjectMapper mapper = new ObjectMapper();