Fix missing session id on subscription events

Author: gasper-vrhovsek

extension/runtime/src/main/java/org/iris_events/producer/RoutingDetails.java

@@ -12,20 +12,22 @@ public final class RoutingDetails {
     private final String routingKey;
     private final Scope scope;
     private final String userId;
+    private final String sessionId;
     private final String subscriptionId;
     private final boolean persistent;
     private final Integer cacheTtl;
     private final boolean propagate;
 
     public RoutingDetails(final String eventName, final String exchange, final ExchangeType exchangeType,
-            final String routingKey, final Scope scope, final String userId,
+            final String routingKey, final Scope scope, final String userId, final String sessionId,
             final String subscriptionId, final boolean persistent, final Integer cacheTtl, final boolean propagate) {
         this.eventName = eventName;
         this.exchange = exchange;
         this.exchangeType = exchangeType;
         this.routingKey = routingKey;
         this.scope = scope;
         this.userId = userId;
+        this.sessionId = sessionId;
         this.subscriptionId = subscriptionId;
         this.persistent = persistent;
         this.cacheTtl = cacheTtl;
@@ -56,6 +58,10 @@ public String getUserId() {
         return userId;
     }
 
+    public String getSessionId() {
+        return sessionId;
+    }
+
     public String getSubscriptionId() {
         return subscriptionId;
     }
@@ -203,6 +209,11 @@ public MiscRoutingDetailsBuilder userId(final String userId) {
             return this;
         }
 
+        public MiscRoutingDetailsBuilder sessionId(final String sessionId) {
+            this.sessionId = sessionId;
+            return this;
+        }
+
         public MiscRoutingDetailsBuilder subscriptionId(final String subscriptionId) {
             this.subscriptionId = subscriptionId;
             return this;
@@ -224,7 +235,7 @@ public MiscRoutingDetailsBuilder propagate(final boolean propagate) {
         }
 
         public RoutingDetails build() {
-            return new RoutingDetails(eventName, exchange, exchangeType, routingKey, scope, userId, subscriptionId,
+            return new RoutingDetails(eventName, exchange, exchangeType, routingKey, scope, userId, sessionId, subscriptionId,
                     persistent, cacheTtl, propagate);
         }
     }

extension/runtime/src/main/java/org/iris_events/runtime/BasicPropertiesProvider.java

@@ -106,6 +106,7 @@ private AMQP.BasicProperties buildAmqpBasicPropertiesWithCustomHeaders(final AMQ
         final var eventName = routingDetails.getEventName();
         final var scope = routingDetails.getScope();
         final var userId = routingDetails.getUserId();
+        final var sessionId = routingDetails.getSessionId();
         final var propagate = routingDetails.getPropagate();
         final var subscriptionId = routingDetails.getSubscriptionId();
         final var cacheTtl = routingDetails.getCacheTtl();
@@ -118,6 +119,7 @@ private AMQP.BasicProperties buildAmqpBasicPropertiesWithCustomHeaders(final AMQ
         headers.put(EVENT_TYPE, eventName);
         headers.put(SERVER_TIMESTAMP, timestampProvider.getCurrentTimestamp());
         Optional.ofNullable(subscriptionId).ifPresent(id -> headers.put(SUBSCRIPTION_ID, id));
+        Optional.ofNullable(sessionId).ifPresent(sId -> headers.put(SESSION_ID, sId));
         Optional.ofNullable(cacheTtl).ifPresent(ttl -> headers.put(CACHE_TTL, ttl));
 
         if (scope != Scope.INTERNAL) {