Skip to content

Implement QoS 1 and QoS 2 #589

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 24 commits into from
Dec 4, 2023
Merged

Implement QoS 1 and QoS 2 #589

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
5d4b512
Checkpoint
bmaidics Nov 23, 2023
b5ed1ab
checkpoint
bmaidics Nov 23, 2023
3a398c5
mqtt-kafka qos1,2 start
bmaidics Nov 24, 2023
99b0716
Checkpoint
bmaidics Nov 27, 2023
647eb92
checkpoint
bmaidics Nov 27, 2023
c39621d
MQTT binding done?
bmaidics Nov 28, 2023
a66e0bc
checkpoint
bmaidics Nov 29, 2023
1263071
Change to multiple subscribe stream for the same filter
bmaidics Nov 30, 2023
27c702e
some mqtt-kafka scenarios
bmaidics Nov 30, 2023
60e200d
checkpoint
bmaidics Nov 30, 2023
e092d6b
checkpoint
bmaidics Dec 1, 2023
51588c7
checkpoint
bmaidics Dec 2, 2023
77794b0
checkpoiint
bmaidics Dec 2, 2023
1b610f7
Merge remote-tracking branch 'upstream/feature/mqtt-kafka' into qos12
bmaidics Dec 3, 2023
3aa7b53
Add 3.1.1 qos1 and 2 support
bmaidics Dec 3, 2023
cde3a6a
Add consumer flush correlationId
bmaidics Nov 30, 2023
895ec03
End-to-end test caused fixes
bmaidics Dec 3, 2023
6b52a4c
Use numeric QoS for zilla:qos header value in scripts
jfallows Dec 4, 2023
8c26fba
Reset qos to null before visiting kafka headers
jfallows Dec 4, 2023
7f3f273
Use MutableInteger instead of AtomicInteger, and retrieve primitive i…
jfallows Dec 4, 2023
86ed165
qosMax is now 2
jfallows Dec 4, 2023
890fee5
Merge branch 'feature/mqtt-kafka' into qos12
akrambek Dec 4, 2023
deff491
Catch up merge
akrambek Dec 4, 2023
3079078
Update matching criteria
akrambek Dec 4, 2023
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
Prev Previous commit
Next Next commit
some mqtt-kafka scenarios
  • Loading branch information
@bmaidics
bmaidics committed Dec 2, 2023
commit 27c702ed53517b940bb3ad1011c635fe6191fa09
98 changes: 98 additions & 0 deletions ...binding/mqtt/kafka/streams/kafka/subscribe.receive.message.qos1.published.qos2/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
#
# Copyright 2021-2023 Aklivity Inc
#
# Licensed under the Aklivity Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# https://www.aklivity.io/aklivity-community-license/
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#

