1

/* code generated by beremiz MQTT extension */

     2

     3

#include <stdint.h>

     4

#include <unistd.h>

     5

#include <pthread.h>

     6

#include <string.h>

     7

#include <stdio.h>

     8

#include <errno.h>

     9

    10

#include "frozen.h"

    11

    12

#include "MQTTClient.h"

    13

#include "MQTTClientPersistence.h"

    14

    15

#include "POUS.h"

    16

    17

#define _Log(level, ...)                                                                          \

    18

    {{                                                                                            \

    19

        char mstr[256];                                                                           \

    20

        snprintf(mstr, 255, __VA_ARGS__);                                                         \

    21

        LogMessage(level, mstr, strlen(mstr));                                                    \

    22

    }}

    23

    24

#define LogInfo(...) _Log(LOG_INFO, __VA_ARGS__);

    25

#define LogError(...) _Log(LOG_CRITICAL, __VA_ARGS__);

    26

#define LogWarning(...) _Log(LOG_WARNING, __VA_ARGS__);

    27

    28

// Selected debug level for paho stack

    29

// can be:

    30

// MQTTCLIENT_TRACE_PROTOCOL, MQTTCLIENT_TRACE_MAXIMUM, MQTTCLIENT_TRACE_ERROR

    31

#define MQTT_DEBUG_LEVEL MQTTCLIENT_TRACE_ERROR

    32

    33

static void trace_callback(enum MQTTCLIENT_TRACE_LEVELS level, char* message)

    34

{{

    35

    if(level >= MQTT_DEBUG_LEVEL)

    36

    {{

    37

        int beremiz_log_level = (level >= MQTTCLIENT_TRACE_ERROR ) ? LOG_CRITICAL :

    38

                                (level > MQTTCLIENT_TRACE_MINIMUM) ? LOG_WARNING : 

    39

                                LOG_INFO;

    40

        _Log(beremiz_log_level,"Paho MQTT Trace : %s\n", message);

    41

    }}

    42

}}

    43

    44

#define CHANGED 1

    45

#define UNCHANGED 0

    46

    47

extern INT CONFIG__MQTT_STATUS_{locstr};

    48

    49

#define DECL_VAR(iec_type, C_type, c_loc_name)                                                     \

    50

static C_type PLC_##c_loc_name##_buf;                                                              \

    51

static C_type MQTT_##c_loc_name##_buf;                                                             \

    52

static int MQTT_##c_loc_name##_state = UNCHANGED;  /* systematically published at init */          \

    53

C_type *c_loc_name = &PLC_##c_loc_name##_buf;

    54

    55

{decl}

    56

    57

/* JSON topic content encoding macros matching "json_decl" in substitution*/

    58

    59

#define printf_fmt_BOOL   "%B"

    60

#define printf_fmt_SINT   "%hhd"

    61

#define printf_fmt_USINT  "%uhhd"

    62

#define printf_fmt_INT    "%hd" 

    63

#define printf_fmt_UINT   "%uhd"

    64

#define printf_fmt_DINT   "%d" 

    65

#define printf_fmt_UDINT  "%ud"

    66

#define printf_fmt_LINT   "%ld"

    67

#define printf_fmt_ULINT  "%uld"

    68

#define printf_fmt_REAL   "%f"

    69

#define printf_fmt_LREAL  "%Lf"

    70

#define printf_fmt_STRING "%.*Q"

    71

    72

#define printf_fmt_separator ", "

    73

    74

#define printf_fmt_SIMPLE(C_type, C_name, name, _A) #name " : " printf_fmt_##C_type

    75

#define printf_fmt_OBJECT(C_type, C_name, name, _A) #name " : {{ " TYPE_##C_type(printf_fmt, _A) " }}"

    76

#define printf_fmt_ARRAY(C_type, C_name, name, _A) #name " : [ " TYPE_##C_type(printf_fmt, _A) " ]"

    77

#define printf_fmt_ARRAY_SIMPLE(C_type, index, _A) printf_fmt_##C_type

    78

