Set decoder to ignoreAll after session is taken over by other MQTT cl…

…ient (#1045)

runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java

@@ -3172,7 +3172,7 @@ private MqttPublishStream resolvePublishStream(
                 final long resolvedId = resolved.id;
 
                 stream = publishes.computeIfAbsent(topicKey, s ->
-                    new MqttPublishStream(routedId, resolvedId, topic, qos, binding.supplyModelConfig(topic)));
+                    new MqttPublishStream(routedId, resolvedId, topic, topicKey, binding.supplyModelConfig(topic)));
                 stream.doPublishBegin(traceId, affinity, qos);
             }
             else
@@ -4890,6 +4890,7 @@ private void closeStreams(
             {
                 session.cleanupEnd(traceId);
             }
+            decoder = decodeIgnoreAll;
         }
 
         private void cleanupBudgetCreditor()
@@ -5669,15 +5670,15 @@ private class MqttPublishStream
                 long originId,
                 long routedId,
                 String topic,
-                int qos,
+                long topicKey,
                 ModelConfig config)
             {
                 this.originId = originId;
                 this.routedId = routedId;
                 this.initialId = supplyInitialId.applyAsLong(routedId);
                 this.replyId = supplyReplyId.applyAsLong(initialId);
                 this.topic = topic;
-                this.topicKey = topicKey(topic, qos);
+                this.topicKey = topicKey;
                 this.contentType = config != null ? supplyValidator.apply(config) : null;
             }
 

runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/PublishIT.java

@@ -66,6 +66,16 @@ public void shouldPublishOneMessage() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${net}/publish.session.takeover/client",
+        "${app}/publish.session.takeover/server"})
+    public void shouldPublishAfterSessionTakeover() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("server.yaml")
     @Specification({

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/publish.session.takeover/client.rpt

@@ -0,0 +1,73 @@
+#
+# Copyright 2021-2023 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+connect "zilla://streams/app0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${mqtt:beginEx()
+                            .typeId(zilla:id("mqtt"))
+                            .session()
+                              .flags("CLEAN_START")
+                              .clientId("client")
+                              .build()
+                            .build()}
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(66560)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("client")
+                                .build()
+                              .build()}
+
+connected
+
+read zilla:data.empty
+read notify RECEIVED_SESSION_STATE
+
+read notify SENT_FIRST_MESSAGE
+read closed
+
+
+connect await RECEIVED_SESSION_STATE
+        "zilla://streams/app0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${mqtt:beginEx()
+                             .typeId(zilla:id("mqtt"))
+                              .publish()
+                                .clientId("client")
+                                .topic("sensor/one")
+                                .build()
+                             .build()}
+
+connected
+
+write zilla:data.ext ${mqtt:dataEx()
+                             .typeId(zilla:id("mqtt"))
+                             .publish()
+                                .build()
+                             .build()}
+
+write "message1"
+
+write close

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/publish.session.takeover/server.rpt

@@ -0,0 +1,74 @@
+#
+# Copyright 2021-2023 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+accept "zilla://streams/app0"
+        option zilla:window 8192
+        option zilla:transmission "duplex"
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .session()
+                                .flags("CLEAN_START")
+                                .clientId("client")
+                                .build()
+                             .build()}
+
+write zilla:begin.ext ${mqtt:beginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(66560)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("client")
+                                .build()
+                              .build()}
+
+connected
+
+write zilla:data.empty
+write flush
+
+write await RECEIVED_FIRST_MESSAGE
+write close
+write notify SENT_CLOSE
+
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .publish()
+                                .clientId("client")
+                                .topic("sensor/one")
+                                .build()
+                             .build()}
+
+connected
+
+read zilla:data.ext ${mqtt:matchDataEx()
+                            .typeId(zilla:id("mqtt"))
+                            .publish()
+                                .build()
+                            .build()}
+
+read "message1"
+read notify RECEIVED_FIRST_MESSAGE
+
+read closed

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/publish.session.takeover/client.rpt

@@ -0,0 +1,54 @@
+#
+# Copyright 2021-2023 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+connect "zilla://streams/net0"
+   option zilla:window 8192
+   option zilla:transmission "duplex"
+  option zilla:byteorder "network"
+
+connected
+
+write [0x10 0x18]                    # CONNECT
+      [0x00 0x04] "MQTT"             # protocol name
+      [0x05]                         # protocol version
+      [0x02]                         # flags = clean start
+      [0x00 0x3c]                    # keep alive = 60s
+      [0x05]                         # properties
+      [0x27] 66560                   # maximum packet size = 66560
+      [0x00 0x06] "client"           # client id
+
+read  [0x20 0x03]                    # CONNACK
+      [0x00]                         # flags = none
+      [0x00]                         # reason code
+      [0x00]                         # properties
+
+write [0x30 0x15]                    # PUBLISH
+      [0x00 0x0a] "sensor/one"       # topic name
+      [0x00]                         # properties
+      "message1"                     # payload
+
+write await SENT_CLOSE
+write [0x30 0x11]                    # PUBLISH
+      [0x00 0x0a] "sensor/one"       # topic name
+      [0x00]                         # properties
+write "test"                         # payload
+
+read  [0xe0 0x02]                    # DISCONNECT
+      [0x8e]                         # session taken over
+      [0x00]                         # properties = none
+
+read closed
+write close

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/publish.session.takeover/server.rpt

@@ -0,0 +1,56 @@
+#
+# Copyright 2021-2023 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+accept "zilla://streams/net0"
+  option zilla:window 8192
+  option zilla:transmission "duplex"
+  option zilla:byteorder "network"
+
+accepted
+connected
+
+read  [0x10 0x18]                    # CONNECT
+      [0x00 0x04] "MQTT"             # protocol name
+      [0x05]                         # protocol version
+      [0x02]                         # flags = clean start
+      [0x00 0x3c]                    # keep alive = 60s
+      [0x05]                         # properties
+      [0x27] 66560                   # maximum packet size = 66560
+      [0x00 0x06] "client"           # client id
+
+write [0x20 0x03]                    # CONNACK
+      [0x00]                         # flags = none
+      [0x00]                         # reason code
+      [0x00]                         # properties = none
+
+read  [0x30 0x15]                    # PUBLISH
+      [0x00 0x0a] "sensor/one"       # topic name
+      [0x00]                         # properties
+      "message1"                     # payload
+
+read notify SENT_CLOSE
+read  [0x30 0x11]                    # PUBLISH
+      [0x00 0x0a] "sensor/one"       # topic name
+      [0x00]                         # properties
+      "test"                         # payload
+
+write [0xe0 0x02]                    # DISCONNECT
+      [0x8e]                         # session taken over
+      [0x00]                         # properties = none
+
+write close
+read closed
+

specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/PublishIT.java

@@ -63,6 +63,15 @@ public void shouldSendMultipleMessages() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${app}/publish.session.takeover/client",
+        "${app}/publish.session.takeover/server"})
+    public void shouldSendMessageAfterSessionTakeover() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${app}/publish.multiple.clients/client",

specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/PublishIT.java

@@ -63,6 +63,15 @@ public void shouldSendMultipleMessages() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${net}/publish.session.takeover/client",
+        "${net}/publish.session.takeover/server"})
+    public void shouldSendMessageAfterSessionTakeover() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${net}/publish.multiple.clients/client",