Skip to content

Commit

Permalink
Fix MQTT binding selecting incorrect publish stream (#1328)
Browse files Browse the repository at this point in the history
  • Loading branch information
@bmaidics
bmaidics authored Nov 13, 2024
1 parent af27e62 commit b88428e
Show file tree
Hide file tree
Showing 12 changed files with 448 additions and 1 deletion.
2 changes: 1 addition & 1 deletion ...c/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1311,8 +1311,8 @@ private int decodePublishV4(
{
break decode;
}
server.decodePublisherKey = topicKey;
}
server.decodePublisherKey = topicKey;

server.decodedQos = mqttPublishHelper.qos;
server.decodedRetained = mqttPublishHelper.retained;
Expand Down
10 changes: 10 additions & 0 deletions ...test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v4/PublishIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,16 @@ public void shouldPublishRetainedMessage() throws Exception
k3po.finish();
}

@Test
@Configuration("server.yaml")
@Specification({
"${net}/publish.retained.multiple.topic/client",
"${app}/publish.retained.multiple.topic/server"})
public void shouldPublishRetainedMessageMultipleTopic() throws Exception
{
k3po.finish();
}

@Test
@Configuration("server.yaml")
@Specification({
Expand Down
10 changes: 10 additions & 0 deletions ...test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/PublishIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,16 @@ public void shouldPublishRetainedMessage() throws Exception
k3po.finish();
}

@Test
@Configuration("server.yaml")
@Specification({
"${net}/publish.retained.multiple.topic/client",
"${app}/publish.retained.multiple.topic/server"})
public void shouldPublishRetainedMessageMultipleTopic() throws Exception
{
k3po.finish();
}

@Test
@Configuration("server.yaml")
@Specification({
Expand Down
105 changes: 105 additions & 0 deletions ...y/zilla/specs/binding/mqtt/streams/application/publish.retained.multiple.topic/client.rpt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
#
# Copyright 2021-2024 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


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")
.flags("RETAIN")
.build()
.build()}

connected

write zilla:data.ext ${mqtt:dataEx()
.typeId(zilla:id("mqtt"))
.publish()
.flags("RETAIN")
.build()
.build()}

write "message"

write zilla:data.ext ${mqtt:dataEx()
.typeId(zilla:id("mqtt"))
.publish()
.flags("RETAIN")
.build()
.build()}

write "message2"


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/two")
.flags("RETAIN")
.build()
.build()}

connected

write zilla:data.ext ${mqtt:dataEx()
.typeId(zilla:id("mqtt"))
.publish()
.flags("RETAIN")
.build()
.build()}

write "message"
101 changes: 101 additions & 0 deletions ...y/zilla/specs/binding/mqtt/streams/application/publish.retained.multiple.topic/server.rpt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
#
# Copyright 2021-2024 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


accepted

read zilla:begin.ext ${mqtt:matchBeginEx()
.typeId(zilla:id("mqtt"))
.publish()
.clientId("client")
.topic("sensor/one")
.flags("RETAIN")
.build()
.build()}

connected

read zilla:data.ext ${mqtt:matchDataEx()
.typeId(zilla:id("mqtt"))
.publish()
.flags("RETAIN")
.build()
.build()}

read "message"

read zilla:data.ext ${mqtt:matchDataEx()
.typeId(zilla:id("mqtt"))
.publish()
.flags("RETAIN")
.build()
.build()}

read "message2"


accepted

read zilla:begin.ext ${mqtt:matchBeginEx()
.typeId(zilla:id("mqtt"))
.publish()
.clientId("client")
.topic("sensor/two")
.flags("RETAIN")
.build()
.build()}

connected

read zilla:data.ext ${mqtt:matchDataEx()
.typeId(zilla:id("mqtt"))
.publish()
.flags("RETAIN")
.build()
.build()}

read "message"
45 changes: 45 additions & 0 deletions ...ty/zilla/specs/binding/mqtt/streams/network/v4/publish.retained.multiple.topic/client.rpt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
#
# Copyright 2021-2024 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 0x12] # CONNECT
[0x00 0x04] "MQTT" # protocol name
[0x04] # protocol version
[0x02] # flags = clean start
[0x00 0x3c] # keep alive = 60s
[0x00 0x06] "client" # client id

read [0x20 0x02] # CONNACK
[0x00] # flags = none
[0x00] # reason code

write [0x31 0x13] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/one" # topic name
"message" # payload

write [0x31 0x13] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/two" # topic name
"message" # payload

write [0x31 0x14] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/one" # topic name
"message2" # payload
46 changes: 46 additions & 0 deletions ...ty/zilla/specs/binding/mqtt/streams/network/v4/publish.retained.multiple.topic/server.rpt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
#
# Copyright 2021-2024 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 0x12] # CONNECT
[0x00 0x04] "MQTT" # protocol name
[0x04] # protocol version
[0x02] # flags = clean start
[0x00 0x3c] # keep alive = 60s
[0x00 0x06] "client" # client id

write [0x20 0x02] # CONNACK
[0x00] # flags = none
[0x00] # reason code

read [0x31 0x13] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/one" # topic name
"message" # payload

read [0x31 0x13] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/two" # topic name
"message" # payload

read [0x31 0x14] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/one" # topic name
"message2" # payload
51 changes: 51 additions & 0 deletions ...ty/zilla/specs/binding/mqtt/streams/network/v5/publish.retained.multiple.topic/client.rpt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
#
# Copyright 2021-2024 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 [0x31 0x14] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/one" # topic name
[0x00] # properties
"message" # payload

write [0x31 0x14] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/two" # topic name
[0x00] # properties
"message" # payload

write [0x31 0x15] # PUBLISH flags = at-most-once, retain
[0x00 0x0a] "sensor/one" # topic name
[0x00] # properties
"message2" # payload
Loading

0 comments on commit b88428e

Please sign in to comment.