Skip to content

Add offTopic() to remove per-topic callbacks registered via onTopic() - #25

Open
hmbacher wants to merge 1 commit into
theelims:mainfrom
hmbacher:fix/on-topic-callback-removal
Open

Add offTopic() to remove per-topic callbacks registered via onTopic()#25
hmbacher wants to merge 1 commit into
theelims:mainfrom
hmbacher:fix/on-topic-callback-removal

Conversation

@hmbacher

Copy link
Copy Markdown

Background

onTopic(topic, qos, callback) 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 API to remove that stored entry. The existing unsubscribe(topic) is a thin wrapper over esp_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 calls onTopic() 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 the malloc'd topic string both remain in _onMessageUserCallbacks for the lifetime of the PsychicMqttClient instance.

That leak has three downstream effects beyond the wasted bytes:

  1. Dispatch slowdown. Both branches of _onMessage() walk the entire _onMessageUserCallbacks vector 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 an alive-flag in the lambda to defend against use-after-free can short-circuit the body, but the dispatch cost remains.
  2. Bandwidth on reconnect. _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.
  3. Unreclaimable from outside. Callers have no handle to the vector entry they created (onTopic() returns *this for 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 to onTopic():

int offTopic(const char *topic);

Behavior:

  • Removes every entry in _onMessageUserCallbacks whose stored topic equals topic (strcmp — exact string equality, not MQTT wildcard expansion). Pass the same string that was used to register.
  • Frees each removed entry's malloc'd topic string.
  • If the client is connected, sends an MQTT UNSUBSCRIBE for topic and returns its message id.
  • If not connected, returns -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 via onTopic() and wants the full reverse operation in one call.

API naming

offTopic() mirrors the existing onConnect/onDisconnect, onSubscribe/onUnsubscribe, onTopic/(new offTopic) pattern. Alternative names considered:

  • removeTopic(topic) — clear, but the on* family suggests off* is more in keeping with the existing style.
  • Extending 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

  • Uses std::remove_if + vector::erase (erase-remove idiom). Adds #include <algorithm> to the .cpp.
  • The lambda frees cb.topic and sets it to nullptr before returning true, so the destructor's _onMessageUserCallbacks.clear() path (which also walks and frees topics) sees nullptr and skips them — no double-free.
  • The ESP_LOGI mirrors the existing one in unsubscribe() so log output stays consistent.

No breaking changes

  • No existing method's signature or behavior changes.
  • unsubscribe(topic) behaves exactly as before.
  • The new method is purely additive.

Testing

Tested on an ESP32-S3 application that previously leaked four onTopic() registrations per destroy/recreate cycle of a config-owned object. After wiring offTopic() into the object's destructor:

  • _onMessageUserCallbacks.size() stays constant across many destroy/recreate cycles instead of growing linearly.
  • Free heap stays flat across cycles.
  • Reconnect no longer re-subscribes to dead topics (broker-side $SYS/broker/subscriptions/count stays constant).
  • No regressions in normal subscribe/receive/publish paths.

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.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant