Direct subscribe for embedded use case

[mdzio]

Great project.

I would like to run the Mochi embedded. For a integration of my application, there is the possibility to publish messages directly via server.Publish. Now I would also like to subscribe to topics directly without going via the TCP/IP stack, for example.

I have not found an API function for this. Does this exist or is there another possibility?

[bkupidura]

@mdzio any reason why not to use hooks system? Can you describe your usecase little more?

[dgduncan]

@mdzio Please let me know if I misunderstand, but I believe your question is that is there any way to subscribe to topics in a way that is similar to the Publish method via the embedded broker that bypasses the networking layers and subscribes directly at the broker level.

In short, the answer is no. To join in on what @bkupidura said, the simplest way to do this is by utilizing the hooks system and listening to OnPublish or OnPublished and filtering out topics you are not interested in. In essence this provides exactly what you are looking for while bypassing the network layers.

[mdzio]

My use case is the CCU-Jack. I am the main developer there. The CCU-Jack offers an open communication interface based on MQTT for the HomeMatic smart home system from eQ3.

The SurgeMQ MQTT-Server that I continue to maintain is currently embedded there. However, it only supports MQTT version 3.1.1. All data point changes of the system are internally published on this, but also topics are internally subscribed to in order to be able to set data points via MQTT.

I could certainly also use the OnPublish or OnPublished hooks, but then I would have to implement the topic matching again myself, which is actually already implemented in Mochi.

[dgduncan]

For curiosity, is the desired function signature something akin to server.Subscribe([]string topics) (*Client, packets.Packet)? Either that or some type of channel that could return information on the topic, client, and packets?

[mdzio]

For applications, the following API would be very nice:

	subscription1 := server.Subscribe("a/b/+/c", func(packet packets.Packet) {
		// handle packet
	})
	// function gets called for packages for which the topic filter applies
	// ...
	subscription2 := server.Subscribe("d/#", func(packet packets.Packet) {
		// handle packet
	})
	// ...
	subscription1.Unsubscribe()
	// ...
	subscription2.Unsubscribe()

[mochi-co]

@mdzio This is a nice pattern. I'm fairly certain I can implement this using the existing hooks framework (and possibly some new internal hooks), I'll have a think about what it requires.

[mochi-co]

I've experimented with a few ideas for this but it's a little bit more complex than I originally anticipated. However, it's my current active task so there will be a solution - I just haven't found one which 'feels' correct as yet.

[mochi-co]

@mdzio A proposal is available here: #283

[thedevop]

@mochi-co , I had similar need as @mdzio. In addition to filtering the topics of interest, I also needed to extract information from the topics. For example: the topic is in the form of some/prefix/{floor}/{room}/some/suffix, if the topic is a/b/12/34/c/d, the variable floor and room would be 12 and 34 respectively. I suspect when filter is needed, some info extraction would also be need. Although it still doesn't interact with any existing internal code (which is beneficial in my use case), but the expense of an additional matching system becomes "free".

I implemented this based on a modified version go-chi radix tree, so it supports named, regex and wildcard parameters to extract info from the topics. It is used in the OnPublish hook. This is supporting tens of millions concurrent clients, with each node (4 vCPU, ~30% utilization for entire app which embeds mochi-mqtt) supporting ~200K clients (every client sends ~2KB messages every minute). The limit is not due to mochi-mqtt, but some external constraints.

If you're interested in re-exploring the hook option, I would be glad to contribute that code.

[dgduncan]

For my own curiosity what are those constraints? Are these TCP connection limits or something else?

[thedevop]

Unfortunately, I can't go into too much details. But it's a constraint below the OS, and effectively it limits max concurrent TCP connections.

[mochi-co]

@thedevop This sounds very interesting. I think it's going to depend a lot on what people's use cases are - i.e. do you just need to intercept and filter every message, or do you need to pretend to be a client in some degree. I'll await feedback from others as I'm largely impartial. However, it could potentially be a good candidate for a new repo in the organisation if we wanted to make it non-server.

As an aside - I haven't got the topics code in front of me currently and it's been a while since I reviewed it, but it does make me wonder if we could collect topic particles for wildcards during the matching process and return them in the packet.Subscription values.

[dgduncan]

@mochi-co I do think the idea presented by @thedevop that they have implemented is very interesting; however, I do feel that it may be a more specific use case in which using the hooks makes more sense. I have not looked over the PR yet but I do feel that at least for this first pass a simple implementation should be in favor over a more robust approach.

[mochi-co]

I have been thinking about since, and given the choice I would favour @thedevop's approach, but perhaps balanced for usability via a new mochi-tools package which contains helper utilities. Having less non-protocol code in the server to support is probably not a bad thing, and this inline client stuff - while very interesting - takes us one precarious step further from the goal of keeping it simple.

