Allow plugin to obtain client source address. And when #24 is done, all relevant info from the proxy.

Format should probably be both sockaddr and std::string.

Originally posted by @slavslavov in #23 (reply in thread)

i don't know how to change the conf file when i try to login with allow anonymous yes i got access denied..

Thank you

I have been faced with a very odd situation where clients seem to be keeping reconnecting. Could not put my finger behind it. I might have reduced it to paho doing a default keep alive interval of 60s, and flashmq setting 5 seconds(?)

When updated the paho connect keepalive interval to 15, the client stayed connected.

I noticed that the situation is observed by a subscribe only process, have low amounts of traffic. I don't know if this is the exclusive case.

I was trying to use some sample plugins from mosquitto with flashmq but it failed to load.

There are two issues here i think:

1. the plugin failed to load
2. flashmq shutdown instead of just rejecting the plugin, logging the error and loading up. It probably is by design to make sure user understand the plugin did not load, but seemed something worth mentioning in the documentation.

plugin in question : https://github.com/eclipse/mosquitto/tree/master/plugins/payload-modification

feel free to close if above is expected and by design

error log:

[2022-12-01 17:43:29] [NOTICE] Loading auth plugin /home/ubuntu/mosquitto/mosquitto/plugins/payload-modification/mosquitto_payload_modification.so
[2022-12-01 17:43:29] [ERROR] Error initializing auth back-end: /home/ubuntu/mosquitto/mosquitto/plugins/payload-modification/mosquitto_payload_modification.so: undefined symbol: mosquitto_callback_register
[2022-12-01 17:43:29] [NOTICE] Quitting FlashMQ

Do you have any comparison on features as compared to other brokers such as clustering etc... I see from the videos that performance is not an issue but If I decide to go with flashmq will I suffer from lack of features and then end up going back to a lesser performing broker ? I am not looking to push 1M messages per second but I have noticed with lower hardware specs other brokers such EMQ and VernMQ have had issues with large volumes of messages so your broker seems to fit well in this area.

Thank you,

Gary

Flashmq randomly drops incoming SSL connections.

Besides the inline logs I share down bellow, I attach the related flasmq.log report.

The test is done with the latest tag (v1.10.0) on a Fedora 39 (Linux 6.7.7-200.fc39.x86_64) but I am facing similar issues after building docker image.

Just in case it helps, I'm using same environment to run similar tests against other MQTT brokers (emqx, hivemq, vernemq, mosquitto) and they seem to run smoothly at this point so I wonder if I am missing any fundamental config setting ? (see flashmq.conf bellow).

FlashMQ running instance

./FlashMQBuildRelease/flashmq -c flashmq.conf
[2024-03-13 10:00:11.311] [NOTICE] Loading config. Reload: false.
[2024-03-13 10:00:11.318] [NOTICE] Creating IPv4 SSL TCP listener on [0.0.0.0]:8883
[2024-03-13 10:00:11.319] [NOTICE] Creating IPv6 SSL TCP listener on [::0]:8883
[2024-03-13 10:00:11.319] [NOTICE] Creating IPv4 SSL TCP listener on [0.0.0.0]:8883
[2024-03-13 10:00:11.319] [NOTICE] Creating IPv6 SSL TCP listener on [::]:8883
[2024-03-13 10:00:11.319] [NOTICE] Creating IPv4 non-SSL TCP listener on [0.0.0.0]:1883
[2024-03-13 10:00:11.319] [INFO] Loading './str/bridgenames.db'
[2024-03-13 10:00:11.320] [INFO] Setting rlimit nofile to 1000000.
[2024-03-13 10:00:11.320] [NOTICE] 8 CPUs are detected, making as many threads. Use 'thread_count' setting to override.
[2024-03-13 10:00:11.320] [INFO] Loading './str/retained.db'
[2024-03-13 10:00:11.320] [INFO] Loading './str/sessions.db'
[2024-03-13 10:00:11.321] [NOTICE] Starting FlashMQ version 1.10.0, release build with SSE4.2 support.
[2024-03-13 10:00:11.321] [INFO] Switching logging from stdout to logfile './flashmq.log'

MqttLoadSimulator already reports dropped connections