#define printf_fmt_ARRAY_OBJECT(C_type, index, _A) "{{ " TYPE_##C_type(printf_fmt, _A) " }}"

    79

    80

#define scanf_fmt_BOOL   "%B"

    81

#define scanf_fmt_SINT   "%hhd"

    82

#define scanf_fmt_USINT  "%uhhd"

    83

#define scanf_fmt_INT    "%hd" 

    84

#define scanf_fmt_UINT   "%uhd"

    85

#define scanf_fmt_DINT   "%d" 

    86

#define scanf_fmt_UDINT  "%ud"

    87

#define scanf_fmt_LINT   "%ld"

    88

#define scanf_fmt_ULINT  "%uld"

    89

#define scanf_fmt_REAL   "%f"

    90

#define scanf_fmt_LREAL  "%Lf"

    91

#define scanf_fmt_STRING "%M"

    92

    93

#define scanf_fmt_separator ", "

    94

    95

#define scanf_fmt_SIMPLE(C_type, C_name, name, _A) #name " : " scanf_fmt_##C_type

    96

#define scanf_fmt_OBJECT(C_type, C_name, name, _A) #name " : {{ " TYPE_##C_type(scanf_fmt, _A) " }}"

    97

#define scanf_fmt_ARRAY(C_type, C_name, name, _A) #name " : [ " TYPE_##C_type(scanf_fmt, _A) " ]"

    98

#define scanf_fmt_ARRAY_SIMPLE(C_type, index, _A) scanf_fmt_##C_type

    99

#define scanf_fmt_ARRAY_OBJECT(C_type, index, _A) "{{ " TYPE_##C_type(scanf_fmt, _A) " }}"

   100

   101

#define   scanf_arg_BOOL(arg) arg

   102

#define   scanf_arg_SINT(arg) arg

   103

#define  scanf_arg_USINT(arg) arg

   104

#define    scanf_arg_INT(arg) arg

   105

#define   scanf_arg_UINT(arg) arg

   106

#define   scanf_arg_DINT(arg) arg

   107

#define  scanf_arg_UDINT(arg) arg

   108

#define   scanf_arg_LINT(arg) arg

   109

#define  scanf_arg_ULINT(arg) arg

   110

#define   scanf_arg_REAL(arg) arg

   111

#define  scanf_arg_LREAL(arg) arg

   112

#define scanf_arg_STRING(arg) scan_string, arg

   113

   114

#define scanf_args_separator ,

   115

   116

#define scanf_args_SIMPLE(C_type, C_name, name, data_ptr) scanf_arg_##C_type(&data_ptr->C_name)

   117

#define scanf_args_OBJECT(C_type, C_name, name, data_ptr) TYPE_##C_type(scanf_args, (&data_ptr->C_name))

   118

#define scanf_args_ARRAY(C_type, C_name, name, data_ptr) TYPE_##C_type(scanf_args, data_ptr->C_name.table)

   119

#define scanf_args_ARRAY_SIMPLE(C_type, index, data_ptr) scanf_arg_##C_type(&data_ptr[index])

   120

#define scanf_args_ARRAY_OBJECT(C_type, index, data_ptr) TYPE_##C_type(scanf_args, (&data_ptr[index]))

   121

   122

#define   printf_arg_BOOL(arg) arg

   123

#define   printf_arg_SINT(arg) arg

   124

#define  printf_arg_USINT(arg) arg

   125

#define    printf_arg_INT(arg) arg

   126

#define   printf_arg_UINT(arg) arg

   127

#define   printf_arg_DINT(arg) arg

   128

#define  printf_arg_UDINT(arg) arg

   129

#define   printf_arg_LINT(arg) arg

   130

#define  printf_arg_ULINT(arg) arg

   131

#define   printf_arg_REAL(arg) arg

   132

#define  printf_arg_LREAL(arg) arg

   133

#define printf_arg_STRING(arg) arg.len, arg.body 

   134

   135

#define printf_args_separator ,

   136

   137

#define printf_args_SIMPLE(C_type, C_name, name, data_ptr) printf_arg_##C_type(data_ptr->C_name)

   138