[dgduncan]

I do agree that keeping core broker implementation as clean and readable is important. One of the really great things about this library is that is FAST compared to things like EMQX and the like and the more that gets added to the broker core the inevitably slower it becomes.

All that being said, I am still in favor of an inline subscribe to the broker as if an inline publish is a first class feature than at least to me having a first class subscribe makes sense as well.

If this becomes contentious I am happy to write up a DACI so all pros and cons of the implementations can be sussed out and decided on.

[mochi-co]

Finally getting around to all this now that it's the weekend! I am particularly interested in @werbenhu's PR #284 as I feel it strikes a nice balance, and it's easy for other developers to understand what's going on and debug issues. It's also largely separated from the core server code, so it will be effectively optimised out if not in use. I'll give it a thorough review now.

[mochi-co]

Following PR #284: we now have three well defined solutions across a whole range of behaviour and functionality:

• Hooks: As @thedevop suggested, using OnPublish hooks with a topic filter (e.g. radix tree). This is a filtered-firehose approach. Provides direct access to all messages (with a little overhead for checking every message). No subscription protocol logic (retained, ids, etc), or destination clients. Very high implementation cost for the embedding developer, but also very high level of customisability.
• @werbenhu's PR Another code implementation for Inline Client Subscriptions. #284: Provides a simple subscribe/unsubscribe mechanism for the developer, and allows them to set callbacks. Treated as a single client so not possible to have multiple subscriptions with different callbacks for the same topic. No subscription protocol logic (retained, etc) - although it should be possible to implement this - I have some ideas. However, very clean implementation, and it's obvious what's going on and how it works.
• PR Inline Client Subscriptions prototype #283 (inline clients): Extends the inline clients concept to include subscriptions, which provides all the features you get for a client. A considerable downside is that it provides all the features you get for a client, which hasn't been fully tested and could create some very unexpected scenarios if you were to start using shared subscriptions, NoLocal, etc. Mixed in with the existing client code, which could be confusing, but as a result there is less code, and opens up the option of making more sophisticated 'inline clients'. However, at a certain point you have to ask if it's worth doing that or if we should simply implement a companion lightweight MQTT client package that can be used in this way (if you were to go down this road, you could also implement an in-code listener for use within the same runtime, but then we are getting into some really out-there ideas).

Each of these ideas have strong strengths and weaknesses, and I would like to hear your thoughts on it to get a better sense for what would be useful for people.

[mochi-co]

On a personal note, I think @werbenhu's is probably the correct solution, however I'd like to see it extended to accept inline clients (as we do with Publish/InjectPacket) in order to allow us to deliver retained messages and support multiple callbacks per topic via different inline client ids.

[werbenhu]

I think this idea is very appropriate. I will look into how to implement them later.

[mochi-co]

@werbenhu Simplest solution I think would be to set the key of inline subscription map on client id, and accept an inline client parameter in the Subscribe and Unsubscribe methods.

[werbenhu]

@mochi-co I have completed the modifications, and you can take a look now.

[thedevop]

As OnPublish hook is not getting deprciated, inline subscription and hook are not mutually exclusive, merely some overlap capabilities. I see the benefit of having a very simple inline client subscription. So with that said, I personally prefer the implementation (#283 or #284) that is:

1. Very simple API to be used by the embedding developer. For more complex use case, the developer can use hook (not much more effort)
2. Introduces least amount of complexity to the MQTT server codebase
3. Has the least impact on the server performance when no inline subscriptions are used.

The following questions may help us determines if we need to suppoort retained message for inline client or not. Do we expect:

1. The embedding app subscribe to a topic immeidately upon start of the application?
2. Inline client to unsubscribe/resubscribe to a topic once it subscribed to a topic?

[werbenhu]

As mochi-server serves as an embedded MQTT broker, we cannot predict the user's usage scenarios. Assuming that users might integrate mochi-server with an HTTP web server and dynamically subscribe or unsubscribe using an HTTP API, I believe supporting retained messages and unsubscribe functionality is necessary.

[mochi-co]

2. Inline client to unsubscribe/resubscribe to a topic once it subscribed to a topic?

In all my experiences, the inline client was subscribing and unsubscribing arbitrarily without restarting the broker, based on user or environmental changes. I think this is probably where the common use-case diverges when compared with hooks, which are more typically designed to be set on build and not changed.

[mochi-co]

I've added some thoughts to #284, and cross-linking them here.

[werbenhu]

I've added some additional ideas here: #284. and cross-linking them here. I'd like to hear your thoughts on them.

[mochi-co]

@mdzio You will find that you can now Subscribe and Unsubscribe, please see v2.4.0 for details 🙂

[mdzio]

@mochi-co Thank you. I have followed the discussion and think that the implementation will fit. I will try out the new version.