/* code generated by beremiz MQTT extension */
#include <stdint.h>
#include <unistd.h>
#include <pthread.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include "frozen.h"
#include "MQTTClient.h"
#include "MQTTClientPersistence.h"
#include "POUS.h"
#define _Log(level, ...) \
{{ \
char mstr[256]; \
snprintf(mstr, 255, __VA_ARGS__); \
LogMessage(level, mstr, strlen(mstr)); \
printf(__VA_ARGS__); \
fflush(stdout); \
}}
#define LogInfo(...) _Log(LOG_INFO, __VA_ARGS__);
#define LogError(...) _Log(LOG_CRITICAL, __VA_ARGS__);
#define LogWarning(...) _Log(LOG_WARNING, __VA_ARGS__);
// Selected debug level for paho stack
// can be:
// MQTTCLIENT_TRACE_PROTOCOL, MQTTCLIENT_TRACE_MAXIMUM, MQTTCLIENT_TRACE_ERROR
#define MQTT_DEBUG_LEVEL MQTTCLIENT_TRACE_ERROR
void trace_callback(enum MQTTCLIENT_TRACE_LEVELS level, char* message)
{{
if(level >= MQTT_DEBUG_LEVEL)
{{
int beremiz_log_level = (level >= MQTTCLIENT_TRACE_ERROR ) ? LOG_CRITICAL :
(level > MQTTCLIENT_TRACE_MINIMUM) ? LOG_WARNING :
LOG_INFO;
_Log(beremiz_log_level,"Paho MQTT Trace : %s\n", message);
}}
}}
#define CHANGED 1
#define UNCHANGED 0
#define DECL_VAR(iec_type, C_type, c_loc_name) \
static C_type PLC_##c_loc_name##_buf; \
static C_type MQTT_##c_loc_name##_buf; \
static int MQTT_##c_loc_name##_state = UNCHANGED; /* systematically published at init */ \
C_type *c_loc_name = &PLC_##c_loc_name##_buf;
{decl}
/* JSON topic content encoding macros matching "json_decl" in substitution*/
#define format_BOOL "%B"
#define format_SINT "%hhd"
#define format_USINT "%uhhd"
#define format_INT "%hd"
#define format_UINT "%uhd"
#define format_DINT "%d"
#define format_UDINT "%ud"
#define format_LINT "%ld"
#define format_ULINT "%uld"
#define format_REAL "%f"
#define format_LREAL "%Lf"
#define format_STRING "%*s"
#define format_separator ", "
#define format_SIMPLE(C_type, name, _A) #name " : " format_##C_type
#define format_OBJECT(C_type, name, _A) #name " : {{ " TYPE_##C_type(format, _A) " }}"
#define arg_separator ,
#define arg_SIMPLE(C_type, name, data_ptr) data_ptr->name
#define arg_OBJECT(C_type, name, data_ptr) TYPE_##C_type(arg, (&data_ptr->name))
#define DECL_JSON_INPUT(C_type, c_loc_name) \
int json_parse_##c_loc_name(char *json, const int len, void *void_ptr) {{ \
C_type *struct_ptr = (C_type *)void_ptr; \
return json_scanf(json, len, "{{" TYPE_##C_type(format,) "}}", TYPE_##C_type(arg, struct_ptr)); \
}}
/* Pre-allocated json output buffer for json_printf */
#define json_out_size 1<<12 // 4K
static char json_out_buf[json_out_size] = {{0,}};
static int json_out_len = 0;
#define DECL_JSON_OUTPUT(C_type, c_loc_name) \
int json_gen_##c_loc_name(C_type *struct_ptr) {{ \
struct json_out out = JSON_OUT_BUF(json_out_buf, json_out_size); \
json_out_len = json_printf(&out, "{{" TYPE_##C_type(format,) "}}", TYPE_##C_type(arg, struct_ptr)); \
if(json_out_len > json_out_size){{ \
json_out_len = 0; \
return -EOVERFLOW; \
}} \
return 0; \
}}
{json_decl}
static MQTTClient client;
#ifdef USE_MQTT_5
static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;
#else
static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
#endif
MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
/* condition to quit publish thread */
static int MQTT_stop_thread = 0;
/* condition to wakeup publish thread */
static int MQTT_any_pub_var_changed = 0;
/* Keep track of connection state */
static volatile int MQTT_is_disconnected = 1;
/* mutex to keep incoming PLC data consistent */
static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;
/* mutex to keep outgoing PLC data consistent, and protect MQTT_any_pub_var_changed */
static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;
/* wakeup publish thread when PLC changed published variable */
static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;
/* thread that handles publish and reconnection */
static pthread_t MQTT_thread;
#define INIT_TOPIC(topic, iec_type, c_loc_name) \
{{#topic, &MQTT_##c_loc_name##_buf, &MQTT_##c_loc_name##_state, 0, .vartype = iec_type##_ENUM}},
#define INIT_JSON_TOPIC(topic, iec_type, c_loc_name) \
{{#topic, &MQTT_##c_loc_name##_buf, &MQTT_##c_loc_name##_state, 1, .json_parse_func=json_parse_##c_loc_name}},
typedef int (*json_parse_func_t)(char *json, int len, void *void_ptr);
static struct {{
const char *topic; //null terminated topic string
void *mqtt_pdata; // pointer to data from/for MQTT stack
int *mqtt_pchanged; // pointer to changed flag
int is_json_type;
union {{
__IEC_types_enum vartype;
json_parse_func_t json_parse_func;
}};
}} topics [] = {{
{topics}
}};
void __cleanup_{locstr}(void)
{{
int rc;
/* stop publish thread */
MQTT_stop_thread = 1;
if (pthread_mutex_lock(&MQTT_thread_wakeup_mutex) == 0){{
/* unblock publish thread so that it can stop normally */
pthread_cond_signal(&MQTT_thread_wakeup);
pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);
}}
pthread_join(MQTT_thread, NULL);
#ifdef USE_MQTT_5
if (rc = MQTTClient_disconnect5(client, 5000, MQTTREASONCODE_SUCCESS, NULL) != MQTTCLIENT_SUCCESS)
#else
if (rc = MQTTClient_disconnect(client, 5000) != MQTTCLIENT_SUCCESS)
#endif
{{
LogError("MQTT Failed to disconnect, return code %d\n", rc);
}}
MQTTClient_destroy(&client);
}}
void connectionLost(void* context, char* reason)
{{
int rc;
LogWarning("ConnectionLost, reconnecting\\n");
if (pthread_mutex_lock(&MQTT_thread_wakeup_mutex) == 0){{
/* unblock publish thread so that it can reconnect */
MQTT_is_disconnected = 1;
pthread_cond_signal(&MQTT_thread_wakeup);
pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);
}}
}}
int messageArrived(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{{
int low = 0;
int size = sizeof(topics) / sizeof(topics[0]);
int high = size - 1;
int mid;
int is_json_type;
// bisect topic among subscribed topics
while (low <= high) {{
int res;
mid = low + (high - low) / 2;
res = strncmp(topics[mid].topic, topicName, topicLen);
// Check if key is present at mid
if (res == 0)
goto found;
// If key greater, ignore left half
if (res < 0)
low = mid + 1;
// If key is smaller, ignore right half
else
high = mid - 1;
}}
// If we reach here, then the element was not present
LogWarning("MQTT unknown topic: %s", topicName);
goto exit;
found:
is_json_type = topics[mid].is_json_type;
if(is_json_type || __get_type_enum_size(topics[mid].vartype) == message->payloadlen){{
if (pthread_mutex_lock(&MQTT_retrieve_mutex) == 0){{
if(is_json_type){{
(topics[mid].json_parse_func)((char*)message->payload, message->payloadlen, topics[mid].mqtt_pdata);
}} else {{
memcpy(topics[mid].mqtt_pdata, (char*)message->payload, message->payloadlen);
}}
*topics[mid].mqtt_pchanged = 1;
pthread_mutex_unlock(&MQTT_retrieve_mutex);
}}
}} else {{
LogWarning("MQTT wrong payload size for topic: %s. Should be %d, but got %d.",
topicName, __get_type_enum_size(topics[mid].vartype), message->payloadlen);
}}
exit:
MQTTClient_freeMessage(&message);
MQTTClient_free(topicName);
return 1;
}}
#define INIT_NoAuth() \
LogInfo("MQTT Init no auth\n");
#define INIT_x509(Verify, KeyStore, TrustStore) \
LogInfo("MQTT Init x509 with %s,%s\n", KeyStore?KeyStore:"NULL", TrustStore?TrustStore:"NULL")\
ssl_opts.verify = Verify; \
ssl_opts.keyStore = KeyStore; \
ssl_opts.trustStore = TrustStore; \
conn_opts.ssl = &ssl_opts;
#define INIT_PSK(Secret, ID) \
LogError("MQTT PSK NOT IMPLEMENTED\n") \
/* LogInfo("MQTT Init PSK for ID %s\n", ID) */ \
/* ssl_opts.ssl_psk_cb = TODO; */ \
/* ssl_opts.ssl_psk_context = TODO; */ \
conn_opts.ssl = &ssl_opts;
#define INIT_UserPassword(User, Password) \
LogInfo("MQTT Init UserPassword %s,%s\n", User, Password); \
conn_opts.username = User; \
conn_opts.password = Password;
#ifdef USE_MQTT_5
#define _SUBSCRIBE(Topic, QoS) \
MQTTResponse response = MQTTClient_subscribe5(client, #Topic, QoS, NULL, NULL); \
/* when using MQTT5 responce code is 1 for some reason even if no error */ \
rc = response.reasonCode == 1 ? MQTTCLIENT_SUCCESS : response.reasonCode; \
MQTTResponse_free(response);
#else
#define _SUBSCRIBE(Topic, QoS) \
rc = MQTTClient_subscribe(client, #Topic, QoS);
#endif
#define INIT_SUBSCRIPTION(Topic, QoS) \
{{ \
int rc; \
_SUBSCRIBE(Topic, QoS) \
if (rc != MQTTCLIENT_SUCCESS) \
{{ \
LogError("MQTT client failed to subscribe to '%s', return code %d\n", #Topic, rc); \
}} \
}}
#ifdef USE_MQTT_5
#define _PUBLISH(Topic, QoS, cstring_size, cstring_ptr, Retained) \
MQTTResponse response = MQTTClient_publish5(client, #Topic, cstring_size, \
cstring_ptr, QoS, Retained, NULL, NULL); \
rc = response.reasonCode; \
MQTTResponse_free(response);
#else
#define _PUBLISH(Topic, QoS, cstring_size, cstring_ptr, Retained) \
rc = MQTTClient_publish(client, #Topic, cstring_size, \
cstring_ptr, QoS, Retained, NULL);
#endif
#define PUBLISH_SIMPLE(Topic, QoS, C_type, c_loc_name, Retained) \
_PUBLISH(Topic, QoS, sizeof(C_type), &MQTT_##c_loc_name##_buf, Retained)
#define PUBLISH_JSON(Topic, QoS, C_type, c_loc_name, Retained) \
int res = json_gen_##c_loc_name(&MQTT_##c_loc_name##_buf); \
if(res == 0) {{ \
_PUBLISH(Topic, QoS, json_out_len, json_out_buf, Retained) \
}}
#define INIT_PUBLICATION(encoding, Topic, QoS, C_type, c_loc_name, Retained) \
{{ \
int rc; \
PUBLISH_##encoding(Topic, QoS, C_type, c_loc_name, Retained) \
if (rc != MQTTCLIENT_SUCCESS) \
{{ \
LogError("MQTT client failed to init publication of '%s', return code %d\n", #Topic, rc);\
/* TODO update status variable accordingly */ \
}} \
}}
#define PUBLISH_CHANGE(encoding, Topic, QoS, C_type, c_loc_name, Retained) \
if(MQTT_##c_loc_name##_state == CHANGED) \
{{ \
int rc; \
PUBLISH_##encoding(Topic, QoS, C_type, c_loc_name, Retained) \
if (rc != MQTTCLIENT_SUCCESS) \
{{ \
LogError("MQTT client failed to publish '%s', return code %d\n", #Topic, rc); \
/* TODO update status variable accordingly */ \
}} else {{ \
MQTT_##c_loc_name##_state = UNCHANGED; \
}} \
}}
static int _connect_mqtt(void)
{{
int rc;
#ifdef USE_MQTT_5
MQTTProperties props = MQTTProperties_initializer;
MQTTProperties willProps = MQTTProperties_initializer;
MQTTResponse response = MQTTResponse_initializer;
response = MQTTClient_connect5(client, &conn_opts, &props, &willProps);
rc = response.reasonCode;
MQTTResponse_free(response);
#else
rc = MQTTClient_connect(client, &conn_opts);
#endif
if (rc != MQTTCLIENT_SUCCESS) {{
MQTT_is_disconnected = 1;
return rc;
}}else{{
MQTT_is_disconnected = 0;
}}
{init_pubsub}
return MQTTCLIENT_SUCCESS;
}}
static void *__MQTT_thread_proc(void *_unused) {{
int rc = 0;
while((rc = pthread_mutex_lock(&MQTT_thread_wakeup_mutex)) == 0 && !MQTT_stop_thread){{
int do_publish;
pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);
if(MQTT_is_disconnected)
{{
/* TODO growing retry delay */
/* TODO max retry delay as config parameter */
sleep(5);
rc = _connect_mqtt();
if (rc == MQTTCLIENT_SUCCESS) {{
LogInfo("MQTT Reconnected\n");
}} else {{
LogError("MQTT Reconnect Failed, return code %d\n", rc);
}}
}}
if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)
{{
/* publish changes, and reset variable's state to UNCHANGED */
{publish_changes}
MQTT_any_pub_var_changed = 0;
}}
pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);
if(MQTT_stop_thread) break;
}}
if(!MQTT_stop_thread){{
/* if thread exits outside of normal shutdown, report error*/
LogError("MQTT client thread exited unexpectedly, return code %d\n", rc);
}}
}}
int __init_{locstr}(int argc,char **argv)
{{
char *uri = "{uri}";
char *clientID = "{clientID}";
int rc;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
#ifdef USE_MQTT_5
conn_opts.MQTTVersion = MQTTVERSION_5;
conn_opts.cleanstart = 1;
createOpts.MQTTVersion = MQTTVERSION_5;
#else
conn_opts.cleansession = 1;
#endif
MQTTClient_setTraceCallback(trace_callback);
MQTTClient_setTraceLevel(MQTT_DEBUG_LEVEL);
rc = MQTTClient_createWithOptions(
&client, uri, clientID, MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);
if (rc != MQTTCLIENT_SUCCESS)
{{
LogError("MQTT Failed to create client, return code %d\n", rc);
goto exit_error;
}}
rc = MQTTClient_setCallbacks(client, NULL, connectionLost, messageArrived, NULL);
if (rc != MQTTCLIENT_SUCCESS)
{{
LogError("MQTT Failed to set callbacks, return code %d\n", rc);
goto exit_error;
}}
{init}
rc = _connect_mqtt();
if (rc == MQTTCLIENT_SUCCESS) {{
LogInfo("MQTT Connected\n");
}} else {{
LogError("MQTT Connect Failed, return code %d\n", rc);
// Connect error at init is fine, publish thread will retry later
}}
/* start MQTT thread */
MQTT_stop_thread = 0;
rc = pthread_create(&MQTT_thread, NULL, &__MQTT_thread_proc, NULL);
if (rc != 0) {{
LogError("MQTT cannot create thread, return code %d\n", rc);
goto exit_error;
}}
return 0;
exit_error:
MQTTClient_destroy(&client);
return rc;
}}
#define READ_VALUE(c_loc_name, C_type) \
if(MQTT_##c_loc_name##_state == CHANGED){{ \
/* TODO care about endianess */ \
PLC_##c_loc_name##_buf = MQTT_##c_loc_name##_buf; \
MQTT_##c_loc_name##_state = UNCHANGED; \
}}
void __retrieve_{locstr}(void)
{{
if (pthread_mutex_trylock(&MQTT_retrieve_mutex) == 0){{
{retrieve}
pthread_mutex_unlock(&MQTT_retrieve_mutex);
}}
}}
#define WRITE_VALUE(c_loc_name, C_type) \
/* TODO care about endianess */ \
if(memcmp(&MQTT_##c_loc_name##_buf, &PLC_##c_loc_name##_buf, sizeof(C_type))){{ \
MQTT_##c_loc_name##_buf = PLC_##c_loc_name##_buf; \
MQTT_##c_loc_name##_state = CHANGED; \
MQTT_any_pub_var_changed = 1; \
}}
void __publish_{locstr}(void)
{{
if (pthread_mutex_trylock(&MQTT_thread_wakeup_mutex) == 0){{
MQTT_any_pub_var_changed = 0;
/* copy PLC_* variables to MQTT_*, and mark those who changed */
{publish}
/* if any change detcted, unblock publish thread */
if(MQTT_any_pub_var_changed || MQTT_is_disconnected){{
pthread_cond_signal(&MQTT_thread_wakeup);
}}
pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);
}} else {{
/* TODO if couldn't lock mutex set status variable accordingly */
}}
}}