{"action":"created","issue":{"url":"https://api.github.com/repos/eclipse/smarthome/issues/5535","repository_url":"https://api.github.com/repos/eclipse/smarthome","labels_url":"https://api.github.com/repos/eclipse/smarthome/issues/5535/labels{/name}","comments_url":"https://api.github.com/repos/eclipse/smarthome/issues/5535/comments","events_url":"https://api.github.com/repos/eclipse/smarthome/issues/5535/events","html_url":"https://github.com/eclipse/smarthome/pull/5535","id":320503543,"node_id":"MDExOlB1bGxSZXF1ZXN0MTg2MTUzNDIy","number":5535,"title":"Add MqttBrokerConnection future API. Add tests. Add MqttService interface","user":{"login":"davidgraeff","id":66436,"node_id":"MDQ6VXNlcjY2NDM2","avatar_url":"https://avatars2.githubusercontent.com/u/66436?v=4","gravatar_id":"","url":"https://api.github.com/users/davidgraeff","html_url":"https://github.com/davidgraeff","followers_url":"https://api.github.com/users/davidgraeff/followers","following_url":"https://api.github.com/users/davidgraeff/following{/other_user}","gists_url":"https://api.github.com/users/davidgraeff/gists{/gist_id}","starred_url":"https://api.github.com/users/davidgraeff/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/davidgraeff/subscriptions","organizations_url":"https://api.github.com/users/davidgraeff/orgs","repos_url":"https://api.github.com/users/davidgraeff/repos","events_url":"https://api.github.com/users/davidgraeff/events{/privacy}","received_events_url":"https://api.github.com/users/davidgraeff/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":35,"created_at":"2018-05-05T11:55:45Z","updated_at":"2018-07-05T10:53:06Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/eclipse/smarthome/pulls/5535","html_url":"https://github.com/eclipse/smarthome/pull/5535","diff_url":"https://github.com/eclipse/smarthome/pull/5535.diff","patch_url":"https://github.com/eclipse/smarthome/pull/5535.patch"},"body":"This is an API breaking commit. The current API has some serious design flaws that can cause timing issues. At the end of this post is a list of PRs that adapt all mqtt bundles to the new API.\r\n\r\n### Design flaws:\r\n* Subscribe (called addConsumer) is an async method. But there was no way to register a callback.\r\n* Stop() calls an async disconnect method. But the method returns before the disconnect process is finished and there is no way to get notified about the process.\r\n* MqttMessageSubscriber was not only responsible for receiving messages for a subscribed topic, but also needed to provide that topic. It was not possible to use one MqttMessageSubscriber for multiple topics of interest.\r\n* It is supported to subscribe to topics and then connect to a broker. But the start() method didn't\r\n  return any feedback if subscriptions are failing (see trySubscribe, it only logs).\r\n* In messageArrived: The MqttMessageSubscribers list was looped and subscribers were directly notified\r\n  of a message. If a subscriber altered the list by for example adding another subscriber or removing itself, a ConcurrentModification exception was to be expected.\r\n\r\n### Is it possible to preserve the current API?\r\n`stop()` and `subscribe()` can be turned into synchronous methods and documented as such. Because massive subscribing to topics (like with the homie mqtt convention thing support) is a valid use-case, a synchronous only API for subscribing is not satisfying though.\r\n\r\n### Changes:\r\n* Rename addConsumer/removeConsumer to subscribe/unsubscribe\r\n* Separate MqttService and MqttServiceImpl\r\n* Remove MqttPublishResult\r\n* Simplify MqttMessageSubscriber\r\n* Move private classes out of MqttBrokerConnection if applicable\r\n* Introduce CompletableFuture based API for MqttBrokerConnection\r\n* Add another publish method where qos and retain can be specified.\r\n* Don't return the deliveryToken message ID for publish. This is paho specific,  and the API should stay library agnostic.\r\n* Clean up in stop(), unsubscribe from all topics, correctly close the persistence datastore and client.\r\n* Add protocol selection (Websocket, TCP)\r\n\r\n### Test additions:\r\n* Add wildcard (+,#) tests and fix related wildcard topic match bug\r\n* Add test for connecting, disconnecting multiple times\r\n* Add tests for subtopic, complicated utf8 topic name subscriptions before and after a connection is established\r\n\r\n### Adapt other Mqtt bindings\r\n* Embedded broker: #5536 \r\n* Mqtt main binding (broker configuration): #5537 \r\n* Mqtt generic things and homie things: #5450 \r\n\r\n### API overview of MqttBrokerConnection\r\n```java\r\n    /**\r\n     * Add a new message consumer to this connection. Multiple subscribers with the same\r\n     * topic are allowed. This method will not protect you from adding a subscriber object\r\n     * multiple times!\r\n     *\r\n     * @param consumer Consumer to add\r\n     * @throws MqttException If connected and the subscribe fails, this exception is thrown.\r\n     */\r\n    public CompletableFuture<Boolean> subscribe(String topic, MqttMessageSubscriber subscriber);\r\n\r\n    /**\r\n     * Subscribes to a topic on the given connection, but does not alter the subscriber list.\r\n     */\r\n    protected CompletableFuture<Boolean> subscribeRaw(String topic);\r\n\r\n    /**\r\n     * Remove a previously registered consumer from this connection.\r\n     *\r\n     * @param subscriber to remove.\r\n     */\r\n    @SuppressWarnings({ \"null\", \"unused\" })\r\n    public CompletableFuture<Boolean> unsubscribe(String topic, MqttMessageSubscriber subscriber);\r\n\r\n    /**\r\n     * Unsubscribes a topic on the given connection, but does not alter the subscriber list.\r\n     *\r\n     * @param client The client connection\r\n     * @param topic The topic to unsubscribe from\r\n     * @return Completes with true if successful. Exceptionally otherwise.\r\n     */\r\n    protected CompletableFuture<Boolean> unsubscribeRaw(MqttAsyncClient client, String topic);\r\n\r\n    /**\r\n     * Unsubscribe from all topics\r\n     *\r\n     * @return Returns a future that completes as soon as all subscriptions have been canceled.\r\n     */\r\n    public CompletableFuture<Void> unsubscribeAll() ;\r\n\r\n    /**\r\n     * This will establish a connection to the MQTT broker and if successful, notify all\r\n     * publishers and subscribers that the connection has become active. This method will\r\n     * do nothing if there is already an active connection.\r\n     *\r\n     * @return Returns a future that completes with true if already connected or connecting,\r\n     *         completes with false if a connection timeout has happened and completes exceptionally otherwise.\r\n     */\r\n    public CompletableFuture<Boolean> start();\r\n\r\n    /**\r\n     * Unsubscribes from all subscribed topics, stops the reconnect strategy, disconnect and close the client.\r\n     *\r\n     * You can re-establish a connection calling {@link #start()} again. Do not call start, before the closing process\r\n     * has finished completely.\r\n     *\r\n     * @return Returns a future that completes as soon as the disconnect process has finished.\r\n     */\r\n    public CompletableFuture<Boolean> stop();\r\n\r\n    /**\r\n     * Publish a message to the broker.\r\n     *\r\n     * @param topic The topic\r\n     * @param payload The message payload\r\n     * @return Returns a future that completes with a result of true if the publishing succeeded and completes\r\n     *         exceptionally on an error or with a result of false if no broker connection is established.\r\n     */\r\n    public CompletableFuture<Boolean> publish(String topic, byte[] payload);\r\n\r\n    /**\r\n     * Publish a message to the broker with the given QoS and retained flag.\r\n     *\r\n     * @param topic The topic\r\n     * @param payload The message payload\r\n     * @param qos The quality of service for this message\r\n     * @param retain Set to true to retain the message on the broker\r\n     * @param listener An optional listener to be notified of success or failure of the delivery.\r\n     * @return Returns a future that completes with a result of true if the publishing succeeded and completes\r\n     *         exceptionally on an error or with a result of false if no broker connection is established.\r\n     */\r\n    public CompletableFuture<Boolean> publish(String topic, byte[] payload, int qos, boolean retain);\r\n```\r\n\r\nFixes #5370\r\nFixes #5527 \r\n\r\nSigned-off-by: David Graeff <david.graeff@web.de>"},"comment":{"url":"https://api.github.com/repos/eclipse/smarthome/issues/comments/402684491","html_url":"https://github.com/eclipse/smarthome/pull/5535#issuecomment-402684491","issue_url":"https://api.github.com/repos/eclipse/smarthome/issues/5535","id":402684491,"node_id":"MDEyOklzc3VlQ29tbWVudDQwMjY4NDQ5MQ==","user":{"login":"davidgraeff","id":66436,"node_id":"MDQ6VXNlcjY2NDM2","avatar_url":"https://avatars2.githubusercontent.com/u/66436?v=4","gravatar_id":"","url":"https://api.github.com/users/davidgraeff","html_url":"https://github.com/davidgraeff","followers_url":"https://api.github.com/users/davidgraeff/followers","following_url":"https://api.github.com/users/davidgraeff/following{/other_user}","gists_url":"https://api.github.com/users/davidgraeff/gists{/gist_id}","starred_url":"https://api.github.com/users/davidgraeff/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/davidgraeff/subscriptions","organizations_url":"https://api.github.com/users/davidgraeff/orgs","repos_url":"https://api.github.com/users/davidgraeff/repos","events_url":"https://api.github.com/users/davidgraeff/events{/privacy}","received_events_url":"https://api.github.com/users/davidgraeff/received_events","type":"User","site_admin":false},"created_at":"2018-07-05T10:53:06Z","updated_at":"2018-07-05T10:53:06Z","author_association":"CONTRIBUTOR","body":"Travis passes :)"}}
{"actor":{"display_login":"davidgraeff"}}