./MqttLoadSimulator --hostname localhost --port 8883 --client-certificate ./new2/client.crt --client-private-key ./new2/client.key --ssl --delay 10
Version: 1.5.0.
Clients: 2 on 8 threads. Sent: 175 (25/s). Recv: 175 (25/s). Recv-Sent: 0. Connects: 2 (0/s). Disconnects: 2 (0/s). Errors: 4 (0/s).
Message latency (min/avg/max): 0.4 ms / 0.7 ms / 1.1 ms.
Thread loop drift: Avg: 1.00 ms, max: 1 ms (OK)

Similar behavior with other benchmark tools like "mqtt-bechmark"

mqtt-benchmark --broker tls://localhost:8883 --count 10 --size 100 --clients 10 --qos 1 --format text -broker-ca-cert ./new2/ca.crt -client-cert ./new2/client.crt -client-key ./new2/client.key
2024/03/13 10:01:10 Starting client  0
2024/03/13 10:01:10 Starting client  1
2024/03/13 10:01:10 Starting client  2
2024/03/13 10:01:10 Starting client  3
2024/03/13 10:01:10 Starting client  4
2024/03/13 10:01:10 Starting client  5
2024/03/13 10:01:10 Starting client  6
2024/03/13 10:01:10 Starting client  7
2024/03/13 10:01:10 Starting client  8
2024/03/13 10:01:10 Starting client  9
2024/03/13 10:01:10 CLIENT 4 had error connecting to the broker: not Authorized
2024/03/13 10:01:10 CLIENT 8 had error connecting to the broker: not Authorized
2024/03/13 10:01:10 CLIENT 0 had error connecting to the broker: not Authorized
2024/03/13 10:01:10 CLIENT 9 had error connecting to the broker: not Authorized
2024/03/13 10:01:10 CLIENT 6 had error connecting to the broker: not Authorized
2024/03/13 10:01:10 CLIENT 1 had error connecting to the broker: not Authorized
2024/03/13 10:01:10 CLIENT 3 is connected to the broker tls://localhost:8883
2024/03/13 10:01:10 CLIENT 7 is connected to the broker tls://localhost:8883
2024/03/13 10:01:10 CLIENT 5 is connected to the broker tls://localhost:8883
2024/03/13 10:01:10 CLIENT 2 is connected to the broker tls://localhost:8883

flashmq.conf

log_file    ./flashmq.log
storage_dir ./str

listen {
  protocol mqtt
  inet_protocol ip4_ip6
  inet4_bind_address 0.0.0.0
  inet6_bind_address ::0
  fullchain ./new2/server.crt
  privkey ./new2/server.key

  # default = 8883
  port 8883
}

listen {
  protocol mqtt
  fullchain ./new2/server.crt
  privkey ./new2/server.key
  client_verification_ca_file ./new2/ca.crt
  client_verification_still_do_authn false
}

listen {
  protocol mqtt
  inet_protocol ip4

  # default = 1883
  port 1883
}

flashmq.log

If you connect to the broker with a Last Will and Testament (LWT) and correct username, the LWT is published even if the password is wrong.

Lots of MQTT broker support lua scripting, such as vernemq and emqtt.
Lua can make it easier to develop plugins.

