From bc8e6a20cd75a761ed6270cfc952ce072064ec9b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Thomas=20J=C3=A4ckle?= <thomas.jaeckle@beyonnex.io>
Date: Wed, 28 Jun 2023 20:05:47 +0200
Subject: [PATCH] fix handling weak ack for built-in ack labels in
 StreamingSessionActor
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* before an error to the WS session was sent declaring that ack-label 'live-response' was not declared if a WS subscribed for live messages, but filtered them out based on an RQL filter
* now this situation is handled correctly by dropping the weak ack

Signed-off-by: Thomas Jäckle <thomas.jaeckle@beyonnex.io>
---
 .../streaming/actors/StreamingSessionActor.java      | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/gateway/service/src/main/java/org/eclipse/ditto/gateway/service/streaming/actors/StreamingSessionActor.java b/gateway/service/src/main/java/org/eclipse/ditto/gateway/service/streaming/actors/StreamingSessionActor.java
index cd5597f415..d847fe6c9c 100755
--- a/gateway/service/src/main/java/org/eclipse/ditto/gateway/service/streaming/actors/StreamingSessionActor.java
+++ b/gateway/service/src/main/java/org/eclipse/ditto/gateway/service/streaming/actors/StreamingSessionActor.java
@@ -27,6 +27,7 @@
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabel;
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabelNotDeclaredException;
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabelNotUniqueException;
+import org.eclipse.ditto.base.model.acks.DittoAcknowledgementLabel;
 import org.eclipse.ditto.base.model.acks.FatalPubSubException;
 import org.eclipse.ditto.base.model.auth.AuthorizationContext;
 import org.eclipse.ditto.base.model.entity.id.NamespacedEntityId;
@@ -268,6 +269,7 @@ private Receive createIncomingSignalBehavior() {
                 .build();
 
         final Receive signalBehavior = ReceiveBuilder.create()
+                .match(Acknowledgement.class, this::isWeakAckForBuiltInAckLabel, this::dropWeakAckForBuiltInAckLabelAcknowledgement)
                 .match(Acknowledgement.class, this::hasUndeclaredAckLabel, this::ackLabelNotDeclared)
                 .match(Acknowledgement.class, this::forwardAcknowledgementOrLiveCommandResponse)
                 .match(CommandResponse.class, CommandResponse::isLiveCommandResponse, liveCommandResponse ->
@@ -493,6 +495,16 @@ private static Receive addPreprocessors(final List<PartialFunction<Object, Objec
                 .orElse(receive);
     }
 
+    private boolean isWeakAckForBuiltInAckLabel(final Acknowledgement acknowledgement) {
+        return acknowledgement.isWeak() &&
+                List.of(DittoAcknowledgementLabel.values()).contains(acknowledgement.getLabel());
+    }
+
+    private void dropWeakAckForBuiltInAckLabelAcknowledgement(final Acknowledgement ack) {
+        logger.withCorrelationId(ack)
+                .info("Dropping weak ack for built-in ack label <{}>", ack.getLabel());
+    }
+
     private boolean hasUndeclaredAckLabel(final Acknowledgement acknowledgement) {
         return !declaredAcks.contains(acknowledgement.getLabel());
     }