Skip to content

Commit 164bea9

Browse files
pvtompvtom
committed
v1.6
1 parent 4b12f61 commit 164bea9

File tree

2 files changed

+71
-44
lines changed

2 files changed

+71
-44
lines changed

‎RscpMain.cpp

Lines changed: 59 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919
#include <mutex>
2020
#include <fcntl.h>
2121

22-
#define RSCP2P "1.5"
23-
#define RSCP2P_LONG "1.5.3.33"
22+
#define RSCP2P "1.6"
23+
#define RSCP2P_LONG "1.6.3.34"
2424

2525
#define AES_KEY_SIZE 32
2626
#define AES_BLOCK_SIZE 32
@@ -739,15 +739,15 @@ void pushNotSupportedTag(uint32_t container, uint32_t tag) {
739739
return;
740740
}
741741

742-
bool existsAdditionalTag(uint32_t container, uint32_t tag, int index) {
742+
bool existsAdditionalTag(uint32_t container, uint32_t tag, int index, int order) {
743743
for (std::vector<RSCP_MQTT::additional_tags_t>::iterator it = RSCP_MQTT::AdditionalTags.begin(); it != RSCP_MQTT::AdditionalTags.end(); ++it) {
744-
if ((it->req_container == container) && (it->req_tag == tag) && (it->req_index == index)) return(true);
744+
if ((it->req_container == container) && (it->req_tag == tag) && (it->req_index == index) && (it->order == order)) return(true);
745745
}
746746
return(false);
747747
}
748748

749749
void pushAdditionalTag(uint32_t req_container, uint32_t req_tag, int req_index, int order, bool one_shot) {
750-
if (existsAdditionalTag(req_container, req_tag, req_index)) return;
750+
if (existsAdditionalTag(req_container, req_tag, req_index, order)) return;
751751
RSCP_MQTT::additional_tags_t v;
752752
v.req_container = req_container;
753753
v.req_tag = req_tag;
@@ -759,38 +759,40 @@ void pushAdditionalTag(uint32_t req_container, uint32_t req_tag, int req_index,
759759
return;
760760
}
761761

762-
bool updateRawData(char *topic, char *payload) {
762+
void initRawData() {
763+
for (std::vector<RSCP_MQTT::raw_data_t>::iterator it = RSCP_MQTT::rawData.begin(); it != RSCP_MQTT::rawData.end(); ++it) {
764+
it->handled = false;
765+
it->changed = false;
766+
}
767+
return;
768+
}
769+
770+
int mergeRawData(char *topic, char *payload, bool *changed) {
771+
int i = 0;
763772
if (topic && payload) {
764-
for (std::vector<RSCP_MQTT::mqtt_data_t>::iterator it = RSCP_MQTT::rawData.begin(); it != RSCP_MQTT::rawData.end(); ++it) {
765-
if (!strcmp(it->topic, topic)) {
773+
for (std::vector<RSCP_MQTT::raw_data_t>::iterator it = RSCP_MQTT::rawData.begin(); it != RSCP_MQTT::rawData.end(); ++it) {
774+
if (!strcmp(it->topic, topic) && !it->handled && (i == it->nr)) {
766775
if (strcmp(it->payload, payload)) {
767776
if (strlen(it->payload) != strlen(payload)) it->payload = (char *)realloc(it->payload, strlen(payload) + 1);
768777
strcpy(it->payload, payload);
778+
it->changed = true;
779+
*changed = true;
769780
}
770-
return(true);
781+
it->handled = true;
782+
return(i);
771783
}
784+
if (!strcmp(it->topic, topic) && it->handled) i++;
772785
}
773-
}
774-
return(false);
775-
}
776-
777-
void insertRawData(char *topic, char *payload) {
778-
if (topic && payload) {
779-
RSCP_MQTT::mqtt_data_t v;
786+
RSCP_MQTT::raw_data_t v;
780787
v.topic = strdup(topic);
781788
v.payload = strdup(payload);
789+
v.handled = true;
790+
v.changed = true;
791+
v.nr = i;
782792
RSCP_MQTT::rawData.push_back(v);
793+
*changed = true;
783794
}
784-
return;
785-
}
786-
787-
char *readRawData(char *topic) {
788-
for (std::vector<RSCP_MQTT::mqtt_data_t>::iterator it = RSCP_MQTT::rawData.begin(); it != RSCP_MQTT::rawData.end(); ++it) {
789-
if (!strcmp(it->topic, topic)) {
790-
return(it->payload);
791-
}
792-
}
793-
return(NULL);
795+
return(i);
794796
}
795797

796798
void refreshCache(std::vector<RSCP_MQTT::cache_t> & v, char *payload) {
@@ -942,6 +944,13 @@ float getFloatValue(std::vector<RSCP_MQTT::cache_t> & c, uint32_t container, uin
942944
return(value);
943945
}
944946

947+
void resetHandleFlag(std::vector<RSCP_MQTT::cache_t> & c) {
948+
for (std::vector<RSCP_MQTT::cache_t>::iterator it = c.begin(); it != c.end(); ++it) {
949+
it->handled = false;
950+
}
951+
return;
952+
}
953+
945954
void preparePayload(RscpProtocol *protocol, SRscpValue *response, char **buf) {
946955
switch (response->dataType) {
947956
case RSCP::eTypeBool: {
@@ -1000,8 +1009,7 @@ int storeResponseValue(std::vector<RSCP_MQTT::cache_t> & c, RscpProtocol *protoc
10001009
int rc = -1;
10011010

10021011
for (std::vector<RSCP_MQTT::cache_t>::iterator it = c.begin(); it != c.end(); ++it) {
1003-
if ((it->container > container) && (it->tag > response->tag)) break;
1004-
if ((!it->container || (it->container == container)) && (it->tag == response->tag) && (it->index == index)) {
1012+
if ((!it->container || (it->container == container)) && (it->tag == response->tag) && (it->index == index) && !it->handled) {
10051013
switch (response->dataType) {
10061014
case RSCP::eTypeBool: {
10071015
if (protocol->getValueAsBool(response)) strcpy(buf, "true");
@@ -1144,6 +1152,8 @@ int storeResponseValue(std::vector<RSCP_MQTT::cache_t> & c, RscpProtocol *protoc
11441152
if ((atoi(it->payload) == 0) && (battery_soc > 1)) snprintf(it->payload, PAYLOAD_SIZE, "%d", battery_soc--);
11451153
else battery_soc = atoi(it->payload);
11461154
}
1155+
it->handled = true;
1156+
break;
11471157
}
11481158
}
11491159
return(rc);
@@ -1970,21 +1980,24 @@ void createRequest(SRscpFrameBuffer * frameBuffer) {
19701980
return;
19711981
}
19721982

1973-
void publishRaw(RscpProtocol *protocol, SRscpValue *response, char *topic) {
1974-
char *payload_new = (char *)malloc(PAYLOAD_SIZE * sizeof(char) + 1);
1975-
char *payload_old = readRawData(topic);
1976-
memset(payload_new, 0, PAYLOAD_SIZE);
1977-
preparePayload(protocol, response, &payload_new);
1978-
if (payload_old && payload_new && strcmp(payload_new, "") && strcmp(payload_old, payload_new)) {
1979-
publishImmediately(topic, payload_new);
1980-
updateRawData(topic, payload_new);
1981-
} else if (!payload_old && payload_new && strcmp(payload_new, "")) {
1982-
publishImmediately(topic, payload_new);
1983-
insertRawData(topic, payload_new);
1984-
}
1985-
if (payload_new) free(payload_new);
1986-
return;
1987-
}
1983+
void publishRaw(RscpProtocol *protocol, SRscpValue *response, char *topic_in) {
1984+
char topic[TOPIC_SIZE];
1985+
char *payload = (char *)malloc(PAYLOAD_SIZE * sizeof(char) + 1);
1986+
bool changed = false;
1987+
memset(payload, 0, PAYLOAD_SIZE);
1988+
preparePayload(protocol, response, &payload);
1989+
1990+
int nr = mergeRawData(topic_in, payload, &changed);
1991+
if (nr > 0) {
1992+
if (snprintf(topic, TOPIC_SIZE, "%s/%d", topic_in, nr) >= TOPIC_SIZE) {
1993+
logMessage(cfg.logfile, (char *)__FILE__, __LINE__, (char *)"publishRaw: Buffer overflow\n");
1994+
return;
1995+
}
1996+
if (changed) publishImmediately(topic, payload);
1997+
} else if (changed) publishImmediately(topic_in, payload);
1998+
if (payload) free(payload);
1999+
return;
2000+
}
19882001

19892002
void handleRaw(RscpProtocol *protocol, SRscpValue *response, uint32_t *cache, int level) {
19902003
int l = level + 1;
@@ -2460,6 +2473,7 @@ static int processReceiveBuffer(const unsigned char * ucBuffer, int iLength) {
24602473
battery_nr = 0;
24612474
pm_nr = 0;
24622475
wb_nr = 0;
2476+
if (cfg.raw_mode) initRawData();
24632477
for (size_t i = 0; i < frame.data.size(); i++)
24642478
handleResponseValue(&protocol, &frame.data[i]);
24652479

@@ -2591,6 +2605,8 @@ static void mainLoop(void) {
25912605

25922606
gettimeofday(&start, NULL);
25932607

2608+
resetHandleFlag(RSCP_MQTT::RscpMqttCache);
2609+
25942610
// create an RSCP frame with requests to some example data
25952611
createRequest(&frameBuffer);
25962612

‎RscpMapping.h

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -152,7 +152,16 @@ typedef struct _mqtt_data_t {
152152
} mqtt_data_t;
153153

154154
std::queue<mqtt_data_t> mqttQ;
155-
std::vector<mqtt_data_t> rawData;
155+
156+
typedef struct _raw_data_t {
157+
char *topic;
158+
char *payload;
159+
int nr;
160+
bool handled;
161+
bool changed;
162+
} raw_data_t;
163+
164+
std::vector<raw_data_t> rawData;
156165

157166
typedef struct _cache_t {
158167
uint32_t container;
@@ -165,6 +174,7 @@ typedef struct _cache_t {
165174
int divisor;
166175
int bit_to_bool;
167176
bool history_log;
177+
bool handled;
168178
bool changed;
169179
bool influx;
170180
bool forced;
@@ -357,6 +367,7 @@ bool compareCache(RSCP_MQTT::cache_t c1, RSCP_MQTT::cache_t c2) {
357367
if (c1.container < c2.container) return(true);
358368
if ((c1.container == c2.container) && (c1.tag < c2.tag)) return(true);
359369
if ((c1.container == c2.container) && (c1.tag == c2.tag) && (c1.index < c2.index)) return(true);
370+
if ((c1.container == c2.container) && (c1.tag == c2.tag) && (c1.index == c2.index) && (strcmp(c1.topic, c2.topic) < 0)) return(true);
360371
return(false);
361372
}
362373

0 commit comments

Comments
 (0)