Add offTopic() to remove per-topic callbacks registered via onTopic() - #25
Open
hmbacher wants to merge 1 commit into
Open
Add offTopic() to remove per-topic callbacks registered via onTopic()#25hmbacher wants to merge 1 commit into
hmbacher wants to merge 1 commit into
Conversation
onTopic() stores the user callback together with a strdup'd copy of the topic in _onMessageUserCallbacks, and (when connected) sends an MQTT SUBSCRIBE. There is currently no symmetric counterpart that removes the stored entry. unsubscribe() only forwards to esp_mqtt_client_unsubscribe() and leaves _onMessageUserCallbacks untouched. The consequence is a slow leak whenever a long-lived caller registers a callback through onTopic() and then needs to stop listening — e.g. when a dynamically created object that subscribes in its constructor is destroyed and replaced. The destroyed object's lambda (with all its captures) and the malloc'd topic string remain in _onMessageUserCallbacks for the lifetime of the PsychicMqttClient. Beyond the wasted bytes: - _onMessage() walks the whole vector for every incoming message, so dead entries linearly slow down dispatch. - _onConnect() re-subscribes to every topic in the vector on every reconnect, generating broker traffic for topics nothing is listening on. - Callers that wrap the lambda capture in a smart-pointer 'alive' flag to defend against use-after-free still cannot reclaim the memory, since the entry itself is unreachable from outside the library. This patch adds offTopic(topic): - Removes every entry whose stored topic equals topic (exact string equality, not wildcard expansion), freeing each malloc'd topic string. - If connected, sends MQTT UNSUBSCRIBE and returns its message id. - If not connected, returns -1 — the entries are still removed so they won't be re-subscribed on the next connect. Naming mirrors the existing onTopic()/onConnect()/onDisconnect() pattern. unsubscribe() is left unchanged to avoid breaking callers that rely on its current protocol-only semantics; offTopic() is the symmetric counterpart of onTopic() specifically.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Background
onTopic(topic, qos, callback)stores the user callback together with astrdup'd copy of the topic in_onMessageUserCallbacks, and (when connected) sends an MQTT SUBSCRIBE. There is currently no API to remove that stored entry. The existingunsubscribe(topic)is a thin wrapper overesp_mqtt_client_unsubscribe()and explicitly does not touch_onMessageUserCallbacks.The problem
For short-lived programs that register a fixed set of
onTopic()callbacks at startup this is invisible — the entries stay valid for the lifetime of the process. But it becomes a real issue once long-lived applications need to register callbacks dynamically. A common pattern is an object that callsonTopic()in its constructor and then gets destroyed and replaced (config reload, runtime add/remove of a feature, rename of a logical entity that owns a topic prefix, etc.). The destruction does nothing to the library state — the std::function lambda (capturing whatever the object captured) and themalloc'd topic string both remain in_onMessageUserCallbacksfor the lifetime of thePsychicMqttClientinstance.That leak has three downstream effects beyond the wasted bytes:
_onMessage()walk the entire_onMessageUserCallbacksvector for every incoming message and invoke every matching callback. Dead entries are still iterated and (if their topic still matches) still invoked — callers that capture analive-flag in the lambda to defend against use-after-free can short-circuit the body, but the dispatch cost remains._onConnect()re-subscribes to every topic in the vector on every reconnect, so each leaked entry pays for itself again every time the broker reconnects.onTopic()returns*thisfor fluency), so even diligent caller code cannot clean up.The leak is observable in practice on ESP32 builds that reload subscriptions when their configuration changes: every reload cycle leaks one entry per
onTopic()call. Memory pressure shows up after enough cycles, but the dispatch and reconnect-bandwidth costs are present from cycle one.The fix
This PR adds one new public method,
offTopic(const char *topic), which is the symmetric counterpart toonTopic():Behavior:
_onMessageUserCallbackswhose stored topic equalstopic(strcmp— exact string equality, not MQTT wildcard expansion). Pass the same string that was used to register.malloc'd topic string.topicand returns its message id.-1. The callback entries are still removed, so they will not be re-subscribed on the next connect.The choice to drop all matching entries (rather than just the first) matches the intuition of "stop listening to this topic" — callers that registered N callbacks for the same topic name almost certainly want all of them cleared in one call. Removing just one would also be hard to define unambiguously without exposing a callback handle, which would be a much bigger API change.
unsubscribe(topic)is intentionally left unchanged: it preserves its current protocol-only semantics for callers that rely on them.offTopic()is the right tool when the caller registered viaonTopic()and wants the full reverse operation in one call.API naming
offTopic()mirrors the existingonConnect/onDisconnect,onSubscribe/onUnsubscribe,onTopic/(newoffTopic) pattern. Alternative names considered:removeTopic(topic)— clear, but theon*family suggestsoff*is more in keeping with the existing style.unsubscribe(topic)to also remove the callback — cleaner one-call API, but technically a behavior change for existing callers, so deferred in favor of the additive approach.Implementation notes
std::remove_if+vector::erase(erase-remove idiom). Adds#include <algorithm>to the .cpp.cb.topicand sets it tonullptrbefore returningtrue, so the destructor's_onMessageUserCallbacks.clear()path (which also walks and frees topics) seesnullptrand skips them — no double-free.ESP_LOGImirrors the existing one inunsubscribe()so log output stays consistent.No breaking changes
unsubscribe(topic)behaves exactly as before.Testing
Tested on an ESP32-S3 application that previously leaked four
onTopic()registrations per destroy/recreate cycle of a config-owned object. After wiringoffTopic()into the object's destructor:_onMessageUserCallbacks.size()stays constant across many destroy/recreate cycles instead of growing linearly.$SYS/broker/subscriptions/countstays constant).