connect "zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("mqtt-messages")
.filter()
.headers("zilla:filter")
.sequence("sensor")
.sequence("one")
.build()
.build()
.headerNot("zilla:qos", "AT_LEAST_ONCE")
.headerNot("zilla:qos", "EXACTLY_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected


connect "zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("mqtt-messages")
.groupId("client")
.filter()
.headers("zilla:filter")
.sequence("sensor")
.sequence("one")
.build()
.build()
.headerNot("zilla:qos", "AT_MOST_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected


read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.fetch()
.timestamp(kafka:timestamp())
.filters(1)
.partition(0, 1, 2)
.progress(0, 2)
.progress(1, 1)
.key("sensor/one")
.header("zilla:filter", "sensor")
.header("zilla:filter", "one")
.header("zilla:local", "client")
.header("zilla:format", "TEXT")
.header("zilla:qos", "EXACTLY_ONCE")
.build()
.build()
.build()}
read "message"

write advise zilla:flush ${kafka:flushEx()
.typeId(zilla:id("kafka"))
.merged()
.consumer()
.partition(0, 2)
.build()
.build()}

read advised zilla:flush ${kafka:matchFlushEx()
.typeId(zilla:id("kafka"))
.merged()
.consumer()
.partition(0, 2)
.build()
.build()}
99 changes: 99 additions & 0 deletions ...binding/mqtt/kafka/streams/kafka/subscribe.receive.message.qos1.published.qos2/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
#
# Copyright 2021-2023 Aklivity Inc
#
# Licensed under the Aklivity Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# https://www.aklivity.io/aklivity-community-license/
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#

accept "zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

accepted

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("mqtt-messages")
.filter()
.headers("zilla:filter")
.sequence("sensor")
.sequence("one")
.build()
.build()
.headerNot("zilla:qos", "AT_LEAST_ONCE")
.headerNot("zilla:qos", "EXACTLY_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected


accepted

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("mqtt-messages")
.groupId("client")
.filter()
.headers("zilla:filter")
.sequence("sensor")
.sequence("one")
.build()
.build()
.headerNot("zilla:qos", "AT_MOST_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.fetch()
.timestamp(kafka:timestamp())
.filters(1)
.partition(0, 1, 2)
.progress(0, 2)
.progress(1, 1)
.key("sensor/one")
.header("zilla:filter", "sensor")
.header("zilla:filter", "one")
.header("zilla:local", "client")
.header("zilla:format", "TEXT")
.header("zilla:qos", "EXACTLY_ONCE")
.build()
.build()
.build()}
write "message"
write flush

read advised zilla:flush ${kafka:matchFlushEx()
.typeId(zilla:id("kafka"))
.merged()
.consumer()
.partition(0, 2)
.build()
.build()}

write advise zilla:flush ${kafka:flushEx()
.typeId(zilla:id("kafka"))
.merged()
.consumer()
.partition(0, 2)
.build()
.build()}

52 changes: 40 additions & 12 deletions ...ty/zilla/specs/binding/mqtt/kafka/streams/kafka/subscribe.receive.message.qos1/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,29 +28,57 @@ write zilla:begin.ext ${kafka:beginEx()
.sequence("one")
.build()
.build()
.header("zilla:qos", "AT_LEAST_ONCE")
.headerNot("zilla:qos", "AT_LEAST_ONCE")
.headerNot("zilla:qos", "EXACTLY_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected

read zilla:data.ext ${kafka:matchDataEx()

connect "zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.filters(1)
.partition(0, 1, 2)
.progress(0, 2)
.progress(1, 1)
.key("sensor/one")
.header("zilla:filter", "sensor")
.header("zilla:filter", "one")
.header("zilla:local", "client")
.header("zilla:format", "TEXT")
.header("zilla:qos", "AT_LEAST_ONCE")
.capabilities("FETCH_ONLY")
.topic("mqtt-messages")
.groupId("client")
.filter()
.headers("zilla:filter")
.sequence("sensor")
.sequence("one")
.build()
.build()
.headerNot("zilla:qos", "AT_MOST_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected


read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.fetch()
.timestamp(kafka:timestamp())
.filters(1)
.partition(0, 1, 2)
.progress(0, 2)
.progress(1, 1)
.key("sensor/one")
.header("zilla:filter", "sensor")
.header("zilla:filter", "one")
.header("zilla:local", "client")
.header("zilla:format", "TEXT")
.header("zilla:qos", "AT_LEAST_ONCE")
.build()
.build()
.build()}
read "message"

write advise zilla:flush ${kafka:flushEx()
Expand Down
24 changes: 23 additions & 1 deletion ...ty/zilla/specs/binding/mqtt/kafka/streams/kafka/subscribe.receive.message.qos1/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,28 @@ accept "zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

accepted

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("mqtt-messages")
.filter()
.headers("zilla:filter")
.sequence("sensor")
.sequence("one")
.build()
.build()
.headerNot("zilla:qos", "AT_LEAST_ONCE")
.headerNot("zilla:qos", "EXACTLY_ONCE")
.evaluation("EAGER")
.build()
.build()}

connected


accepted

read zilla:begin.ext ${kafka:matchBeginEx()
Expand All @@ -31,7 +53,7 @@ read zilla:begin.ext ${kafka:matchBeginEx()
.sequence("one")
.build()
.build()
.header("zilla:qos", "AT_LEAST_ONCE")
.headerNot("zilla:qos", "AT_MOST_ONCE")
.evaluation("EAGER")
.build()
.build()}
Expand Down
Loading