Filter replaying unrelated retained MQTT messages when subscribing to share topics

[jbouwh]

Proposed change

The handling of MQTT messages with the retain flag has changed with this PR.

Each time a new subscription is created to a shared topic, existing subscriptions that already received a retained payload will no longer receive the replayed retained payload except when re-connecting to the broker.

A retain flag is added to a received message when we are (re)subscribing to the broker and (some) retained payload(s) available. The broker will replay these payloads directly after (re)subscribing—also, the same behavior occurs when the connection is lost, and we are reconnecting. When multiple subscriptions subscribe to the same topic, all shared subscriptions receive retained messages. Till this point, the behavior does not change.

When another subscription with shared topic(s) is added, we will subscribe to the broker to ensure we receive any retained messages for that new subscription, existing subscriptions (that already received a retained payload and were initialized earlier) will no longer receive the replayed retained payload triggered by the new subscribes.

For wildcards this works different, as there might be more than one payload. for the same subscription.
Basically the change effects shared subscriptions that receive a retained payload on a specific topic. When another shared subscription is added, we resubscribe to make sure any retained payloads are replayed. We only want to receive a retained payload once for a subscription per topic, so we filter out retained payloads when a new subscription is added. When reconnecting we allow replaying retained payloads for all subscriptions once again. I think we should filter out retained payloads for existing subscriptions to avoid the unexpected side effect on current subscriptions to receive a retained payload while that is not expected.

---

This part is communicated via a developers blogpost: home-assistant/developers.home-assistant#1775

The way how MQTT messages with a retain flag are handled has changed. Basically a retain flag is added to a received message when we are (re)subscribing to the broker and retained messages are available. The broker will replay these payloads directly after (re)subscribing. Also when the connection was lost, and we are reconnecting, the same behavior is seen. When there are multiple subscriptions that subscribe to the same topic, all shared subscriptions are receiving the retained messages. Up until this point the behavior has not changed.
When another subscription with shared topic(s) is added, we will resubscribe to the broker to ensure we receive any retained messages for that new subscription. Existing subscriptions (that already received a retained payload and were initialized earlier) will no longer receive the replayed retained payload triggered by the new subscribes. This will improve performance.

After replaying retained message the broker will send updates without a retain flag even when the remote client published with a retain flag, hence we cannot tell if that update was retained. See OASIS MQTT Version 3.1.1 spec (on how the RETAIN flag is used) and Normative Statement Number MQTT-3.3.1-9.

Type of change

• Dependency upgrade
• Bugfix (non-breaking change which fixes an issue)
• New integration (thank you!)
• New feature (which adds functionality to an existing integration)
• Deprecation (breaking change to happen in the future)
• Breaking change (fix/feature causing existing functionality to break)
• Code quality improvements to existing code or addition of tests

Additional information

• This PR fixes or closes issue: fixes #
• This PR is related to issue:
• Link to documentation pull request:

Checklist

• The code change is tested and works locally.
• Local tests pass. Your PR cannot be merged unless tests pass
• There is no commented out code in this PR.
• I have followed the development checklist
• The code has been formatted using Black (black --fast homeassistant tests)
• Tests have been added to verify that the new code works.

If user exposed functionality or configuration variables are added/changed:

• Documentation added/updated for www.home-assistant.io

If the code communicates with devices, web services, or third-party tools:

• The manifest file has all fields filled out correctly.
  Updated and included derived files by running: python3 -m script.hassfest.
• New or updated dependencies have been added to requirements_all.txt.
  Updated by running python3 -m script.gen_requirements_all.
• For the updated dependencies - a link to the changelog, or at minimum a diff between library versions is added to the PR description.
• Untested files have been added to .coveragerc.

To help with the load of incoming pull requests:

• I have reviewed two other open pull requests in this repository.

[home-assistant]

Hey there @emontnemery, mind taking a look at this pull request as it has been labeled with an integration (mqtt) you are listed as a code owner for? Thanks!

Code owner commands

Code owners of mqtt can trigger bot actions by commenting:

• @home-assistant close Closes the pull request.
• @home-assistant rename Awesome new title Renames the pull request.
• @home-assistant reopen Reopen the pull request.
• @home-assistant unassign mqtt Removes the current integration label and assignees on the pull request, add the integration domain after the command.

[corporategoth]

What's preventing this from being moved out of draft?

[jbouwh]

@corporategoth I have reworked the replaying so it will also work for wildcard topics, do you think this will work?