feat: add option to generate clound event request in HTTP Sender Even…

…t Listener
p2-inc · Dec 4, 2024 · 6154448 · 6154448
1 parent 4dbd2c2
commit 6154448
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 13 deletions.
diff --git a/README.md b/README.md
@@ -96,6 +96,7 @@ Configuration values:
 | `backoffMaxInterval` | N | 60000 | Maximum back off time in milliseconds |
 | `backoffMultiplier` | N | 1.5 | Multiplier value (E.g. 1.5 is 50% increase per back off) |
 | `backoffRandomizationFactor` | N | 0.5 | Randomization factor (E.g. 0.5 results in a random period ranging between 50% below and 50% above the retry interval) |
+| `cloudEventEnabled` | N | false | Generate a cloud event request to Http Server |
 
 ### Adding Configuration to your EventListenerProvider
 

diff --git a/src/main/java/io/phasetwo/keycloak/events/HttpSenderEventListenerProvider.java b/src/main/java/io/phasetwo/keycloak/events/HttpSenderEventListenerProvider.java
@@ -16,6 +16,8 @@
 import org.apache.http.impl.client.HttpClients;
 import org.keycloak.broker.provider.util.LegacySimpleHttp;
 import org.keycloak.models.KeycloakSession;
+import org.keycloak.representations.idm.AdminEventRepresentation;
+import org.keycloak.representations.idm.EventRepresentation;
 import org.keycloak.util.JsonSerialization;
 
 @JBossLog
@@ -30,6 +32,7 @@ public class HttpSenderEventListenerProvider extends SenderEventListenerProvider
   protected static final String BACKOFF_MAX_INTERVAL = "backoffMaxInterval";
   protected static final String BACKOFF_MULTIPLIER = "backoffMultiplier";
   protected static final String BACKOFF_RANDOMIZATION_FACTOR = "backoffRandomizationFactor";
+  protected static final String CLOUD_EVENTS_ENABLED = "cloudEventEnabled";
 
   public HttpSenderEventListenerProvider(KeycloakSession session, ScheduledExecutorService exec) {
     super(session, exec);
@@ -54,6 +57,9 @@ String getTargetUri() {
     return config.get(TARGET_URI).toString();
   }
 
+  Boolean getCloudEventsEnabled() {
+    return getBooleanOr(config,CLOUD_EVENTS_ENABLED, false);
+  }
   Optional<String> getSharedSecret() {
     return Optional.ofNullable(config.get(SHARED_SECRET)).map(Object::toString);
   }
@@ -72,19 +78,45 @@ protected void send(
       throws SenderException, IOException {
     log.debugf("attempting send to %s", targetUri);
     try (CloseableHttpClient http = HttpClients.createDefault()) {
-      LegacySimpleHttp request = LegacySimpleHttp.doPost(targetUri, http).json(task.getEvent());
-      sharedSecret.ifPresent(
-          secret ->
-              request.header(
-                  "X-Keycloak-Signature",
-                  hmacFor(task.getEvent(), secret, algorithm.orElse(HMAC_SHA256_ALGORITHM))));
-      LegacySimpleHttp.Response response = request.asResponse();
-      int status = response.getStatus();
-      log.debugf("sent to %s (%d)", targetUri, status);
-      if (status < HTTP_OK || status >= HTTP_MULT_CHOICE) { // any 2xx is acceptable
-        log.warnf("Sending failure (Server response:%d)", status);
-        throw new SenderException(true);
-      }
+        LegacySimpleHttp request = LegacySimpleHttp.doPost(targetUri, http).json(task.getEvent());
+        sharedSecret.ifPresent(
+                secret ->
+                        request.header(
+                                "X-Keycloak-Signature",
+                                hmacFor(task.getEvent(), secret, algorithm.orElse(HMAC_SHA256_ALGORITHM))));
+
+        if (getCloudEventsEnabled()) {
+            Object eventObject = null;
+            eventObject = task.getEvent();
+            if (eventObject instanceof EventRepresentation) {
+                EventRepresentation event = (EventRepresentation) eventObject;
+                request
+                        .header("content-type", "application/json")
+                        .header("ce-specversion", "1.0")
+                        .header("ce-source", "keycloak")
+                        .header("ce-type", event.getType())
+                        .header("ce-id", event.getRealmId() +"-"+ event.getTime())
+                        .header("ce-partitionkey", event.getRealmId());
+
+            } else {
+              AdminEventRepresentation event = (AdminEventRepresentation) eventObject;
+              request
+                      .header("content-type", "application/json")
+                      .header("ce-specversion", "1.0")
+                      .header("ce-source", "keycloak")
+                      .header("ce-type", event.getOperationType() + "-"+event.getResourceType())
+                      .header("ce-id", event.getRealmId() +"-"+ event.getTime())
+                      .header("ce-partitionkey", event.getRealmId());
+            }
+        }
+
+        LegacySimpleHttp.Response response = request.asResponse();
+        int status = response.getStatus();
+        log.debugf("sent to %s (%d)", targetUri, status);
+        if (status < HTTP_OK || status >= HTTP_MULT_CHOICE) { // any 2xx is acceptable
+            log.warnf("Sending failure (Server response:%d)", status);
+            throw new SenderException(true);
+        }
     } catch (SenderException se) {
       // rethrow existing SenderException
       throw se;