#define printf_args_OBJECT(C_type, C_name, name, data_ptr) TYPE_##C_type(printf_args, (&data_ptr->C_name))

   139

#define printf_args_ARRAY(C_type, C_name, name, data_ptr) TYPE_##C_type(printf_args, (&data_ptr->C_name.table))

   140

#define printf_args_ARRAY_SIMPLE(C_type, index, data_ptr) printf_arg_##C_type(data_ptr[index])

   141

#define printf_args_ARRAY_OBJECT(C_type, index, data_ptr) TYPE_##C_type(printf_args, (data_ptr[index]))

   142

   143

static void scan_string(const char *str, int len, void *user_data) {{

   144

	IEC_STRING *iecstr = (IEC_STRING*)user_data;

   145

	__strlen_t ieclen = len > STR_MAX_LEN ? STR_MAX_LEN : len;

   146

	memcpy(iecstr->body, str, ieclen);

   147

    iecstr->len = ieclen;

   148

}}

   149

   150

#define DECL_JSON_INPUT(C_type, c_loc_name) \

   151

static int json_parse_##c_loc_name(char *json, const int len, void *void_ptr) {{ \

   152

    C_type *struct_ptr = (C_type *)void_ptr; \

   153

    return json_scanf(json, len, "{{" TYPE_##C_type(scanf_fmt,) "}}", TYPE_##C_type(scanf_args, struct_ptr)); \

   154

}}

   155

   156

/* Pre-allocated json output buffer for json_printf */

   157

#define json_out_size 1<<12 // 4K

   158

static char json_out_buf[json_out_size] = {{0,}};

   159

static int json_out_len = 0;

   160

   161

#define DECL_JSON_OUTPUT(C_type, c_loc_name) \

   162

static int json_gen_##c_loc_name(C_type *struct_ptr) {{ \

   163

    struct json_out out = JSON_OUT_BUF(json_out_buf, json_out_size); \

   164

    json_out_len = json_printf(&out, "{{" TYPE_##C_type(printf_fmt,) "}}", TYPE_##C_type(printf_args, struct_ptr)); \

   165

    if(json_out_len > json_out_size){{ \

   166

        json_out_len = 0; \

   167

        return -EOVERFLOW; \

   168

    }} \

   169

    return 0; \

   170

}}

   171

   172

{json_decl}

   173

   174

static MQTTClient client;

   175

#ifdef USE_MQTT_5

   176

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;

   177

#else

   178

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

   179

#endif

   180

   181

MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;

   182

   183

/* condition to quit publish thread */

   184

static int MQTT_stop_thread = 0;

   185

   186

/* condition to wakeup publish thread */

   187

static int MQTT_any_pub_var_changed = 0;

   188

   189

/* Keep track of connection state */

   190

static volatile int MQTT_is_disconnected = 1;

   191

   192

/* mutex to keep incoming PLC data consistent */

   193

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

   194

   195

/* mutex to keep outgoing PLC data consistent, and protect MQTT_any_pub_var_changed */

   196

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

   197

   198

/* wakeup publish thread when PLC changed published variable */

   199

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

   200

   201

/* thread that handles publish and reconnection */

   202

static pthread_t MQTT_thread;

   203

   204

#define INIT_TOPIC(topic, iec_type, c_loc_name) \

   205

