Add Reconnection Timer

This commit is contained in:
2020-09-03 09:06:09 +02:00
parent f9547111b5
commit cb45769c20
7 changed files with 68 additions and 20 deletions
+1
View File
@@ -0,0 +1 @@
../../Libraries/Timer
+1
View File
@@ -4,6 +4,7 @@
LIBRARIES += Http
LIBRARIES += MQTT
LIBRARIES += Timer
LIBRARIES += Logging
LIBRARIES += Utilities
+40 -9
View File
@@ -40,7 +40,8 @@ void Bridge::Start()
m_pHttpServer.reset(new Http::HttpServer(m_port, httpCallback));
m_pWebSocketSubscription.reset(new WebSocketSubscription(m_toonSettings));
m_messageHandler.Connect(std::bind(&Bridge::TimeToLiveCallback, this, m_pWebSocketSubscription.get(), std::placeholders::_1));
m_messageHandler.Connect(std::bind(&Bridge::ReconnectCallback, this, m_pWebSocketSubscription.get()));
m_messageCheckTimer.StartContinuous(300000, static_cast<std::function<void()>>(std::bind(&Bridge::CheckMessageReceived, this)));
std::stringstream ss;
ss << m_mqttSettings.topic << "/BridgeState";
@@ -60,28 +61,58 @@ Http::HttpServer::HttpReply Bridge::HttpCallback(const std::string& uri, const s
{
if (postData.size() > 0)
{
for (auto& data : postData)
for (auto& item : postData)
{
if (data.name == "data")
m_messageHandler.HandleMessage(data.value);
if (item.name == "data")
m_messageHandler.HandleMessage(item.value);
}
}
}
catch (const std::exception& e)
{
std::stringstream ss;
ss << "Bridge::HttpCallback() - Error: " << e.what() << std::endl;
Logging::Log(Logging::Severity::Error, ss.str());
std::string data;
for (auto& item : postData)
{
if (item.name == "data")
data = (item.value);
}
}
{
std::unique_lock<std::mutex> lock(m_mutex);
m_messageReceived = true;
}
return reply;
}
void Bridge::TimeToLiveCallback(ToonBridge::Toon::WebSocketSubscription* pSubscription, int timeToLive)
void Bridge::ReconnectCallback(ToonBridge::Toon::WebSocketSubscription* pSubscription)
{
if (timeToLive < 20)
std::stringstream ss;
ss << "Bridge::TimeToLiveCallback() - TTL Expired, reconnecting" << std::endl;
Logging::Log(Logging::Severity::Info, ss.str());
pSubscription->Reconnect();
}
void Bridge::CheckMessageReceived()
{
bool messageReceived;
{
std::unique_lock<std::mutex> lock(m_mutex);
messageReceived = m_messageReceived;
m_messageReceived = false;
}
if (!messageReceived)
{
std::stringstream ss;
ss << "Bridge::CheckMessageReceived() - No message received for 5 minutes, reconnecting" << std::endl;
Logging::Log(Logging::Severity::Info, ss.str());
m_pWebSocketSubscription->Reconnect();
}
}
} // namespace Toon
} // namespace ToonBridge
+9 -1
View File
@@ -7,7 +7,9 @@
#include "WebSocketSubscription.h"
#include <HttpServer.h>
#include <MQTT.h>
#include <Timer.h>
#include <memory>
#include <mutex>
#include <string>
@@ -26,7 +28,8 @@ private:
void Start();
Http::HttpServer::HttpReply HttpCallback(const std::string& uri, const std::vector<Http::HttpPostData>& postData);
void TimeToLiveCallback(WebSocketSubscription* pSubscription, int timeToLive);
void ReconnectCallback(WebSocketSubscription* pSubscription);
void CheckMessageReceived();
private:
std::unique_ptr<Http::HttpServer> m_pHttpServer;
@@ -37,6 +40,11 @@ private:
ToonSettings m_toonSettings;
MQTTSettings m_mqttSettings;
std::unique_ptr<WebSocketSubscription> m_pWebSocketSubscription;
Timer::Timer m_messageCheckTimer;
std::mutex m_mutex;
bool m_messageReceived;
};
} // namespace Toon
+10 -4
View File
@@ -1,4 +1,5 @@
#include "MessageHandler.h"
#include <Logging.h>
#include <sstream>
@@ -18,8 +19,15 @@ MessageHandler::~MessageHandler()
void MessageHandler::HandleMessage(const std::string& message)
{
nlohmann::json data = nlohmann::json::parse(message);
auto updateDataSet = data["updateDataSet"];
if (data.contains("code"))
{
if (data["code"] == 510)
m_signal.emit();
return;
}
auto updateDataSet = data["updateDataSet"];
if (updateDataSet.contains("powerUsage"))
HandlePowerUsageMessage(updateDataSet["powerUsage"]);
if (updateDataSet.contains("gasUsage"))
@@ -30,11 +38,9 @@ void MessageHandler::HandleMessage(const std::string& message)
HandleThermostatInfoMessage(updateDataSet["thermostatInfo"]);
if (updateDataSet.contains("thermostatStates"))
HandleThermostatStatesMessage(updateDataSet["thermostatStates"]);
m_signal.emit(data["timeToLiveSeconds"]);
}
size_t MessageHandler::Connect(TTLCallbackMethod function)
size_t MessageHandler::Connect(ReconnectCallbackMethod function)
{
return m_signal.connect(function);
}
+3 -3
View File
@@ -19,14 +19,14 @@ public:
void HandleMessage(const std::string& message);
public:
typedef std::function<void(int timeToLive)> TTLCallbackMethod;
typedef std::function<void()> ReconnectCallbackMethod;
public:
size_t Connect(TTLCallbackMethod function);
size_t Connect(ReconnectCallbackMethod function);
void Disconnect(size_t connection);
private:
typedef Simple::Signal<void(int timeToLive)> TTLMessage;
typedef Simple::Signal<void()> TTLMessage;
private:
void HandlePowerUsageMessage(const nlohmann::json& data);
+3 -2
View File
@@ -1,5 +1,6 @@
#include "WebSocketSubscription.h"
#include <json.hpp>
#include <Logging.h>
#include <sstream>
@@ -45,7 +46,7 @@ void WebSocketSubscription::OpenWebSocket()
request.Headers(RequestHeaders());
request.Data(data.dump());
m_httpClient.Open(request);
std::string returnValue = m_httpClient.Open(request);
}
void WebSocketSubscription::WebSocketStatus()
@@ -73,7 +74,7 @@ void WebSocketSubscription::CloseWebSocket()
request.Method(Http::HttpRequest::Method::DELETE);
request.Headers(RequestHeaders());
m_httpClient.Open(request);
std::string returnValue = m_httpClient.Open(request);
}
std::vector<std::string> WebSocketSubscription::RequestHeaders()