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
checkpoint
  • Loading branch information
@bmaidics
bmaidics committed Nov 23, 2023
commit b5ed1ab5931696f0d95e9813b000af5046ec7884
54 changes: 53 additions & 1 deletion ...t/streams/application/subscribe.receive.message.overlapping.wildcard.mixed.qos/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,30 @@ read ${mqtt:session()
.build()}


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"))
.subscribe()
.clientId("client")
.filter("sensor/+/#", 1, "AT_MOST_ONCE", "SEND_RETAINED")
.build()
.build()}

connected

write advise zilla:flush ${mqtt:flushEx()
.typeId(zilla:id("mqtt"))
.subscribe()
.filter("sensor/+/#", 1, "AT_MOST_ONCE", "SEND_RETAINED")
.filter("sensor/+/1", 2, "AT_MOST_ONCE", "SEND_RETAINED")
.build()
.build()}


connect await RECEIVED_SESSION_STATE
"zilla://streams/app0"
option zilla:window 8192
Expand All @@ -95,7 +119,7 @@ write advise zilla:flush ${mqtt:flushEx()
.typeId(zilla:id("mqtt"))
.subscribe()
.filter("sensor/+/#", 1, "AT_LEAST_ONCE", "SEND_RETAINED")
.filter("sensor/+/1", 2, "EXACTLY_ONCE", "SEND_RETAINED")
.filter("sensor/+/1", 2, "AT_LEAST_ONCE", "SEND_RETAINED")
.build()
.build()}

Expand All @@ -110,7 +134,35 @@ read zilla:data.ext ${mqtt:matchDataEx()
.format("TEXT")
.build()
.build()}
read "message"


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"))
.subscribe()
.clientId("client")
.filter("sensor/+/1", 2, "EXACTLY_ONCE", "SEND_RETAINED")
.build()
.build()}

connected

read zilla:data.ext ${mqtt:matchDataEx()
.typeId(zilla:id("mqtt"))
.subscribe()
.topic("sensor/one/1")
.packetId(1)
.qos("EXACTLY_ONCE")
.subscriptionId(1)
.subscriptionId(2)
.format("TEXT")
.build()
.build()}
read "message"

write advise zilla:flush ${mqtt:flushEx()
Expand Down
14 changes: 14 additions & 0 deletions .../mqtt/streams/network/subscribe.receive.message.overlapping.wildcard.mixed.qos/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,15 @@ write [0x90 0x04] # SUBACK
[0x00] # properties = none
[0x02] # reason code = qos2

write [0x32 0x1e] # PUBLISH, qos = 1
[0x00 0x0c] "sensor/one/1" # topic name
[0x00 0x01] # packet id = 1
[0x06] # properties
[0x0b 0x01] # subscription id = 1
[0x0b 0x02] # subscription id = 2
[0x01 0x01] # format = utf-8
"message" # payload

write [0x34 0x1e] # PUBLISH
[0x00 0x0c] "sensor/one/1" # topic name
[0x00 0x01] # packet id = 1
Expand All @@ -70,6 +79,11 @@ write [0x34 0x1e] # PUBLISH
[0x01 0x01] # format = utf-8
"message" # payload

read [0x40 0x04] # PUBACK
[0x00 0x01] # packet id = 1
[0x00] # reason code
[0x00] # properties

read [0x50 0x04] # PUBREC
[0x00 0x01] # packet id = 1
[0x00] # reason code
Expand Down
1 change: 1 addition & 0 deletions ...c/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SubscribeIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,6 +350,7 @@ public void shouldReplayRetainedQos2() throws Exception
{
k3po.finish();
}

@Test
@Specification({
"${app}/subscribe.receive.message.overlapping.wildcard.mixed.qos/client",
Expand Down