{{#topic, sizeof(#topic)-1, &MQTT_##c_loc_name##_buf, &MQTT_##c_loc_name##_state, .is_json_type=0, .vartype = iec_type##_ENUM}},

   206

   207

#define INIT_JSON_TOPIC(topic, iec_type, c_loc_name) \

   208

{{#topic, sizeof(#topic)-1, &MQTT_##c_loc_name##_buf, &MQTT_##c_loc_name##_state, .is_json_type=1, .json_parse_func=json_parse_##c_loc_name}},

   209

   210

typedef int (*json_parse_func_t)(char *json, int len, void *void_ptr);

   211

   212

static struct {{

   213

    const char *topic; //null terminated topic string

   214

    const unsigned int topic_len;

   215

    void *mqtt_pdata; // pointer to data from/for MQTT stack

   216

    int *mqtt_pstate; // pointer to changed flag

   217

    int is_json_type;

   218

    union {{

   219

       __IEC_types_enum vartype;

   220

       json_parse_func_t json_parse_func;

   221

    }};

   222

}} topics [] = {{

   223

{topics}

   224

}};

   225

   226

void __cleanup_{locstr}(void)

   227

{{

   228

    int rc;

   229

   230

    /* stop publish thread */

   231

    MQTT_stop_thread = 1;

   232

    if (pthread_mutex_lock(&MQTT_thread_wakeup_mutex) == 0){{

   233

        /* unblock publish thread so that it can stop normally */

   234

        pthread_cond_signal(&MQTT_thread_wakeup);

   235

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   236

    }}

   237

    pthread_join(MQTT_thread, NULL);

   238

   239

#ifdef USE_MQTT_5

   240

    if (rc = MQTTClient_disconnect5(client, 5000, MQTTREASONCODE_SUCCESS, NULL) != MQTTCLIENT_SUCCESS)

   241

#else

   242

    if (rc = MQTTClient_disconnect(client, 5000) != MQTTCLIENT_SUCCESS)

   243

#endif

   244

    {{

   245

        LogError("MQTT Failed to disconnect, return code %d\n", rc);

   246

    }}

   247

    MQTTClient_destroy(&client);

   248

}}

   249

   250

static void connectionLost(void* context, char* reason)

   251

{{

   252

    int rc;

   253

    LogWarning("ConnectionLost, reconnecting\\n");

   254

    if (pthread_mutex_lock(&MQTT_thread_wakeup_mutex) == 0){{

   255

        /* unblock publish thread so that it can reconnect */

   256

        MQTT_is_disconnected = 1;

   257

        pthread_cond_signal(&MQTT_thread_wakeup);

   258

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   259

    }}

   260

}}

   261

   262

   263

typedef int(*callback_fptr_t)(char* topic, char* data, uint32_t datalen);

   264

static callback_fptr_t __mqtt_python_callback_fptr_{name} = NULL;

   265

   266

static int messageArrived(void *context, char *topicName, int topicLen, MQTTClient_message *message)

   267

{{

   268

    int low = 0;

   269

    int size = sizeof(topics) / sizeof(topics[0]);

   270

    int high = size - 1;

   271

    int mid;

   272

    int is_json_type;

   273

   274

    // bisect topic among subscribed topics

   275

    while (low <= high) {{

   276

        int res, len, delta_len;

   277

        mid = low + (high - low) / 2;

   278

        if(topicLen == 0) {{

   279

            // zero topic len means null-terminated

   280

            topicLen = strlen(topicName);

   281

        }}

   282

   283

        len = topics[mid].topic_len;

   284

   285

        // keep track of length difference

   286

        delta_len = len - topicLen;

   287

   288

        // len = min(len, topicLen)

   289

        len = topicLen > len ? len : topicLen;

   290

   291

        // compare strings as far as possible

   292

        res = strncmp(topics[mid].topic, topicName, len);

   293

   294

        // if partial comparison matches, longest string is the greatest

   295

        if (res == 0)

   296

            // update res to continue bisection

   297

            res = delta_len;

   298

   299

        // Check if key is present at mid

   300

        if (res == 0)

   301

            goto found;

   302

   303

        // If key greater, ignore left half

   304

        if (res < 0)

   305

            low = mid + 1;

   306

   307

        // If key is smaller, ignore right half

   308

        else

   309

            high = mid - 1;

   310

    }}

   311

    // If we reach here, then the element was not present

   312

    if(__mqtt_python_callback_fptr_{name} && 

   313

       (*__mqtt_python_callback_fptr_{name})(topicName,

   314

                                     (char*)message->payload,

   315

                                     message->payloadlen) == 0){{

   316

        // Topic was handled in python

   317

        goto exit;

   318

    }} else {{

   319

        LogWarning("MQTT unknown topic: %s\n", topicName);

   320

    }}

   321

    goto exit;

   322

   323

found:

   324

   325

    is_json_type = topics[mid].is_json_type;

   326

    if(is_json_type || (__get_type_enum_size(topics[mid].vartype) == message->payloadlen)){{

   327

        if (pthread_mutex_lock(&MQTT_retrieve_mutex) == 0){{

   328

            if(is_json_type){{

   329

                (topics[mid].json_parse_func)((char*)message->payload, message->payloadlen, topics[mid].mqtt_pdata);

   330

            }} else {{

   331

                memcpy(topics[mid].mqtt_pdata, (char*)message->payload, message->payloadlen);

   332

            }}

   333

            *topics[mid].mqtt_pstate = CHANGED;

   334

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   335

        }}

   336

    }} else {{

   337

        LogWarning("MQTT wrong payload size for topic: %s. Should be %d, but got %d.", 

   338

            topicName, __get_type_enum_size(topics[mid].vartype), message->payloadlen);

   339

    }}

   340

exit:

   341

    MQTTClient_freeMessage(&message);

   342

    MQTTClient_free(topicName);

   343

    return 1;

   344

}}

   345

   346

#define INIT_NoAuth()                                                                             \

   347

    LogInfo("MQTT Init no auth\n");

   348

   349

#define INIT_x509(Verify, KeyStore, TrustStore)                                                   \

   350

    LogInfo("MQTT Init x509 with %s,%s\n", KeyStore?KeyStore:"NULL", TrustStore?TrustStore:"NULL")\

   351

    ssl_opts.verify = Verify;                                                                     \

   352

    ssl_opts.keyStore = KeyStore;                                                                 \

   353

    ssl_opts.trustStore = TrustStore;                                                             \

   354

    conn_opts.ssl = &ssl_opts;

   355

   356

#define INIT_PSK(Secret, ID)                                                                      \

   357

    LogError("MQTT PSK NOT IMPLEMENTED\n")                                                        \

   358

    /* LogInfo("MQTT Init PSK for ID %s\n", ID) */                                                \

   359

    /* ssl_opts.ssl_psk_cb = TODO; */                                                             \

   360

    /* ssl_opts.ssl_psk_context = TODO; */                                                        \

   361

    conn_opts.ssl = &ssl_opts;

   362

   363

#define INIT_UserPassword(User, Password)                                                         \

   364

    LogInfo("MQTT Init UserPassword %s,%s\n", User, Password);                                    \

   365

    conn_opts.username = User;                                                                    \

   366

    conn_opts.password = Password;

   367

   368

#ifdef USE_MQTT_5

   369

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   370

        MQTTResponse response = MQTTClient_subscribe5(client, Topic, QoS, NULL, NULL);            \

   371

        /* when using MQTT5 responce code is 1 for some reason even if no error */                \

   372

        rc = response.reasonCode == 1 ? MQTTCLIENT_SUCCESS : response.reasonCode;                 \

   373

        MQTTResponse_free(response);

   374

#else

   375

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   376

        rc = MQTTClient_subscribe(client, Topic, QoS);

   377

#endif

   378

   379

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   380

    {{                                                                                            \

   381

        int rc;                                                                                   \

   382

        _SUBSCRIBE(#Topic, QoS)                                                                   \

   383

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   384

        {{                                                                                        \

   385

            LogError("MQTT client failed to subscribe to '%s', return code %d\n", #Topic, rc);    \

   386

        }}                                                                                        \

   387

    }}

   388

   389

   390

#ifdef USE_MQTT_5

   391

#define _PUBLISH(Topic, QoS, cstring_size, cstring_ptr, Retained)                                 \

   392

        MQTTResponse response = MQTTClient_publish5(client, Topic, cstring_size,                  \

   393

            cstring_ptr, QoS, Retained, NULL, NULL);                                              \

   394

        rc = response.reasonCode;                                                                 \

   395

        MQTTResponse_free(response);

   396

#else

   397

#define _PUBLISH(Topic, QoS, cstring_size, cstring_ptr, Retained)                                 \

   398

        rc = MQTTClient_publish(client, Topic, cstring_size,                                      \

   399

            cstring_ptr, QoS, Retained, NULL);

   400

#endif

   401

   402

#define PUBLISH_SIMPLE(Topic, QoS, C_type, c_loc_name, Retained)                                  \

   403

        _PUBLISH(#Topic, QoS, sizeof(C_type), &MQTT_##c_loc_name##_buf, Retained)

   404

   405

#define PUBLISH_JSON(Topic, QoS, C_type, c_loc_name, Retained)                                    \

   406

        int res = json_gen_##c_loc_name(&MQTT_##c_loc_name##_buf);                                \

   407

        if(res == 0) {{                                                                           \

   408

            _PUBLISH(Topic, QoS, json_out_len, json_out_buf, Retained)                            \

   409

        }}

   410

   411

#define INIT_PUBLICATION(encoding, Topic, QoS, C_type, c_loc_name, Retained)                      \

   412

    {{                                                                                            \

   413

        int rc;                                                                                   \

   414

        PUBLISH_##encoding(Topic, QoS, C_type, c_loc_name, Retained)                              \

   415

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   416

        {{                                                                                        \

   417

            LogError("MQTT client failed to init publication of '%s', return code %d\n", #Topic, rc);\

   418

            /* TODO update status variable accordingly */                                         \

   419

        }}                                                                                        \

   420

    }}

   421

   422

#define PUBLISH_CHANGE(encoding, Topic, QoS, C_type, c_loc_name, Retained)                        \

   423

    if(MQTT_##c_loc_name##_state == CHANGED)                                                      \

   424

    {{                                                                                            \

   425

        int rc;                                                                                   \

   426

        PUBLISH_##encoding(Topic, QoS, C_type, c_loc_name, Retained)                              \

   427

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   428

        {{                                                                                        \

   429

            LogError("MQTT client failed to publish '%s', return code %d\n", #Topic, rc);         \

   430

            /* TODO update status variable accordingly */                                         \

   431

        }} else {{                                                                                \

   432

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   433

        }}                                                                                        \

   434

    }}

   435

   436

static int _connect_mqtt(void)

   437

{{

   438

    int rc;

   439

   440

#ifdef USE_MQTT_5

   441

    MQTTProperties props = MQTTProperties_initializer;

   442

    MQTTProperties willProps = MQTTProperties_initializer;

   443

    MQTTResponse response = MQTTResponse_initializer;

   444

   445

    response = MQTTClient_connect5(client, &conn_opts, &props, &willProps);

   446

    rc = response.reasonCode;

   447

    MQTTResponse_free(response);

   448

#else

   449

    rc = MQTTClient_connect(client, &conn_opts);

   450

#endif

   451

   452

    if (rc != MQTTCLIENT_SUCCESS) {{

   453

        MQTT_is_disconnected = 1;

   454

        return rc;

   455

    }}else{{

   456

        MQTT_is_disconnected = 0;

   457

    }}

   458

   459

{init_pubsub}

   460

   461

    return MQTTCLIENT_SUCCESS;

   462

}}

   463

   464

static void *__MQTT_thread_proc(void *_unused) {{

   465

    int rc = 0;

   466

   467

    while((rc = pthread_mutex_lock(&MQTT_thread_wakeup_mutex)) == 0 && !MQTT_stop_thread){{

   468

        int do_publish; 

   469

   470

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   471

   472

        if(MQTT_is_disconnected)

   473

        {{

   474

            /* TODO growing retry delay */

   475

            /* TODO max retry delay as config parameter */

   476

            sleep(5);

   477

            rc = _connect_mqtt();

   478

            if (rc == MQTTCLIENT_SUCCESS) {{

   479

                LogInfo("MQTT Reconnected\n");

   480

            }} else {{

   481

                LogError("MQTT Reconnect Failed, return code %d\n", rc);

   482

            }}

   483

        }} 

   484

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   485

        {{

   486

            /* publish changes, and reset variable's state to UNCHANGED */

   487

{publish_changes}