Client: improve handling of reply context

wirew0rm · RalphSteinhagen · commit d6cf8d91a0f3 · 2026-01-27T11:22:27.000+01:00
- Future: set atomic to ready only after the reply has been set
- ReplyQuery: only include entries from reply context, do not duplicate
  entries from request url
- DataSourcePublisher.get: allow access to the reply context from the
  returned future
- DataSourcePublsher subscription listener: fix handling of Context type
  none

Signed-off-by: Alexander Krimm &lt;A.Krimm@gsi.de&gt;
diff --git a/client/src/main/java/io/opencmw/client/DataSourcePublisher.java b/client/src/main/java/io/opencmw/client/DataSourcePublisher.java
@@ -20,10 +20,7 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -317,9 +314,9 @@ protected boolean handleDataSourceSockets() {
         return dataAvailable;
     }
 
-    protected <R> ThePromisedFuture<R, ?> newRequestFuture(final URI endpoint, final Class<R> requestedDomainObjType, final Command requestType, final String requestId) {
+    protected <R, C> ThePromisedFuture<R, C> newRequestFuture(final URI endpoint, final Class<R> requestedDomainObjType, final Command requestType, final String requestId) {
         FilterRegistry.checkClassForNewFilters(requestedDomainObjType);
-        final ThePromisedFuture<R, ?> requestFuture = new ThePromisedFuture<>(endpoint, requestedDomainObjType, null, requestType, requestId, null);
+        final ThePromisedFuture<R, C> requestFuture = new ThePromisedFuture<>(endpoint, requestedDomainObjType, null, requestType, requestId, null);
         final Object oldEntry = requests.put(requestId, requestFuture);
         assert oldEntry == null : "requestID '" + requestId + "' already present in requestFutureMap";
         return requestFuture;
@@ -367,19 +364,19 @@ protected void internalEventHandler(final RingBufferEvent event, final long sequ
                     replyDomainObject = ioClassSerialiser.deserialiseObject(reqClassType);
                     ioClassSerialiser.setDataBuffer(byteBuffer); // allow received byte array to be released
                 }
-                if (notifyFuture) {
-                    domainObject.future.castAndSetReply(replyDomainObject); // notify callback
+                final Object contextObject;
+                if (domainObject.future.contextType == null || domainObject.future.contextType.equals(Map.class)) {
+                    contextObject = QueryParameterParser.getMap(endpointURI.getQuery());
+                } else {
+                    contextObject = QueryParameterParser.parseQueryParameter(domainObject.future.contextType, endpointURI.getQuery());
                 }
                 if (domainObject.future.listener != null) {
                     final var finalDomainObj = replyDomainObject;
-                    final Object contextObject;
-                    if (domainObject.future.contextType == null) {
-                        contextObject = QueryParameterParser.getMap(endpointURI.getQuery());
-                    } else {
-                        contextObject = QueryParameterParser.parseQueryParameter(domainObject.future.contextType, endpointURI.getQuery());
-                    }
                     executor.submit(() -> domainObject.future.notifyListener(finalDomainObj, contextObject)); // NOPMD - threads are ok, not a webapp
                 }
+                if (notifyFuture) {
+                    domainObject.future.castAndSetReplyWithContext(replyDomainObject, contextObject); // notify callback
+                }
             } catch (Exception e) { // NOPMD: exception is forwarded to client
                 final var sw = new StringWriter();
                 final var pw = new PrintWriter(sw);
@@ -453,18 +450,18 @@ private Client() { // accessed via outer class method
             clientSocket.connect(inprocCtrl);
         }
 
-        public <R, C> Future<R> get(URI endpoint, final C requestContext, final Class<R> requestedDomainObjType, final RbacProvider... rbacProvider) {
+        public <R, C> ThePromisedFuture<R, C> get(URI endpoint, final C requestContext, final Class<R> requestedDomainObjType, final RbacProvider... rbacProvider) {
             final String requestId = clientId + internalReqIdGenerator.incrementAndGet();
             final URI endpointQuery = getEndpointQuery(endpoint, requestContext);
-            final ThePromisedFuture<R, ?> rThePromisedFuture = newRequestFuture(endpointQuery, requestedDomainObjType, Command.GET_REQUEST, requestId);
+            final ThePromisedFuture<R, C> rThePromisedFuture = newRequestFuture(endpointQuery, requestedDomainObjType, Command.GET_REQUEST, requestId);
             request(requestId, Command.GET_REQUEST, endpointQuery, null, requestContext, rbacProvider);
             return rThePromisedFuture;
         }
 
-        public <R, C> Future<R> set(final URI endpoint, final R requestBody, final C requestContext, final Class<R> requestedDomainObjType, final RbacProvider... rbacProvider) {
+        public <R, C> ThePromisedFuture<R, C> set(final URI endpoint, final R requestBody, final C requestContext, final Class<R> requestedDomainObjType, final RbacProvider... rbacProvider) {
             final String requestId = clientId + internalReqIdGenerator.incrementAndGet();
             final URI endpointQuery = getEndpointQuery(endpoint, requestContext);
-            final ThePromisedFuture<R, ?> rThePromisedFuture = newRequestFuture(endpointQuery, requestedDomainObjType, Command.SET_REQUEST, requestId);
+            final ThePromisedFuture<R, C> rThePromisedFuture = newRequestFuture(endpointQuery, requestedDomainObjType, Command.SET_REQUEST, requestId);
             request(requestId, Command.SET_REQUEST, endpointQuery, requestBody, requestContext, rbacProvider);
             return rThePromisedFuture;
         }
@@ -551,8 +548,9 @@ private <R, C> void request(final String requestId, final Command requestType, f
         }
     }
 
-    protected static class ThePromisedFuture<R, C> extends CustomFuture<R> { // NOPMD - no need for setters/getters here
+    public static class ThePromisedFuture<R, C> extends CustomFuture<R> { // NOPMD - no need for setters/getters here
         private final URI endpoint;
+        private C replyContext;
         private final Class<R> requestedDomainObjType;
         private final Class<C> contextType;
         private final Command requestType;
@@ -586,10 +584,12 @@ public String getInternalRequestID() {
         }
 
         public void notifyListener(final Object obj, final Object contextObject) {
-            if (obj == null || !requestedDomainObjType.isAssignableFrom(obj.getClass()) || !contextType.isAssignableFrom(contextObject.getClass())) {
+            if (obj == null || !requestedDomainObjType.isAssignableFrom(obj.getClass())) {
                 LOGGER.atError().addArgument(requestedDomainObjType.getName()).addArgument(obj == null ? "null" : obj.getClass().getName()).log("Got wrong type for notification, got {} expected {}");
+            } else if (contextType != null && !contextType.isAssignableFrom(contextObject.getClass())) {
+                LOGGER.atError().addArgument(contextObject.getClass().getName()).addArgument(contextType.getName()).log("Got wrong context type for notification, got {} expected {}");
             } else {
-                //noinspection unchecked - cast is checked dynamically
+                // noinspection unchecked - cast is checked dynamically
                 listener.dataUpdate((R) obj, (C) contextObject); // NOPMD NOSONAR - cast is checked before implicitly
             }
         }
@@ -598,6 +598,17 @@ public void notifyListener(final Object obj, final Object contextObject) {
         protected void castAndSetReply(final Object newValue) {
             this.setReply((R) newValue);
         }
+
+        @SuppressWarnings("unchecked")
+        protected void castAndSetReplyWithContext(final Object newValue, final Object contextObject) {
+            this.replyContext = (C) contextObject;
+            this.setReply((R) newValue);
+        }
+
+        public C getReplyContext() throws ExecutionException, InterruptedException {
+            super.get();
+            return this.replyContext;
+        }
     }
 
     protected static class InternalDomainObject {
diff --git a/client/src/test/java/io/opencmw/client/OpenCmwDataSourceTest.java b/client/src/test/java/io/opencmw/client/OpenCmwDataSourceTest.java
@@ -8,6 +8,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Queue;
@@ -257,18 +258,21 @@ void testGetRequest() throws InterruptedException, ExecutionException, TimeoutEx
         // get request
         final URI requestURI = URI.create(brokerAddress + "/testWorker?ctx=FAIR.SELECTOR.C=3&contentType=application/octet-stream");
         LOGGER.atDebug().addArgument(requestURI).log("requesting GET from endpoint: {}");
-        final Future<TestObject> future;
+        final DataSourcePublisher.ThePromisedFuture<TestObject, Map<String, List<String>>> future;
         try (final DataSourcePublisher.Client client = dataSourcePublisher.getClient()) {
             future = client.get(requestURI, null, TestObject.class); // uri_without_query oder serviceName + resolver, requestContext, type
         }
 
         // assert result
         final TestObject result = future.get(1000, TimeUnit.MILLISECONDS);
         assertEquals(referenceObject, result);
+        final Map<String, List<String>> context = future.getReplyContext();
+        assertEquals("FAIR.SELECTOR.C=3:P=5", context.get("ctx").get(0));
 
         eventStore.stop();
         dataSourcePublisher.stop();
     }
+
     @Test
     void testGetRequestWithContext() throws InterruptedException, ExecutionException, TimeoutException {
         final TestObject referenceObject = new TestObject("asdf", 42);
diff --git a/core/src/main/java/io/opencmw/utils/CustomFuture.java b/core/src/main/java/io/opencmw/utils/CustomFuture.java
@@ -92,18 +92,18 @@ public boolean isDone() {
      * @throws IllegalStateException in case this method has been already called or Future has been cancelled
      */
     public void setReply(final T newValue) {
+        this.reply = newValue;
         if (done.getAndSet(true)) {
             throw new IllegalStateException("future is not running anymore (either cancelled or already notified)");
         }
-        this.reply = newValue;
         notifyListener();
     }
 
     public void setException(final Throwable exception) {
+        this.exception = exception;
         if (done.getAndSet(true)) {
             throw new IllegalStateException("future is not running anymore (either cancelled or already notified)");
         }
-        this.exception = exception;
         notifyListener();
     }
 
diff --git a/server/src/main/java/io/opencmw/server/MajordomoWorker.java b/server/src/main/java/io/opencmw/server/MajordomoWorker.java
@@ -214,9 +214,7 @@ protected void serialiseData(final IoClassSerialiser classSerialiser, final IoBu
             final URI reqTopic = rawCtx.req.topic;
             rawCtx.rep.topic = new URI(reqTopic.getScheme(), reqTopic.getAuthority(), reqTopic.getPath(), replyQuery, reqTopic.getFragment());
         } else {
-            final String oldQuery = rawCtx.rep.topic.getQuery();
-            final String newQuery = oldQuery == null || oldQuery.isBlank() ? replyQuery : (oldQuery + "&" + replyQuery);
-            rawCtx.rep.topic = new URI(rawCtx.rep.topic.getScheme(), rawCtx.rep.topic.getAuthority(), rawCtx.rep.topic.getPath(), newQuery, null);
+            rawCtx.rep.topic = new URI(rawCtx.rep.topic.getScheme(), rawCtx.rep.topic.getAuthority(), rawCtx.rep.topic.getPath(), replyQuery, null);
         }
         final MimeType replyMimeType = QueryParameterParser.getMimeType(replyQuery);
         // no MIME type given -> stick with the one specified in the request (if it exists) or keep default: copy of raw binary data

Original file line number	Diff line number	Diff line change
`@@ -92,18 +92,18 @@ public boolean isDone() {`
`92`	`92`	`* @throws IllegalStateException in case this method has been already called or Future has been cancelled`
`93`	`93`	`*/`
`94`	`94`	`public void setReply(final T newValue) {`
	`95`	`+ this.reply = newValue;`
`95`	`96`	`if (done.getAndSet(true)) {`
`96`	`97`	`throw new IllegalStateException("future is not running anymore (either cancelled or already notified)");`
`97`	`98`	`}`
`98`		`- this.reply = newValue;`
`99`	`99`	`notifyListener();`
`100`	`100`	`}`
`101`	`101`
`102`	`102`	`public void setException(final Throwable exception) {`
	`103`	`+ this.exception = exception;`
`103`	`104`	`if (done.getAndSet(true)) {`
`104`	`105`	`throw new IllegalStateException("future is not running anymore (either cancelled or already notified)");`
`105`	`106`	`}`
`106`		`- this.exception = exception;`
`107`	`107`	`notifyListener();`
`108`	`108`	`}`
`109`	`109`