MQTT-support #766

kickster97 · 2024-08-26T14:50:13Z

WHAT is this pull request doing?

An MQTT-session inherits AMQP::Queue and extends with MQTT functionality.
MQTT exchange(topic) will only accept activity from an MQTT client, that will be responsible for routing messages to the MQTT-session.
The mqtt exchange handles MQTT routing keys etc. without needing to do conversions.

NOTE
The first version of MQTT Support does not support binding an AMQP exchange to and MQTT exchange. (Undefined behaviour if done)
In a separate PR we need to extract the AMQP exchange types into the AMQP namespace and in turn be able to differentiate between an AMQP::Destination and an MQTT::Destiantion when calling bind.
With that in place we will be able to safely do exchange to exchange bindings.

Docs for the website

HOW can this pull request be tested?

Specs have been migrated from Myra and fully cover the MQTT protocol, run with crystal spec.
Connect with mqtt client library of your choice and test your usage against this branch

static/js/connections.js

src/lavinmq/vhost.cr

src/lavinmq/mqtt/client.cr

src/lavinmq/mqtt/broker.cr

spuun · 2024-09-24T07:52:42Z

I think you can (and maybe even should) build clean session by reusing the auto delete feature of queues.

spec/mqtt/integrations/connect_spec.cr

spuun · 2024-10-12T06:33:56Z

About #803 (comment)

That connect spec passes, but for the wrong reason. Or well, the socket is closed but not because it receives the wrong packet type, but because of the bug that #803 will fix. When that bug is fixed i still think that the spec will pass, but for the right reason.

I guess the spec could be changed to send a PUBLISH instead (the packet must be greater than 8 byte to not trigger the bug).

spuun · 2024-10-12T06:46:11Z

After giving it a second thought I've concluded that we should change the spec to use some other packet type.

kickster97 · 2024-10-14T07:53:16Z

After giving it a second thought I've concluded that we should change the spec to use some other packet type.

ok I will change the spec then! thanks :)

carlhoerberg · 2024-10-16T16:31:59Z

Often it's more efficient to yield than to capture and then call a block.

src/lavinmq/mqtt/retain_store.cr

src/lavinmq/mqtt/broker.cr

src/lavinmq/exchange/mqtt.cr

src/lavinmq/mqtt/client.cr

src/lavinmq/vhost.cr

src/lavinmq/prefix_validation.cr

* deliver packet not msg from session * it's already a topic, no convert needed * fixes

carlhoerberg

Partial review

carlhoerberg · 2024-12-05T00:15:57Z

src/lavinmq/clustering/follower.cr

@@ -73,8 +73,10 @@ module LavinMQ
      end

      private def read_ack(socket = @socket) : Int64
+        # Sometimes when running clustering_spec len is greater than sent_bytes. Causing lag_in_bytes to be negative.


how is that possible? And should this comment really be here?

src/lavinmq/clustering/follower.cr

src/lavinmq/config.cr

carlhoerberg · 2024-12-05T00:22:44Z

src/lavinmq/http/handler/websocket.cr

@@ -11,7 +11,7 @@ module LavinMQ
                         Socket::IPAddress.new("127.0.0.1", 0) # Fake when UNIXAddress
        connection_info = ConnectionInfo.new(remote_address, local_address)
        io = WebSocketIO.new(ws)
-        spawn amqp_server.handle_connection(io, connection_info), name: "HandleWSconnection #{remote_address}"
+        spawn amqp_server.handle_connection(io, connection_info, Server::Protocol::AMQP), name: "HandleWSconnection #{remote_address}"


maybe at some point we should support MQTT over websockets too?

src/lavinmq/mqtt/retain_store.cr

carlhoerberg · 2024-12-05T01:40:24Z

src/lavinmq/mqtt/retain_store.cr

+        end
+      end
+
+      def each(subscription : String, &block : String, Bytes -> Nil) : Nil


in the only location we use this method we convert the body to a String, which will raise if there is binary data that can't be parsed as utf8, but also, why don't we just return the File/IO object?

Suggested change

def each(subscription : String, &block : String, Bytes -> Nil) : Nil

def each(subscription : String, &block : String, IO -> Nil) : Nil