All,
Building for deployment on bare metal (so don't want docker image).
Steps:
git clone https://github.com/halfgaar/FlashMQ.git
cd FlashMQ
git checkout v1.1.0
mkdir build
cd build
cmake ..
make
sudo make install

Consolidate compiler generated dependencies of target flashmq
[100%] Built target flashmq
Install the project...
-- Install configuration: ""
-- Installing: /usr/bin/flashmq
-- Up-to-date: /var/lib/flashmq
-- Up-to-date: /var/log/flashmq
-- Installing: /etc/flashmq/flashmq.conf
-- Installing: /lib/systemd/system/flashmq.service
CMake Error at cmake_install.cmake:126 (file):
file INSTALL cannot find "/home/myusername/FlashMQ/man/flashmq.conf.5":
No such file or directory.

It seems that I didn't have docbook2x installed. Would be nice if the README.md included the dependencies for those of us that don't intend to deploy in docker. If I missed it somehow in the documentation then I apologize.

Looks like an awesome broker.

In flashmq_plugin.h, the argument of flashmq_poll_remove_fd has type uint32_t while in the other two flashmq_poll_XX functions it is int

I tried to compile with musl, but got errors:

flashmq> [ 18%] Building CXX object CMakeFiles/flashmq.dir/types.cpp.o
flashmq> In file included from /build/source/utils.h:34,
flashmq>                  from /build/source/utils.cpp:20:
flashmq> /build/source/cirbuf.h:55:21: error: ‘uint’ has not been declared
flashmq>    55 |     void doubleSize(uint factor = 2);
flashmq>       |                     ^~~~
flashmq> In file included from /build/source/utils.h:34,
flashmq>                  from /build/source/mainapp.h:35,
flashmq>                  from /build/source/mainapp.cpp:18:
flashmq> /build/source/cirbuf.h:55:21: error: ‘uint’ has not been declared
flashmq>    55 |     void doubleSize(uint factor = 2);
flashmq>       |                     ^~~~
flashmq> In file included from /build/source/session.cpp:20:
flashmq> /build/source/session.h:95:38: error: ‘u_int16_t’ has not been declared
flashmq>    95 |     bool removeIncomingQoS2MessageId(u_int16_t packet_id);
flashmq>       |                                      ^~~~~~~~~
flashmq> /build/source/session.h:98:38: error: ‘u_int16_t’ has not been declared
flashmq>    98 |     void removeOutgoingQoS2MessageId(u_int16_t packet_id);
flashmq>       |                                      ^~~~~~~~~
flashmq> In file included from /build/source/variablebyteint.h:4,
flashmq>                  from /build/source/mqtt5properties.h:7,
flashmq>                  from /build/source/types.cpp:21:
flashmq> /build/source/cirbuf.h:55:21: error: ‘uint’ has not been declared
flashmq>    55 |     void doubleSize(uint factor = 2);
flashmq>       |                     ^~~~
flashmq> /build/source/session.cpp:330:6: error: no declaration matches ‘bool Session::removeIncomingQoS2MessageId(u_int16_t)’
flashmq>   330 | bool Session::removeIncomingQoS2MessageId(u_int16_t packet_id)
flashmq>       |      ^~~~~~~
flashmq> /build/source/session.h:95:10: note: candidate is: ‘bool Session::removeIncomingQoS2MessageId(int)’
flashmq>    95 |     bool removeIncomingQoS2MessageId(u_int16_t packet_id);
flashmq>       |          ^~~~~~~~~~~~~~~~~~~~~~~~~~~
flashmq> /build/source/session.h:33:7: note: ‘class Session’ defined here
flashmq>    33 | class Session
flashmq>       |       ^~~~~~~
flashmq> /build/source/session.cpp:358:6: error: no declaration matches ‘void Session::removeOutgoingQoS2MessageId(u_int16_t)’
flashmq>   358 | void Session::removeOutgoingQoS2MessageId(u_int16_t packet_id)
flashmq>       |      ^~~~~~~
flashmq> /build/source/session.h:98:10: note: candidate is: ‘void Session::removeOutgoingQoS2MessageId(int)’
flashmq>    98 |     void removeOutgoingQoS2MessageId(u_int16_t packet_id);
flashmq>       |          ^~~~~~~~~~~~~~~~~~~~~~~~~~~
flashmq> /build/source/session.h:33:7: note: ‘class Session’ defined here
flashmq>    33 | class Session
flashmq>       |       ^~~~~~~

In threaddata.cpp some "$SYS" topics are currently defined. In your opinion:

1. Should the server be able to provide throughput per topic so monitoring it becomes possible?
2. Should a client aggregate these statistics, at the cost of receiving all information?

Is possible to compile in a Raspberry PI?. Or is a package for Raspbian?

Async interfaces of libpq, libcurl, etc, give you a socket to watch. Add ability to register this to the thread's epoll.

Fails to build with gcc 13.1.1. Some standard include files are missing.

The non-normative comment in the MQTT v5 standard, point 4.7.1.2 says:

Client subscribes to “sport/tennis/player1/#”, it would receive messages published using these Topic Names:
· “sport/tennis/player1”
· “sport/tennis/player1/ranking
· “sport/tennis/player1/score/wimbledon”

When I subscribe with a mosquitto client:
mosquitto_sub -t "sport/tennis/player1/#" -v

and publish:
mosquitto_pub -t "sport/tennis/player1" -m "hello"

I don't get the message. Publishing the other two of the examples is OK.

A potential fix may be to add the following after the while loop in topicsMatch

    if (subscribe_itr != subscribe_itr.end() && *subscribe_itr == "#")
    {
        return true;
    }

Hi,
FlashMQ seems like a fantastic broker :)
I was wondering if load balancing strategies have been considered for FlashMQ.
Specifically, I'm interested in emqx's 'hash' strategy, which effectively pairs publishers with subscribers.

