Another fix for connection interrupted/resumed. Add test. (#27)

Author: Octogonapus

Project.toml

@@ -1,6 +1,6 @@
 name = "AWSCRT"
 uuid = "df31ea59-17a4-4ebd-9d69-4f45266dc2c7"
-version = "0.4.0"
+version = "0.4.1"
 
 [deps]
 CountDownLatches = "621fb831-fdad-4fff-93ac-1af7b7ed19e3"

src/AWSMQTT.jl

@@ -222,7 +222,7 @@ function _c_on_connection_interrupted(
 
     data = Base.unsafe_pointer_to_objref(userdata)::_OnConnectionInterruptedUserData
     try
-        put!(data.ch, _OnConnectionInterruptedEvent(data.callback, error_code))
+        put!(data.ch, _OnConnectionInterruptedEvent(data.callback, data.conn, error_code))
     catch ex
         if ex isa InvalidStateException && ex.state == :closed
         else
@@ -265,7 +265,12 @@ function _c_on_connection_resumed(
     try
         put!(
             data.ch,
-            _OnConnectionResumedEvent(data.callback, aws_mqtt_connect_return_code(return_code), session_present != 0),
+            _OnConnectionResumedEvent(
+                data.callback,
+                data.conn,
+                aws_mqtt_connect_return_code(return_code),
+                session_present != 0,
+            ),
         )
     catch ex
         if ex isa InvalidStateException && ex.state == :closed

test/interrupt_connection.jl

@@ -0,0 +1,38 @@
+@testset "interrupt and resume connection" begin
+    topic1 = "test-topic-$(Random.randstring(6))"
+    payload1 = Random.randstring(48)
+    client_id1 = random_client_id()
+    @show topic1 payload1 client_id1
+
+    client = MQTTClient(new_tls_ctx())
+    connection = MQTTConnection(client)
+
+    interruptions = Threads.Atomic{Int}(0)
+    resumes = Threads.Atomic{Int}(0)
+
+    task = connect(
+        connection,
+        ENV["ENDPOINT"],
+        8883,
+        client_id1;
+        will = Will(topic1, AWS_MQTT_QOS_AT_LEAST_ONCE, "The client has gone offline!", false),
+        on_connection_interrupted = (conn, error_code) -> begin
+            Threads.atomic_add!(interruptions, 1)
+            @warn "connection interrupted" error_code
+        end,
+        on_connection_resumed = (conn, return_code, session_present) -> begin
+            Threads.atomic_add!(resumes, 1)
+            @info "connection resumed" return_code session_present
+        end,
+    )
+    @test fetch(task) == Dict(:session_present => false)
+
+    # kill the MQTT connection
+    run(`sudo ss -K dport = 8883`)
+
+    wait_for(() -> resumes[] > 0, Timer(30))
+    @test interruptions[] == 1
+    @test resumes[] == 1
+
+    disconnect(connection)
+end

test/runtests.jl

@@ -22,6 +22,11 @@ end)
             Aqua.test_all(AWSCRT, ambiguities = false)
             Aqua.test_ambiguities(AWSCRT)
         end
+        @testset "interrupt_connection.jl" begin
+            # running this in parallel will kill the connections used by other tests so don't do that
+            @info "Starting interrupt_connection.jl"
+            include("interrupt_connection.jl")
+        end
     end
     @testset "mqtt_test.jl" begin
         @info "Starting mqtt_test.jl"