carlhoerberg · 2024-12-05T01:43:06Z

src/lavinmq/mqtt/retain_store.cr

+        hash = @hasher.hexfinal
+        @hasher.reset
+        "#{hash}#{MESSAGE_FILE_SUFFIX}"


Suggested change

hash = @hasher.hexfinal

@hasher.reset

"#{hash}#{MESSAGE_FILE_SUFFIX}"

"#{@hasher.hexfinal}#{MESSAGE_FILE_SUFFIX}"

ensure

@hasher.reset

carlhoerberg · 2024-12-05T01:44:47Z

src/lavinmq/mqtt/connection_factory.cr

+      def initialize(@users : UserStore,
+                     @vhosts : VHostStore,
+                     replicator : Clustering::Replicator)
+        @brokers = Brokers.new(@vhosts, replicator)


could this be initialized in the Server instead? Along with VHostStore?

carlhoerberg · 2024-12-05T01:45:23Z

src/lavinmq/mqtt/client.cr

+      end
+
+      def close(reason = "")
+        @log.trace { "Client#close" }


is this really a trace? Rather a debug i think

Suggested change

@log.trace { "Client#close" }

@log.debug { "Client#close" }

spuun · 2024-12-12T07:09:28Z

src/lavinmq/http/controller/bindings.cr

+            elsif q.is_a?(LavinMQ::MQTT::Session)
+              access_refused(context, "Not allowed to bind to an MQTT session")
+            elsif e.is_a?(LavinMQ::MQTT::Exchange)
+              access_refused(context, "Not allowed to bind to the default MQTT exchange")


How about handling this with different overloads of Exchange#bind?

We're already doing that in MQTT::Exchange to prevent it from being bound to AMQP queues:
https://github.com/cloudamqp/lavinmq/blob/mqtt-poc/src/lavinmq/mqtt/exchange.cr#L90-L122

By doing that you get the same validation for AMQP clients and mgmt ui.

We worked a bit on this and turns out to be trickier than we thought. I put this in the description

The first version of MQTT Support does not support binding an AMQP exchange to and MQTT exchange. (Undefined behaviour if done)
In a separate PR we need to extract the AMQP exchange types into the AMQP namespace and in turn be able to differentiate between an AMQP::Destination and an MQTT::Destiantion when calling bind.
With that in place we will be able to safely do exchange to exchange bindings.

kickster97 force-pushed the mqtt-poc branch from af54f0b to ee30279 Compare August 27, 2024 07:43

spuun reviewed Sep 5, 2024

View reviewed changes

static/js/connections.js Outdated Show resolved Hide resolved

spuun reviewed Sep 5, 2024

View reviewed changes

static/js/connections.js Outdated Show resolved Hide resolved

spuun reviewed Sep 16, 2024

View reviewed changes

src/lavinmq/vhost.cr Outdated Show resolved Hide resolved

spuun reviewed Sep 16, 2024

View reviewed changes

src/lavinmq/vhost.cr Outdated Show resolved Hide resolved

spuun reviewed Sep 16, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Outdated Show resolved Hide resolved

spuun reviewed Sep 16, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Show resolved Hide resolved

spuun reviewed Sep 16, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Outdated Show resolved Hide resolved

kickster97 force-pushed the mqtt-poc branch from cb77ee8 to 7130ee3 Compare September 23, 2024 08:14

spuun reviewed Sep 24, 2024

View reviewed changes

src/lavinmq/mqtt/broker.cr Outdated Show resolved Hide resolved

kickster97 force-pushed the mqtt-poc branch 2 times, most recently from 787840d to 4e747f2 Compare September 27, 2024 07:19

snichme force-pushed the mqtt-poc branch from bdb2d03 to c4dd8ab Compare October 10, 2024 10:03

spuun reviewed Oct 10, 2024

View reviewed changes

spec/mqtt/integrations/connect_spec.cr Outdated Show resolved Hide resolved

kickster97 force-pushed the mqtt-poc branch from c9100cf to bb4394c Compare October 14, 2024 08:45

spuun reviewed Oct 18, 2024

View reviewed changes

src/lavinmq/mqtt/retain_store.cr Outdated Show resolved Hide resolved