I've cloned repository to build the most recent image :
docker build . -t xxxx/flashmq:v230226.2208
and pushed it.
When I restart my container with this new image I get the following error 👍
flashmq | /bin/flashmq: error while loading shared libraries: libresolv.so.2: cannot open shared object file: No such file or directory
several times
It's the same result if I try with the v1.2.1 code source

This involves:

• Verifying the client certificate
• Reading the username from the certificate

Is it possible, on server level, to prevent retained message with the exact same contents to be resend?

So something like:

if incomming.message == retained messages [incomming.topic] then
   noop

FlashMQ currently doesn't except MQTT connect packets that flag a username as present, but is empty. It says:

Username flagged as present, but it's 0 bytes.

The MQTT spec says:

If the User Name Flag is set to 1, a user name MUST be present in the payload

But, there is some ambiguity if a zero byte username means 'present'. FlashMQ has the current behavior to not have two ways of having an anonymous login.

But, because other servers do accept this situation, we should deal with it. Perhaps with a config option.

Does ssl cert reload the listener when changed ?

Thank you

It seems that I am not getting all the (expected) retained messages. Is there a configurable maximum for storage?

Hello,

Is it possible to configure a bridge to another MQTT broker so that my broker forwards messages from some selected topics to another broker? It is a core feature e.g. in mosquitto, see http://www.steves-internet-guide.com/mosquitto-bridge-configuration/

I cannot find any settings for that in the flashmq config file. Please help. Thanks.

Can you create a tutorial how to install FlasMQ on RaspberryPi ?
I tried but failed:
N: Skipping loading the configured file "main/binary-armhf/Packages" because the depot "http://repo.flashmq.org/apt bullseye InRelease" does not support the "armhf" architecture.

If I didn't know for sure that it would run on another RaspberryPi (Victron Venus) - I would not ask for.

Review + test + think about slavslavov#1.

Was discussed here.

The last action upon a restart is saving retained messages to /var/lib/flashmq/retained.db, would it be possible to do this in a different way, since it takes quite long to write it. In addition to it, it looks that flashmq needs a whopping 6GB of RAM, and my expectation is that this is mostly due to these retained messages, any chance to offload this?

TODO:

• Implement.
• Add warning in man page that with thousands or millions of clients, purposefully slow hashing algorithms can be a problem, because it may require hours for all clients to connect.

Example:

wiebe:$7$101$oW4vum9VREb/0aH6K3ge7kY9u26JboPDpF+9SKjd0fdz6SXidJ9n+TtBbg5n0ImfM/LkIEn6BZxT78Wn6Uzgvw==$Sc31amckhadG9Sd7/qTR8L+ROxljBLCMTrFj/Isb0rikvolsWnucdfDDY3bcIk03Nm/ymzfenE7LuTtgbz3wkA==

Research / implement the HA proxy protocol per listener. Remember to extend the functionality in #25 if implemented.

At this point, I'm not familiar with it. If it's merely a few extra bytes before MQTT connect, it should be possible.

Originally posted by @slavslavov in #23 (reply in thread)

This unexpectedly passes:

$ FlashMQ --test-config -c /dev/null
Config OK

$ ulimit -m 12345678
$ FlashMQ --test-config -c /dev/zero
Config OK

When I modify the code and call infile.exceptions(std::ifstream::failbit | std::ifstream::badbit); after opening infile they fail as expected:

$ FlashMQ --test-config -c /dev/null
basic_ios::clear: iostream error

$ FlashMQ --test-config -c /dev/zero
std::bad_alloc

I am trying to get haproxy support to work as SSL terminator for port 8883. My expectation was the setup below would be in the right direction. I have tested it with mode tcp and mode http.

frontend mqtt
    bind :::8883 ssl crt /etc/haproxy/ssl
    tcp-request content reject unless { req.payload(0,0),mqtt_is_valid }
    use_backend flashmq
    mode tcp
    maxconn 1000

backend flashmq
    mode tcp

    # Create a stick table for session persistence
    stick-table type string len 32 size 100k expire 30m

    # Use ClientID / client_identifier as persistence key
    stick on req.payload(0,0),mqtt_field_value(connect,client_identifier)
    server flashmq 127.0.0.1:2883

