Skip to content

Fix MQTT binding selecting incorrect publish stream #1328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jfallows merged 1 commit into from
Nov 13, 2024
Merged

Fix MQTT binding selecting incorrect publish stream #1328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...c/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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