fixes and formatting

Author: NickMcSweeney

Project.toml

@@ -15,7 +15,7 @@ PrecompileTools = "aea7be01-6a6a-4083-8856-8a6e6704d82a"
 PrecompileMQTT = "PrecompileTools"
 
 [compat]
-julia = "1.7"
+julia = "1.8"
 
 [extras]
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"

README.md

@@ -4,7 +4,6 @@
 [![Dev](https://img.shields.io/badge/docs-dev-blue.svg)](https://JuliaMessaging.github.io/MQTTClient.jl/dev/)
 [![Build Status](https://github.com/JuliaMessaging/MQTTClient.jl/actions/workflows/CI.yml/badge.svg?branch=main)](https://github.com/JuliaMessaging/MQTTClient.jl/actions/workflows/CI.yml?query=branch%3Amain)
 [![Coverage](https://codecov.io/gh/JuliaMessaging/MQTTClient.jl/branch/main/graph/badge.svg)](https://codecov.io/gh/JuliaMessaging/MQTTClient.jl)
-[![Coverage](https://coveralls.io/repos/github/JuliaMessaging/MQTTClient.jl/badge.svg?branch=main)](https://coveralls.io/github/JuliaMessaging/MQTTClient.jl?branch=main)
 [![Code Style: Blue](https://img.shields.io/badge/code%20style-blue-4495d1.svg)](https://github.com/invenia/BlueStyle)
 
 MQTT Client Library for Julia

docs/make.jl

@@ -1,4 +1,4 @@
-push!(LOAD_PATH,"../src/")
+push!(LOAD_PATH, "../src/")
 
 using MQTTClient
 using Documenter
@@ -21,15 +21,10 @@ makedocs(;
         "Getting Started" => "getting-started.md",
         "MQTT Interface Functions" => "interfaces.md",
         "MQTT Client" => "client.md",
-        "MQTT API" => [
-            "Client" => "api/client.md",
-            "Interfacing Functions" => "api/interface.md",
-            ],
+        "MQTT API" =>
+            ["Client" => "api/client.md", "Interfacing Functions" => "api/interface.md"],
         "Utils" => "utils.md",
     ],
 )
 
-deploydocs(;
-    repo="github.com/JuliaMessaging/MQTTClient.jl",
-    devbranch="main",
-)
+deploydocs(; repo="github.com/JuliaMessaging/MQTTClient.jl", devbranch="main")

examples/basic.jl

@@ -6,22 +6,22 @@ payload = "Hello World!"
 
 # Define the callback for receiving messages.
 function on_msg(topic, payload)
-    println("Received message topic: [", topic, "] payload: [", String(payload), "]")
+    return println("Received message topic: [", topic, "] payload: [", String(payload), "]")
 end
 
 # Instantiate a client.
-client, connection = MakeConnection(broker,1883)
+client, connection = MakeConnection(broker, 1883)
 
 connect(client, connection)
 println("connected to $client at $(connection.protocol)")
 
 # Subscribe to the topic.
-subscribe(client, topic, on_msg, qos=QOS_2)
+subscribe(client, topic, on_msg; qos=QOS_2)
 println("subscribed to $topic")
 
 sleep(0.5)
 
-publish(client, topic, payload, qos=QOS_2)
+publish(client, topic, payload; qos=QOS_2)
 println("published $payload to $topic")
 
 # Unsubscribe from the topic

ext/PrecompileMQTT.jl

@@ -14,7 +14,7 @@ using MQTTClient
     # Putting some things in `@setup_workload` instead of `@compile_workload` can reduce the size of the
     # precompile file and potentially make loading faster.
 
-    cb(t,p) = nothing
+    cb(t, p) = nothing
     topic = "foo"
     payload = "bar"
 
@@ -36,7 +36,11 @@ using MQTTClient
 
         message = MQTTClient.Message(false, UInt8(MQTTClient.QOS_0), false, topic, payload)
         optional = message.qos == 0x00 ? () : (0)
-        cmd = MQTTClient.PUBLISH | ((message.dup & 0x1) << 3) | (message.qos << 1) | message.retain
+        cmd =
+            MQTTClient.PUBLISH |
+            ((message.dup & 0x1) << 3) |
+            (message.qos << 1) |
+            message.retain
         packet = MQTTClient.Packet(cmd, [message.topic, optional..., message.payload])
         buffer = PipeBuffer()
         for i in packet.data
@@ -70,7 +74,6 @@ using MQTTClient
         MQTTClient.handle_pubrel(c, s, cmd, flags)
         p = take!(c.write_packets)
 
-
         # handle_suback
         c = MQTTClient.Client()
         s = IOBuffer()
@@ -96,23 +99,22 @@ using MQTTClient
         ## Interfaces:
         # subscribe
         c = MQTTClient.Client()
-        future = MQTTClient.subscribe_async(c, topic, cb, qos=MQTTClient.QOS_2)
+        future = MQTTClient.subscribe_async(c, topic, cb; qos=MQTTClient.QOS_2)
 
         # unsubscribe
-        c =  MQTTClient.Client()
+        c = MQTTClient.Client()
         insert!(c.on_msg, topic, cb)
         @atomicswap c.last_id = 0x0
         future = unsubscribe_async(c, topic)
 
-
         ## TCP Basic Run
         server = MQTTClient.MockMQTTBroker(ip"127.0.0.1", 1889)
         client, conn = MakeConnection(ip"127.0.0.1", 1889)
 
         connect(client, conn)
 
         subscribe(client, "foo/bar", cb)
-        publish(client, "bar/foo", qos=QOS_2)
+        publish(client, "bar/foo", "baz"; qos=QOS_2)
         unsubscribe(client, "foo/bar")
 
         disconnect(client)
@@ -125,7 +127,7 @@ using MQTTClient
         connect(client, conn)
 
         subscribe(client, "foo/bar", cb)
-        publish(client, "bar/foo")
+        publish(client, "bar/foo", "baz")
         unsubscribe(client, "foo/bar")
 
         disconnect(client)

src/MQTTClient.jl

@@ -8,7 +8,6 @@ import Base: ReentrantLock, lock, unlock, convert, PipeEndpoint, fetch, show
 import Base: @atomic, @atomicreplace, @atomicswap, Ref, RefValue, isready
 using Base.Threads
 
-
 include("utils.jl")
 include("internals.jl")
 include("topic.jl")
@@ -17,8 +16,7 @@ include("connection.jl")
 include("handlers.jl")
 include("interface.jl")
 
-export
-    MakeConnection,
+export MakeConnection,
     Configuration,
     Client,
     Connection,

src/client.jl

@@ -1,28 +1,30 @@
 """
     Client
 
-The MQTT client in Julia facilitates communication between a device and an MQTT broker over a network. 
-It manages connections, message handling, and maintains the state of communication. 
-The client operates through three main loops: the read loop listens for incoming messages from the broker and processes them using designated handlers; 
-the write loop sends packets to the broker from a queue, ensuring thread safety with a socket lock; 
-and the keep-alive loop periodically sends ping requests to the broker to maintain the connection and detect disconnections. 
+The MQTT client in Julia facilitates communication between a device and an MQTT broker over a network.
+It manages connections, message handling, and maintains the state of communication.
+The client operates through three main loops: the read loop listens for incoming messages from the broker and processes them using designated handlers;
+the write loop sends packets to the broker from a queue, ensuring thread safety with a socket lock;
+and the keep-alive loop periodically sends ping requests to the broker to maintain the connection and detect disconnections.
 This client uses atomic operations to ensure thread safety for shared variables and supports asynchronous task management for efficient, non-blocking operations.
 
 # Fields
-- `state::UInt8`: client state.
-- `on_msg::TrieNode`: A trie mapping topics to callback functions.
-- `keep_alive::UInt16`: The keep-alive time in seconds.
-- `last_id::UInt16`: The last packet identifier used.
-- `in_flight::Dict{UInt16, Future}`: A dictionary mapping packet identifiers to futures.
-- `write_packets::AbstractChannel`: A channel for writing packets.
-- `socket`: The socket used for communication with the broker.
-- `socket_lock`: A lock for synchronizing access to the socket.
-- `ping_timeout::UInt64`: The ping timeout in seconds.
-- `ping_outstanding::Atomic{UInt8}`: An atomic counter for the number of outstanding ping requests.
-- `last_sent::Atomic{Float64}`: An atomic float representing the timestamp of the last sent packet.
-- `last_received::Atomic{Float64}`: An atomic float representing the timestamp of the last received packet.
+
+  - `state::UInt8`: client state.
+  - `on_msg::TrieNode`: A trie mapping topics to callback functions.
+  - `keep_alive::UInt16`: The keep-alive time in seconds.
+  - `last_id::UInt16`: The last packet identifier used.
+  - `in_flight::Dict{UInt16, Future}`: A dictionary mapping packet identifiers to futures.
+  - `write_packets::AbstractChannel`: A channel for writing packets.
+  - `socket`: The socket used for communication with the broker.
+  - `socket_lock`: A lock for synchronizing access to the socket.
+  - `ping_timeout::UInt64`: The ping timeout in seconds.
+  - `ping_outstanding::Atomic{UInt8}`: An atomic counter for the number of outstanding ping requests.
+  - `last_sent::Atomic{Float64}`: An atomic float representing the timestamp of the last sent packet.
+  - `last_received::Atomic{Float64}`: An atomic float representing the timestamp of the last received packet.
 
 # Constructor
+
 `Client(ping_timeout::UInt64=UInt64(60))` constructs a new `Client` object with the specified ping timeout (default: 60 seconds).
 """
 mutable struct Client <: AbstractConfigElement
@@ -50,26 +52,27 @@ mutable struct Client <: AbstractConfigElement
     read_task::Task
     keep_alive_task::Task
 
-    Client(ping_timeout::UInt64 = UInt64(60)) = new(
-        0x00,
-        TrieNode(),
-        0x0000,
-        0x0000,
-        Dict{UInt16,Future}(),
-        Channel{Packet}(typemax(Int64)),
-        IOBuffer(),
-        ReentrantLock(),
-        ping_timeout,
-        0x00,
-        0.0,
-        0.0,
-        Task(nothing),
-        Task(nothing),
-        Task(nothing),
-    )
+    function Client(ping_timeout::UInt64=UInt64(60))
+        return new(
+            0x00,
+            TrieNode(),
+            0x0000,
+            0x0000,
+            Dict{UInt16,Future}(),
+            Channel{Packet}(typemax(Int64)),
+            IOBuffer(),
+            ReentrantLock(),
+            ping_timeout,
+            0x00,
+            0.0,
+            0.0,
+            Task(nothing),
+            Task(nothing),
+            Task(nothing),
+        )
+    end
 end
 
-
 function write_loop(client::Client)::UInt8
     try
         while !isclosed(client)
@@ -115,7 +118,6 @@ function write_loop(client::Client)::UInt8
     end
 end
 
-
 function read_loop(client::Client)::UInt8
     try
         while !isclosed(client)
@@ -157,18 +159,17 @@ function read_loop(client::Client)::UInt8
     end
 end
 
-
 function keep_alive_loop(client::Client)::UInt8
     ping_sent = time()
 
     # TODO: improve, this causes reconnect to take ~1 second. is there a way to interupt?
     check_interval = 1
-     
-    timer = Timer(0, interval = check_interval)
+
+    timer = Timer(0; interval=check_interval)
 
     while !isclosed(client)
         if time() - @atomic(client.last_sent) >= client.keep_alive ||
-           time() - @atomic(client.last_received) >= client.keep_alive
+            time() - @atomic(client.last_received) >= client.keep_alive
             if @atomic(client.ping_outstanding) == 0x0
                 @atomicswap client.ping_outstanding = 0x1
                 try
@@ -191,7 +192,7 @@ function keep_alive_loop(client::Client)::UInt8
         end
 
         if @atomic(client.ping_outstanding) == 1 &&
-           time() - ping_sent >= client.ping_timeout
+            time() - ping_sent >= client.ping_timeout
             try # No pingresp received
                 disconnect(client)
                 break
@@ -222,18 +223,20 @@ end
 
 # write packet to mqtt broker
 function write_packet(client::Client, cmd::UInt8, data...)
-    put!(client.write_packets, Packet(cmd, data))
+    return put!(client.write_packets, Packet(cmd, data))
 end
 
 isready(client::Client)::Bool = client.state == 0x00
 isconnected(client::Client)::Bool = client.state == 0x01
 isclosed(client::Client)::Bool = client.state >= 0x02
 iserror(client::Client)::Bool = client.state == 0x03
 
-show(io::IO, client::Client) = print(
-    io,
-    "MQTTClient[state: $(get(CLIENT_STATE, client.state, :unknown)), read_loop: $(taskstatus(client.read_task)), write_loop: $(taskstatus(client.write_task)), keep_alive: $(taskstatus(client.keep_alive_task))]\n$(show_tree(client.on_msg))",
-)
+function show(io::IO, client::Client)
+    return print(
+        io,
+        "MQTTClient[state: $(get(CLIENT_STATE, client.state, :unknown)), read_loop: $(taskstatus(client.read_task)), write_loop: $(taskstatus(client.write_task)), keep_alive: $(taskstatus(client.keep_alive_task))]\n$(show_tree(client.on_msg))",
+    )
+end
 
 fetch(client::Client)::Tuple{UInt8,UInt8,UInt8} = begin
     try