Refactory repository interface with single pending message and subscriptions management #33
Conversation
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
Namoshek
reviewed
Aug 12, 2020
|
I updated and rebased, still very far just take a look to get an idea of where I'm going. |
Namoshek
requested changes
Aug 21, 2020
|
I'll look into the |
I think most of the FIXMEs should be handled in separate PRs, we can discuss them on discord and i can just remove them from this PR and start separate PRs once this one is merged, which in turn requires the Redis to be upgraded. |
|
If we mark a message published with QoS 2 as received after getting the PUBREC reply from the broker, do we also need to resend the PUBREL message if we do not receive a PUBCOMP in time? |
Namoshek
approved these changes
Aug 23, 2020
|
I think I leave the Please do another review yourself, I'll do one later or the next days as well to avoid overseeing things. |
Details: * Refactored the Repository interface. The new interface stores a generic serializable PendingMessage without knowing which type it is, using two different queues one for incoming and one for outgoing messages. There are currently three types of outgoing messages (PublishedMessage, SubscribeRequest, UnsubscribeRequest) and one incoming message (PublishedMessaged) handled. * The `TopicSubscription` class has been renamed to `Subscription` and now represents only the subscription itself, separating it from the concept of SubscribeRequest which might include several topic filters in one message. * Removed exception `UnexpectedAcknowledgementException`. It is expected to have duplicate packets over the wire, so there is no need for an expection in this case, a notice to the logger is enough. * Added `ProtocolViolationException`. This explicitly marks situations where the broker is misbehaving and the connection should be terminated. * Fixed some problems with the MessageProcessor.
|
Looks good to go! I squashed the commits keeping the logical separation. |
thg2k
changed the title
Closed
|
I changed some things (visibility, formatting, descriptions) we discussed already which only came to my eye during the last review. Looks good to merge now. |
Namoshek
added a commit
that referenced
this pull request
Jan 2, 2021
* Add more documentation * Use QoS constant in ConnectionSettings * Add isConnected() method to MQTTClient (contract) * Ensure client is connected before performing actions * Use QoS constants in favour of integers * Document hooks in MQTTClient contract * Improve PhpDoc of the interrupt() method * Remove unused shift($buffer, $limit) method * Extract pure functions into traits * Replace local message id with a manager implementation This change allows us to guarantee that message ids are never used twice at the same time. It also allows for different implementations, e.g. with persistence. * Add missing connection handshake errors of v3.1 * Merge MessageIdManager into Repository * Refactor ConnectionSettings The connection settings do now provide a fluent setter interface instead of a large constructor. Besides this change, the configurable caFile has been moved from the MQTTClient class to the connection settings. A bunch of new TLS settings have been added as well. * Avoid passing null for cafile/capath to context * Improve logging All logging statements have been reworked to utilize the log context interpolation provided by the PSR-3 log standard. Some log statements have been reduced in severity (from info to debug). Most log statements now contain the broker (referenced by host and port). Co-Authored-By: Giovanni Giacobbi <thg2k@users.noreply.github.com> * Add and use log helpers New log helper methods are used to add common information and context to log messages. All log messages are now prepended with broker information (host and port) as well as the client id. Also the word "MQTT" is now part of the common log output, making it obsolete for individual messages. Co-Authored-By: Giovanni Giacobbi <thg2k@users.noreply.github.com> * Wrap calling registered callbacks in try-catch Whenever registered callbacks are called, we catch all exceptions thrown by user provided code to avoid crashes of the client. We still log such exceptions as errors though. Co-Authored-By: Giovanni Giacobbi <thg2k@users.noreply.github.com> * Rename MQTTClient to MqttClient * Always create a socket context We also create a socket context without any settings if none are configured. This prevents accidental socket context pollution by other parts of an application. * Rename remaining occurences of MQTTClient* to MqttClient* (#18) * Refactor MqttClient protocol logic into MessageProcessor implementation (#28) * Implement basic MQTT v3 message processor The current implementation does parse messages, but it does not handle them just yet. * Refactor more functionality into message processor * Log sending/receiving data * Implement new loop logic * Split message parsing and processing * Throw more correct exceptions * Fix required bytes calculation * Improve data readin / connection logic * Fix wrong change in readFromSocket() * Add phpunit/phpunit as dev dependency * Only pass clientId to message processor * Add initial test for message processor and fix issue * Fix typo * Wait for first byte in connection logic Doing so avoids polling over and over again. Since we know, there must be at least one byte in the response. * Add an abstract BaseMessageProcessor * Fix protocol version in message processor docs * Use string to represent protocol version Since there are different protocol versions like 3, 3.1, 3.1.1 of which some use the same protocol level (like 3 and 3.1, which both use `3` as protocol level), it makes more sense to represent the protocol version as string. This also makes it less error prone if more protocol version levels are added. * Make ConnectionSettings immutable Having immutable connection settings prevents the settings from being altered after passing them to the connect() method. This is important since some settings are only used during the life time of a session. And if auto-reconnect is being implemented at some point, using the same settings for the reconnect will be essential as well. * Fix double use of $ * Restructure and rename connection settings The $qualityOfService and $retain settings are actually intended for the last will. We now transport this information through the name. * Restructure ConnectionSetting properties again * Update README.md The latest documentation improvements of the master branch have been added, but updated for the refactoring branch details. * Remove the option use blocking mode for the socket Since we decide based on the protocol logic whether to block for data on a per-call-basis or not, it is not necessary to set a global blocking mode on the socket itself. * Add connection settings validation * Fix phpcs * Add constructors to exceptions * Implement TLS updates of master * Validate subscription before calling unsubscribe() * Remove obsolete TODO It is not necessary to validate the message id anymore, since the message id is now guaranteed to be unique. * Add client certificate support for TLS * Fix issue when not passing any connection settings * Make stream non-blocking A blocking stream causes connection abortions, and we don't want that. * Add safety break when decoding message length * Fix message length encoding This fix has been taken from the master branch, see #30. * Add composer scripts for tests * Add getters for received and sent bytes * Add unit test for buildConnectMessage() * Add unit test for buildSubscribeMessage() * Add unit test for buildPingMessage() and buildDisconnectMessage() * Add remaining unit tests for message processor * Implement draft of RedisRepository * Subscribers of lower QoS receive higher QoS too In case multiple subscriptions of different topics matching the topic of a published message with different QoS exist, we also deliver higher QoS messages to subscribers with a lower QoS. Actually, the QoS doesn't matter at all in this case, since the broker handles it accordingly. * Run test and code style pipeline on push and PR * Fix code style issues * Inline 'addNew(...)' methods to MqttClient to reduce Repository interface (#35) * Add missing PUBREL in response to PUBREC * Use camelCase for logging context * Removed 'BaseRepository' class as it's covered by the 'Repository' interface (#36) * Fix message format of PUBREL since we expect ack * Refactory repository interface with single pending message and subscriptions management (#33) * Refactored Repository interface and pending message handling. Details: * Refactored the Repository interface. The new interface stores a generic serializable PendingMessage without knowing which type it is, using two different queues one for incoming and one for outgoing messages. There are currently three types of outgoing messages (PublishedMessage, SubscribeRequest, UnsubscribeRequest) and one incoming message (PublishedMessaged) handled. * The `TopicSubscription` class has been renamed to `Subscription` and now represents only the subscription itself, separating it from the concept of SubscribeRequest which might include several topic filters in one message. * Removed exception `UnexpectedAcknowledgementException`. It is expected to have duplicate packets over the wire, so there is no need for an expection in this case, a notice to the logger is enough. * Added `ProtocolViolationException`. This explicitly marks situations where the broker is misbehaving and the connection should be terminated. * Fixed some problems with the MessageProcessor. * Removed Redis repository implementation until updated to the new Repository interface * Fix MessageProcessor, exceptions and documentation * Change visibility and description of DTOs * Simplify acknowledgement parsing logic * Change code formatting * Change field visibility and formatting Co-authored-by: Marvin Mall <marvin-mall@msn.com> * Update from master * Fix composer.json * Change how CI runs tests * Fix security vulnerability * Refactor the loop() method to reduce complexity Co-authored-by: Giovanni Giacobbi <thg2k@users.noreply.github.com> Co-authored-by: Giovanni Giacobbi <giovanni@giacobbi.net>
# for free
to join this conversation on GitHub.
Already have an account?
# 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.
Ideas for refactoring repository and subscriptions
To be discussed.