mqttclient.h
10.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
/* ****************************************************************************
* Copyright 2019 Open Systems Development BV *
* *
* Permission is hereby granted, free of charge, to any person obtaining a *
* copy of this software and associated documentation files (the "Software"), *
* to deal in the Software without restriction, including without limitation *
* the rights to use, copy, modify, merge, publish, distribute, sublicense, *
* and/or sell copies of the Software, and to permit persons to whom the *
* Software is furnished to do so, subject to the following conditions: *
* *
* The above copyright notice and this permission notice shall be included in *
* all copies or substantial portions of the Software. *
* *
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR *
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *
* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER *
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *
* DEALINGS IN THE SOFTWARE. *
* ***************************************************************************/
#ifndef OSDEV_COMPONENTS_MQTT_MQTTCLIENT_H
#define OSDEV_COMPONENTS_MQTT_MQTTCLIENT_H
// std
#include <condition_variable>
#include <memory>
#include <mutex>
#include <set>
#include <thread>
#include <vector>
// osdev::components::mqtt
#include "synchronizedqueue.h"
#include "istatecallback.h"
#include "serverstate.h"
#include "imqttclient.h"
#include "mqtt_lwt.h"
namespace osdev {
namespace components {
namespace mqtt {
// Forward definition
class IMqttClientImpl;
class MqttClient : public virtual IMqttClient
{
public:
/*!
* \brief Construct an instance of the MqttClient.
* \param clientId The client identification used in the connection to the mqtt broker.
* \param deliveryCompleteCallback Optional callback used to signal completion of a publication.
*/
MqttClient( const std::string& clientId, const std::function<void(const Token& token)>& deliveryCompleteCallback = std::function<void(const Token& token)>{});
virtual ~MqttClient() override;
// Non copyable, non movable
MqttClient(const MqttClient&) = delete;
MqttClient& operator=(const MqttClient&) = delete;
MqttClient(MqttClient&&) = delete;
MqttClient& operator=(MqttClient&&) = delete;
/**
* @see IStateCallback
*/
virtual std::string clientId() const override;
/**
* @see IStateCallback
*/
virtual StateChangeCallbackHandle registerStateChangeCallback(const SlotStateChange& cb) override;
/**
* @see IStateCallback
*/
virtual void unregisterStateChangeCallback(StateChangeCallbackHandle handle) override;
/**
* @see IStateCallback
*/
virtual void clearAllStateChangeCallbacks() override;
/**
* @see IStateCallback
*/
virtual StateEnum state() const override;
// MqttClient interface
/**
* @see IMqttClient
*/
virtual void connect( const std::string& host, int port, const Credentials &credentials, const mqtt_LWT &lvt = mqtt_LWT() ) override;
/**
* @see IMqttClient
*/
virtual void connect( const std::string &endpoint, const mqtt_LWT &lwt = mqtt_LWT() ) override;
/**
* @see IMqttClient
*/
virtual void disconnect() override;
/**
* @see IMqttClient
*/
virtual Token publish(const MqttMessage& message, int qos) override;
/**
* @see IMqttClient
* When an overlapping subscription is detected a new connection has to be created. This is a synchronous action
* and this method will wait for the connection to be up.
*/
virtual Token subscribe(const std::string& topic, int qos, const std::function<void(MqttMessage)>& cb) override;
/**
* @see IMqttClient
*/
virtual std::set<Token> unsubscribe(const std::string& topic, int qos) override;
/**
* @see IMqttClient
*/
virtual bool waitForCompletion(std::chrono::milliseconds waitFor) const override;
/**
* @see IMqttClient
*/
virtual bool waitForCompletion(std::chrono::milliseconds waitFor, const Token& token) const override;
/**
* @see IMqttClient
*/
virtual bool waitForCompletion(std::chrono::milliseconds waitFor, const std::set<Token>& tokens) const override;
/**
* @see IMqttClient
*/
virtual boost::optional<bool> commandResult(const Token& token) const override;
/**
* @see IMqttClient
*/
virtual std::string endpoint() const override;
private:
/*!
* \brief Callback used to pick up the connection status of the wrappers.
* \param id The client id.
* \param cs The connection status.
*/
void connectionStatusChanged( const std::string& id, ConnectionStatus cs );
/*!
* \brief Callback for handling delivery complete.
* \param clientId The identifier of the client on which the publish command is executed.
* \param token The token identifies the publish command.
*/
void deliveryComplete( const std::string& clientId, std::int32_t token );
/**
* \brief Wait for commands to complete including active tokens in this client.
* The interface mutex is not locked by this method.
* First wait for client commands to complete (use method waitForCompletionInternalClients)
* and then wait for publish delivery callbacks to complete.
* \param clients - Vector with client wrapper pointers that need to be waited on.
* \param[in,out] waitFor - The number of milliseconds to wait for completetion of all commands and delivery callbacks.
* \param tokens - The tokens to wait for. An empty set means to wait for all commands on all clients to complete
* including all publish delivery callbacks.
* \return True when commands have completed in time including delivery callbacks or false if not.
*/
bool waitForCompletionInternal(const std::vector<IMqttClientImpl*>& clients, std::chrono::milliseconds waitFor, const std::set<Token>& tokens) const;
/**
* \brief Wait for commands on the wrapper clients to complete.
* The interface mutex is not locked by this method.
* \param clients - Vector with client wrapper pointers that need to be waited on.
* \param[in,out] waitFor - The number of milliseconds to wait for completetion of all commands.
* On return waitFor contains the time left.
* \param tokens - The tokens to wait for. An empty set means to wait for all commands
* on all clients to complete.
* \return True when all commands have completed in time or false if not.
*/
bool waitForCompletionInternalClients(const std::vector<IMqttClientImpl*>& clients, std::chrono::milliseconds& waitFor, const std::set<Token>& tokens) const;
/**
* @brief Determine the state of this client based on the connection statusses of its client wrappers.
* The states this client can communicate are:
* Unknown : When at least one wrapper is in a different state then Connected or ReconnectInProgress or when no wrappers are available.
* Good : When all wrappers are connected.
* ConnectionFailure : When at least one wrapper attempts reconnection.
* Unregister : When the serverstate instance is destroyed.
*
* The other states are about the information providers to the mqtt broker (the publishers) and we cannot say anything about them here.
* The state "Good" is the exception. This state means in this case that this clients connection is ok and not that the underlying data
* source (publisher) is ok.
*
* @param connectionStates A vector with the connection statusses of all client wrappers.
*/
StateEnum determineState(const std::vector<ConnectionStatus>& connectionStates);
/**
* @brief Add an event to the synchronized queue.
* @param ev A function object that performs work in the context of this class.
*/
void pushEvent(std::function<void()> ev);
/**
* @brief Worker method that executes the events.
*/
void eventHandler();
mutable std::mutex m_interfaceMutex; ///< Makes the interface mutual exclusive
mutable std::mutex m_internalMutex; ///< Protect the internal state.
std::string m_endpoint; ///< The endpoint uri.
std::string m_clientId; ///< The main client identification.
std::set<Token> m_activeTokens; ///< Set with active command tokens. Callbacks still need to be made for these tokens.
mutable std::condition_variable m_activeTokensCV; ///< Wait on a condition to become true w.r.t. the active token set.
std::function<void(const Token&)> m_deliveryCompleteCallback; ///< Optional callback for publish completion.
ServerState m_serverState; ///< Records the state of the connection to the broker that this client is connected to.
std::unique_ptr<IMqttClientImpl> m_principalClient; ///< The main wrapper client.
std::vector<std::unique_ptr<IMqttClientImpl>> m_additionalClients; ///< A vector of additional wrapper clients.
SynchronizedQueue<std::function<void()>> m_eventQueue; ///< Synchronized queue for scheduling additional work.
std::thread m_workerThread; ///< A worker thread that is used to perform actions that cannot be done on the callback threads.
};
} // End namespace mqtt
} // End namespace components
} // End namespace osdev
#endif // OSDEV_COMPONENTS_MQTT_MQTTCLIENT_H