kickster97 force-pushed the mqtt-poc branch from da94d00 to 7cfbfa1 Compare October 21, 2024 11:13

spuun reviewed Oct 21, 2024

View reviewed changes

src/lavinmq/mqtt/broker.cr Outdated Show resolved Hide resolved

spuun reviewed Oct 21, 2024

View reviewed changes

src/lavinmq/exchange/mqtt.cr Outdated Show resolved Hide resolved

spuun reviewed Oct 22, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Outdated Show resolved Hide resolved

spuun reviewed Oct 23, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Outdated Show resolved Hide resolved

spuun reviewed Oct 23, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Outdated Show resolved Hide resolved

spuun reviewed Oct 23, 2024

View reviewed changes

src/lavinmq/mqtt/client.cr Outdated Show resolved Hide resolved

spuun reviewed Oct 30, 2024

View reviewed changes

src/lavinmq/vhost.cr Outdated Show resolved Hide resolved

kickster97 commented Oct 31, 2024

View reviewed changes

src/lavinmq/prefix_validation.cr Outdated Show resolved Hide resolved

kickster97 and others added 15 commits November 21, 2024 10:20

handle unexpected close from client

61a15e2

connection_at for mqtt connections

0525c76

deliver packet not msg from session (#843)

131043d

* deliver packet not msg from session * it's already a topic, no convert needed * fixes

truncate the previous content before you retain a message

bf18a56

safely overwrite retained messages

9410a13

Cant use constant as key in NamedTuple

7ac09ab

merge solutions for retain store

cabf445

general fixup after comments

b29bc61

format

d19e4d1

set flaky spec to pending

1b87d0e

just a test

1af2f13

just a test

9345b85

tmp: debug clustering_spec

f097cc9

finalize clustering spec

69871d2

use instance var for index file name

1dc8cb6

kickster97 force-pushed the mqtt-poc branch from deff3b8 to 1dc8cb6 Compare November 21, 2024 09:26

rescue argumenterror in deliver loop

b6b1059

kickster97 requested review from spuun, snichme and carlhoerberg December 2, 2024 11:44

kickster97 and others added 3 commits December 2, 2024 15:09

format

c67089d

Fix mqtt unix listener (and some logging)

53854ce

Prevent Channel::Closed error from being raised

88a2f28

carlhoerberg requested changes Dec 5, 2024

View reviewed changes

kickster97 added 4 commits December 10, 2024 13:03

exception handling for mqtt default bindings

7eecf47

only allow selected policies to be applied for mqtt session

d841337

handle qos 2 at build_packet

06162d9

dont allow amqp queues or exchanges to bind to mqtt queues or exchanges

56a71de

spuun reviewed Dec 12, 2024

View reviewed changes

forbidden to bind AMQP excahnges to the MQTT Session

95be31f

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MQTT-support #766

MQTT-support #766

kickster97 commented Aug 26, 2024 •

edited

Loading

spuun commented Sep 24, 2024

spuun commented Oct 12, 2024

spuun commented Oct 12, 2024

kickster97 commented Oct 14, 2024

carlhoerberg commented Oct 16, 2024

carlhoerberg left a comment

carlhoerberg Dec 5, 2024

carlhoerberg Dec 5, 2024

carlhoerberg Dec 5, 2024

carlhoerberg Dec 5, 2024

carlhoerberg Dec 5, 2024

carlhoerberg Dec 5, 2024

spuun Dec 12, 2024

kickster97 Dec 12, 2024

	def each(subscription : String, &block : String, Bytes -> Nil) : Nil
	def each(subscription : String, &block : String, IO -> Nil) : Nil

MQTT-support #766

Are you sure you want to change the base?

MQTT-support #766

Conversation

kickster97 commented Aug 26, 2024 • edited Loading

WHAT is this pull request doing?

HOW can this pull request be tested?

spuun commented Sep 24, 2024

spuun commented Oct 12, 2024

spuun commented Oct 12, 2024

kickster97 commented Oct 14, 2024

carlhoerberg commented Oct 16, 2024

carlhoerberg left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kickster97 commented Aug 26, 2024 •

edited

Loading