irext/iris-kit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implemented MQTT connection to EMQX dev

Browse Source
This commit is contained in:
strawmanbobi
2024-01-11 13:00:45 +08:00
parent 34259c8023
commit f81a3f6b2e
15 changed files with 162 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
src/IRISKitIoT.cpp
View File

@@ -23,6 +23,7 @@
#include <Arduino.h>
#include <WString.h>
#include <PubSubClient.h>
#include "IRISKitSerials.h"
#include "IRISKitIoT.h"
@@ -31,35 +32,41 @@
#include "IRISKit.h"
#define TOPIC_NAME_MAX (64)
#define IOT_RETRY_MAX (3)
#define IRIS_KIT_PK_DEV "a1WlzsJh50b"
#define IRIS_KIT_PK_REL "a1ihYt1lqGH"
#define TOPIC_DOWNSTREAM_DEV "/user/iris/downstream_dev"
#define TOPIC_UPSTREAM_DEV "/user/iris/upstream_dev"
#define TOPIC_DOWNSTREAM_REL "/user/iris/downstream"
#define TOPIC_UPSTREAM_REL "/user/iris/upstream"
#define TOPIC_DOWNSTREAM_DEV "/user/iris_kit_downstream_dev"
#define TOPIC_UPSTREAM_DEV "/user/iris_kit_upstream_dev"
#define TOPIC_DOWNSTREAM_REL "/user/iris_kit_downstream"
#define TOPIC_UPSTREAM_REL "/user/iris_kit_upstream"
String g_mqtt_server = "";
String g_product_key = "";
String g_device_name = "";
String g_device_secret = "";
String g_region_id = "cn-shanghai";
String g_mqtt_client_id = "";
String g_mqtt_user_name = "";
String g_mqtt_password = "";
String g_upstream_topic = "";
String g_downstream_topic = "";
int g_app_id = 0;
int g_mqtt_port = 1883;
int g_app_id = 0;
static bool downstream_topic_subscribed = false;
static ep_state_t endpoint_state = FSM_IDLE;
static void registerCallback(const char* topic, int qos);
static void irisIrextIoTCallback(const char *topic, uint8_t *data, int length);
static int connectToMQTTBroker();
static void irisIrextIoTCallback(char *topic, uint8_t *data, uint32_t length);
static int iot_retry = 0;
void connectToIrextIoT() {
static PubSubClient mqtt_client(wifi_client);
int connectToIrextIoT() {
downstream_topic_subscribed = false;
int conn_ret = -1;
if (g_product_key.equals(IRIS_KIT_PK_DEV)) {
g_upstream_topic = "/" + g_product_key + "/" + g_device_name + TOPIC_UPSTREAM_DEV;
@@ -69,70 +76,77 @@ void connectToIrextIoT() {
g_downstream_topic = "/" + g_product_key + "/" + g_device_name + TOPIC_DOWNSTREAM_REL;
} else {
ERRORF("IRIS Kit release key is not supported yet\n");
return;
factoryReset();
return -1;
}
INFOF("Try connecting to Aliyun IoT : %s, %s, %s, %s\n",
g_product_key.c_str(), g_device_name.c_str(), g_device_secret.c_str(), g_region_id.c_str());
g_mqtt_user_name = getDeviceID();
/*
if (0 == iot.begin(wifi_client, g_product_key.c_str(), g_device_name.c_str(), g_device_secret.c_str(),
g_region_id.c_str())) {
sendIrisKitConnect();
INFOF("Try connecting to IRext IoT %s:%d, client_id = %s, user_name = %s, password.size = %d\n",
g_mqtt_server.c_str(), g_mqtt_port,
g_mqtt_client_id.c_str(), g_mqtt_user_name.c_str(), g_mqtt_password.length());
mqtt_client.setServer(g_mqtt_server.c_str(), g_mqtt_port);
mqtt_client.setCallback(irisIrextIoTCallback);
conn_ret = connectToMQTTBroker();
if (0 != conn_ret) {
ERRORLN("Something may went wrong with your credential, please retry connect to Wifi...");
factoryReset();
return -1;
}
*/
INFOLN("Aliyun IoT connect done");
// send connect request
sendIrisKitConnect();
return 0;
}
void irextIoTKeepAlive() {
/*
iot.loop();
*/
if (!mqtt_client.connected()) {
connectToMQTTBroker();
}
mqtt_client.loop();
}
// not only for IRIS related topic based session
void sendData(const char* topic, const uint8_t *data, int length) {
/*
iot.sendCustomData(topic, data, length);
*/
mqtt_client.publish(topic, data, length);
}
void* getSession() {
return NULL;
return &mqtt_client;
}
void checkIrextIoT() {
int mqttStatus = 0;
sendIrisKitHeartBeat();
}
if (0 == mqttStatus) {
iot_retry = 0;
if (false == downstream_topic_subscribed) {
INFOF("subscribe topic : %s\n", g_downstream_topic.c_str());
registerCallback(g_downstream_topic.c_str(), 0);
downstream_topic_subscribed = true;
static int connectToMQTTBroker() {
int retry_times = 0;
while (!mqtt_client.connected() && retry_times < MQTT_RETRY_MAX) {
INFOF("Connecting to MQTT Broker as %s.....\n", g_mqtt_client_id.c_str());
if (mqtt_client.connect(g_mqtt_client_id.c_str(), g_mqtt_user_name.c_str(), g_mqtt_password.c_str())) {
INFOF("Connected to MQTT broker\n");
mqtt_client.subscribe(g_downstream_topic.c_str());
} else {
sendIrisKitHeartBeat();
ERRORF("Failed to connect to MQTT broker, rc = %d\n", mqtt_client.state());
INFOF(" try again in 5 seconds\n");
retry_times++;
delay(MQTT_RETRY_DELAY);
}
}
if (mqtt_client.connected()) {
INFOF("IRext IoT connect done\n");
return 0;
} else {
INFOF("Alink MQTT check failed, retry = %d\n", iot_retry);
iot_retry++;
}
if (iot_retry >= IOT_RETRY_MAX) {
ERRORLN("Alink could not established, something went wrong, reset...");
factoryReset();
ERRORF("IRext IoT failed to connect\n");
return -1;
}
}
static void registerCallback(const char* topic, int qos) {
/*
if (iot.subscribe(topic, qos)) {
INFOLN("topic subscribed");
iot.registerCustomCallback(irisIrextIoTCallback);
}
*/
}
static void irisIrextIoTCallback(const char *topic, uint8_t *data, int length) {
static void irisIrextIoTCallback(char *topic, uint8_t *data, uint32_t length) {
INFOF("downstream message received, topic = %s, length = %d\n", topic, length);
if (NULL != g_downstream_topic.c_str() && 0 == strcmp(topic, g_downstream_topic.c_str())) {
handleIrisKitMessage((const char*) data, length);