listen {
  protocol mqtt
  port 2883
  haproxy on
}

FlashMQ Version 1.4.5 with SSE4.2 support
HAProxy version 2.4.18-0ubuntu1 2022/08/25 - https://haproxy.org/

A little bit more elaboration;

The client succesfully gets the SSL handshake done. The log of FlashMQ raises:

[2023-05-24 17:09:53] [NOTICE] Accepting connection from: address='127.0.0.1', transport='TCP/HAProxy/Non-SSL', fd=18
[2023-05-24 17:09:53] [NOTICE] Removing client '[ClientID='', username='', fd=18, keepalive=0s, transport='TCP/HAProxy/Non-SSL', address='127.0.0.1', prot=none, clean=0]'. Reason(s): HAProxy health check, epoll says socket is in ERR or HUP state.

I think LWT is not working as it should in FlashMQ. I had some alerts in my uptime monitor that are not working after I changed my Server to FlashMQ.

I try to compare Mosquitto and FlashMQ with two shell windows.
First I connect to my Mosquitto Server:
On the 1st I do:
mosquitto_sub --will-topic foobar_topic --will-payload crashed --will-retain -i foobar -t foobar_topic -v -h localhost --disable-clean-session
no output yet. 👍

On the 2nd I do:
mosquitto_pub --will-topic foobar_topic --will-payload crashed --will-retain -i foobar -t foobar_topic -m online -h localhost -r --disable-clean-session
now the 1st outputs: foobar_topic online 👍

I only get the crashed LWT once I kill -9 the mosquitto_sub process.

Now I try my FlashMQ Server:
On the 1st I do:
mosquitto_sub --will-topic foobar_topic --will-payload crashed --will-retain -i foobar -t foobar_topic -v -h localhost --disable-clean-session
instant output foobar_topic crashed👎

On the 2nd I do:
mosquitto_pub --will-topic foobar_topic --will-payload crashed --will-retain -i foobar -t foobar_topic -m online -h localhost -r --disable-clean-session
now the 1st outputs: foobar_topic crashed 👎

Also I watch the topic with MQTT Explorer and once I pub, I get both topics, crashed and online afterwards. This is different from Mosquitto.

what would be the best approach to archive all the messages received in a DB ? would it be to tap into AlterPublish path for the plugin and pipe to DB or you would suggest to instead create a dumb subscriber whose sole purpose is to subscribe to all topics and keep writing to a DB ?

Building 1.7.2 fails with gcc (GCC) 13.1.1 20230511 (Red Hat 13.1.1-2). The error is that std::string is used in globalstats.h but the <string> header is not included.

Simply including <string> fixes the issue

Currently all plugin functions are called for each thread but it will be good to have one which is called only once from the main thread before the other threads are created.
To minimise the impact on the current design its signature may look like this:
typedef void(*F_flashmq_plugin_main_init_v2)(std::unordered_map<std::string, std::string> &auth_opts);
No global data pointer is added to keep the multi-core design of FlashMQ unchanged and in case needed, a static variable can be used as suggested in the documentation for flashmq_plugin_allocate_thread_memory.

One use case for this function would be to check which OS user is running the process.

The best place to call this function would be here:

    GlobalStats *globalStats = GlobalStats::getInstance();
->
    for (int i = 0; i < num_threads; i++)
    {
        std::shared_ptr<ThreadData> t = std::make_shared<ThreadData>(i, settings);

But the main problem is Authentication, which is a member of ThreadData is instantiated a few lines after this location.

Hi,

While FlashMQ supports prefixing topics on messages it sends and receives, it doesn't seem to be able to remove a prefix from incoming or outgoing topics, at least as far as flashmq.conf manpage is concerned.

This would afford transparently tunneling arbitrary messages through a shared server.

Mosquitto can do it ;-)

Hello,

when i try to give access for flashmq user i didn't gain the access with chown flashmq:flashmq & chmod..

File '/var/log/flashmq/flashmq.log' is not there and can't be created, because '/var/log/flashmq' is also not writable

I'm not a linux expert so..

#Edit (not for this case)
For mTLS i also add SSL_CTX_use_certificate_chain_file in listener.cpp

Preliminary tests are good. Should be doable.