iot/sprocket-plugin-mqtt
1
0
Fork 0
mirror of https://gitlab.com/wirelos/sprocket-plugin-mqtt.git synced 2025-12-14 21:52:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

apply config from file

Browse Source
This commit is contained in:
0x1d
2018-11-19 12:59:04 +01:00
parent 30ad1041cb
commit e1d605c615
4 changed files with 36 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/MqttPlugin.cpp
View File

@@ -2,17 +2,30 @@
MqttPlugin::MqttPlugin(MqttConfig cfg) MqttPlugin::MqttPlugin(MqttConfig cfg)
{ {
mqttConfig.brokerHost = cfg.brokerHost; applyConfig(cfg);
mqttConfig.brokerPort = cfg.brokerPort; }
mqttConfig.clientName = cfg.clientName;
mqttConfig.inTopicRoot = cfg.inTopicRoot; void MqttPlugin::applyConfig(MqttConfig cfg) {
mqttConfig.outTopicRoot = cfg.outTopicRoot; brokerHost = String(cfg.brokerHost);
brokerPort = cfg.brokerPort;
clientName = String(cfg.clientName);
inTopicRoot = String(cfg.inTopicRoot);
outTopicRoot = String(cfg.outTopicRoot);
}
void MqttPlugin::applyConfigFromFile(const char* fileName) {
MqttConfigJson configFile;
configFile.fromFile(fileName);
if(configFile.valid){
PRINT_MSG(Serial, "MQTT", "apply config from file");
applyConfig(configFile);
}
} }
void MqttPlugin::activate(Scheduler *scheduler) void MqttPlugin::activate(Scheduler *scheduler)
{ {
mqttConfig.fromFile("/mqttConfig.json"); applyConfigFromFile("/mqttConfig.json");
client = new PubSubClient(mqttConfig.brokerHost, mqttConfig.brokerPort, bind(&MqttPlugin::downstreamHandler, this, _1, _2, _3), wifiClient); client = new PubSubClient(brokerHost.c_str(), brokerPort, bind(&MqttPlugin::downstreamHandler, this, _1, _2, _3), wifiClient);
enableConnectTask(scheduler); enableConnectTask(scheduler);
enableProcessTask(scheduler); enableProcessTask(scheduler);
PRINT_MSG(Serial, "MQTT", "plugin activated"); PRINT_MSG(Serial, "MQTT", "plugin activated");
@@ -37,7 +50,7 @@ void MqttPlugin::connect()
if (!client->connected()) if (!client->connected())
{ {
PRINT_MSG(Serial, "MQTT", "try connect"); PRINT_MSG(Serial, "MQTT", "try connect");
if (client->connect(mqttConfig.clientName)) if (client->connect(clientName.c_str()))
{ {
// bind handlers to all local subscriptions // bind handlers to all local subscriptions
if (!subscribed) if (!subscribed)
@@ -48,7 +61,7 @@ void MqttPlugin::connect()
subscribe(localSub.first.c_str(), bind(&MqttPlugin::upstreamHandler, this, localSub.first.c_str(), _1)); subscribe(localSub.first.c_str(), bind(&MqttPlugin::upstreamHandler, this, localSub.first.c_str(), _1));
// subscribe topic on remote queue to dispatch messages to local subscriptions // subscribe topic on remote queue to dispatch messages to local subscriptions
client->subscribe( client->subscribe(
(String(mqttConfig.inTopicRoot) + String(localSub.first.c_str())) (inTopicRoot + String(localSub.first.c_str()))
.c_str()); .c_str());
} }
subscribed = true; subscribed = true;
@@ -61,9 +74,10 @@ void MqttPlugin::connect()
void MqttPlugin::upstreamHandler(String topic, String msg) void MqttPlugin::upstreamHandler(String topic, String msg)
{ {
// publish message on remote queue // publish message on remote queue
String remoteTopic = String(mqttConfig.outTopicRoot) + topic; String remoteTopic = outTopicRoot + topic;
Serial.println(remoteTopic);
client->publish(remoteTopic.c_str(), msg.c_str()); client->publish(remoteTopic.c_str(), msg.c_str());
PRINT_MSG(Serial, "MQTT", String("publish remote: " + remoteTopic).c_str()); PRINT_MSG(Serial, "MQTT", remoteTopic.c_str());
} }
void MqttPlugin::downstreamHandler(char *topic, uint8_t *payload, unsigned int length) void MqttPlugin::downstreamHandler(char *topic, uint8_t *payload, unsigned int length)
@@ -75,7 +89,7 @@ void MqttPlugin::downstreamHandler(char *topic, uint8_t *payload, unsigned int l
free(cleanPayload); free(cleanPayload);
// substract the topic root and publish msg on local topic // substract the topic root and publish msg on local topic
int topicRootLength = String(mqttConfig.inTopicRoot).length(); int topicRootLength = inTopicRoot.length();
String localTopic = String(topic).substring(topicRootLength); String localTopic = String(topic).substring(topicRootLength);
publish(localTopic, msg); publish(localTopic, msg);

9
src/MqttPlugin.h
View File

@@ -17,7 +17,6 @@ class MqttPlugin : public Plugin
{ {
public: public:
PubSubClient *client; PubSubClient *client;
MqttConfigJson mqttConfig;
MqttPlugin(MqttConfig cfg); MqttPlugin(MqttConfig cfg);
@@ -32,10 +31,16 @@ class MqttPlugin : public Plugin
Task connectTask; Task connectTask;
Task processTask; Task processTask;
bool subscribed = false; bool subscribed = false;
String brokerHost;
int brokerPort;
String clientName;
String inTopicRoot;
String outTopicRoot;
void applyConfig(MqttConfig cfg);
void applyConfigFromFile(const char* fileName);
void enableConnectTask(Scheduler *scheduler); void enableConnectTask(Scheduler *scheduler);
void enableProcessTask(Scheduler *scheduler); void enableProcessTask(Scheduler *scheduler);
/** /**
* Connects to MQTT server and subscribes all topics available in the mediator * Connects to MQTT server and subscribes all topics available in the mediator
* to the corresponding topic on the remote queue by prefixing the topic with inRootTopic. * to the corresponding topic on the remote queue by prefixing the topic with inRootTopic.

2
src/examples/basic/config.h
View File

@@ -32,5 +32,7 @@
#define WEB_DEFAULT_FILE "index.html" #define WEB_DEFAULT_FILE "index.html"
#define WEB_PORT 80 #define WEB_PORT 80
#define MQTT_CONFIG_FILE "/mqttConfig.json"
#endif #endif

2
src/examples/basic/main.cpp
View File

@@ -26,7 +26,7 @@ void setup()
}); });
sprocket->addTask(publishHeap); sprocket->addTask(publishHeap);
sprocket->addPlugin( sprocket->addPlugin(
new MqttPlugin({"sprocket", "broker.lan", 1883, "wirelos/mqttSprocket-in/", "wirelos/mqttSprocket-out/"})); new MqttPlugin({"mqttSprocket", "192.168.1.2", 1883, "wirelos/mqttSprocket-in/", "wirelos/mqttSprocket-out/"}));
network = new WiFiNet( network = new WiFiNet(
SPROCKET_MODE, SPROCKET_MODE,