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

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

#define DECL_VAR(iec_type, C_type, c_loc_name)                                                     \

    48

static C_type PLC_##c_loc_name##_buf;                                                              \

    49

static C_type MQTT_##c_loc_name##_buf;                                                             \

    50

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

    51

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

    52

    53

{decl}

    54

    55

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

    56

    57

#define printf_fmt_BOOL   "%B"

    58

#define printf_fmt_SINT   "%hhd"

    59

#define printf_fmt_USINT  "%uhhd"

    60

#define printf_fmt_INT    "%hd" 

    61

#define printf_fmt_UINT   "%uhd"

    62

#define printf_fmt_DINT   "%d" 

    63

#define printf_fmt_UDINT  "%ud"

    64

#define printf_fmt_LINT   "%ld"

    65

#define printf_fmt_ULINT  "%uld"

    66

#define printf_fmt_REAL   "%f"

    67

#define printf_fmt_LREAL  "%Lf"

    68

#define printf_fmt_STRING "%.*Q"

    69

    70

#define printf_fmt_separator ", "

    71

    72

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

    73

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

    74

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

    75

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

    76

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

    77

    78

#define scanf_fmt_BOOL   "%B"

    79

#define scanf_fmt_SINT   "%hhd"

    80

#define scanf_fmt_USINT  "%uhhd"

    81

#define scanf_fmt_INT    "%hd" 

    82

#define scanf_fmt_UINT   "%uhd"

    83

#define scanf_fmt_DINT   "%d" 

    84

#define scanf_fmt_UDINT  "%ud"

    85

#define scanf_fmt_LINT   "%ld"

    86

#define scanf_fmt_ULINT  "%uld"

    87

#define scanf_fmt_REAL   "%f"

    88

#define scanf_fmt_LREAL  "%Lf"

    89

#define scanf_fmt_STRING "%M"

    90

    91

#define scanf_fmt_separator ", "

    92

    93

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

    94

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

    95

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

    96

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

    97

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

    98

    99

#define   scanf_arg_BOOL(arg) arg

   100

#define   scanf_arg_SINT(arg) arg

   101

#define  scanf_arg_USINT(arg) arg

   102

#define    scanf_arg_INT(arg) arg

   103

#define   scanf_arg_UINT(arg) arg

   104

#define   scanf_arg_DINT(arg) arg

   105

#define  scanf_arg_UDINT(arg) arg

   106

#define   scanf_arg_LINT(arg) arg

   107

#define  scanf_arg_ULINT(arg) arg

   108

#define   scanf_arg_REAL(arg) arg

   109

#define  scanf_arg_LREAL(arg) arg

   110

#define scanf_arg_STRING(arg) scan_string, arg

   111

   112

#define scanf_args_separator ,

   113

   114

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

   115

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

   116

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

   117

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

   118

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

   119

   120

#define   printf_arg_BOOL(arg) arg

   121

#define   printf_arg_SINT(arg) arg

   122

#define  printf_arg_USINT(arg) arg

   123

#define    printf_arg_INT(arg) arg

   124

#define   printf_arg_UINT(arg) arg

   125

#define   printf_arg_DINT(arg) arg

   126

#define  printf_arg_UDINT(arg) arg

   127

#define   printf_arg_LINT(arg) arg

   128

#define  printf_arg_ULINT(arg) arg

   129

#define   printf_arg_REAL(arg) arg

   130

#define  printf_arg_LREAL(arg) arg

   131

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

   132

   133

#define printf_args_separator ,

   134

   135

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

   136

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

   137

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

   138

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

   139

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

   140

   141

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

   142

	IEC_STRING *iecstr = (IEC_STRING*)user_data;

   143

	__strlen_t ieclen = len > STR_MAX_LEN ? STR_MAX_LEN : len;

   144

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

   145

    iecstr->len = ieclen;

   146

}}

   147

   148

#define DECL_JSON_INPUT(C_type, c_loc_name) \

   149

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

   150

    C_type *struct_ptr = (C_type *)void_ptr; \

   151

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

   152

}}

   153

   154

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

   155

#define json_out_size 1<<12 // 4K

   156

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

   157

static int json_out_len = 0;

   158

   159

#define DECL_JSON_OUTPUT(C_type, c_loc_name) \

   160

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

   161

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

   162

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

   163

    if(json_out_len > json_out_size){{ \

   164

        json_out_len = 0; \

   165

        return -EOVERFLOW; \

   166

    }} \

   167

    return 0; \

   168

}}

   169

   170

{json_decl}

   171

   172

static MQTTClient client;

   173

#ifdef USE_MQTT_5

   174

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;

   175

#else

   176

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

   177

#endif

   178

   179

MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;

   180

   181

/* condition to quit publish thread */

   182

static int MQTT_stop_thread = 0;

   183

   184

/* condition to wakeup publish thread */

   185

static int MQTT_any_pub_var_changed = 0;

   186

   187

/* Keep track of connection state */

   188

static volatile int MQTT_is_disconnected = 1;

   189

   190

/* mutex to keep incoming PLC data consistent */

   191

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

   192

   193

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

   194

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

   195

   196

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

   197

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

   198

   199

/* thread that handles publish and reconnection */

   200

static pthread_t MQTT_thread;

   201

   202

#define INIT_TOPIC(topic, iec_type, c_loc_name) \

   203

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

   204

   205

#define INIT_JSON_TOPIC(topic, iec_type, c_loc_name) \

   206

{{#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}},

   207

   208

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

   209

   210

static struct {{

   211

    const char *topic; //null terminated topic string

   212

    const unsigned int topic_len;

   213

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

   214

    int *mqtt_pstate; // pointer to changed flag

   215

    int is_json_type;

   216

    union {{

   217

       __IEC_types_enum vartype;

   218

       json_parse_func_t json_parse_func;

   219

    }};

   220

}} topics [] = {{

   221

{topics}

   222

}};

   223

   224

void __cleanup_{locstr}(void)

   225

{{

   226

    int rc;

   227

   228

    /* stop publish thread */

   229

    MQTT_stop_thread = 1;

   230

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

   231

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

   232

        pthread_cond_signal(&MQTT_thread_wakeup);

   233

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   234

    }}

   235

    pthread_join(MQTT_thread, NULL);

   236

   237

#ifdef USE_MQTT_5

   238

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

   239

#else

   240

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

   241

#endif

   242

    {{

   243

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

   244

    }}

   245

    MQTTClient_destroy(&client);

   246

}}

   247

   248

void connectionLost(void* context, char* reason)

   249

{{

   250

    int rc;

   251

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

   252

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

   253

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

   254

        MQTT_is_disconnected = 1;

   255

        pthread_cond_signal(&MQTT_thread_wakeup);

   256

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   257

    }}

   258

}}

   259

   260

   261

   262

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

   263

{{

   264

    int low = 0;

   265

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

   266

    int high = size - 1;

   267

    int mid;

   268

    int is_json_type;

   269

   270

    // bisect topic among subscribed topics

   271

    while (low <= high) {{

   272

        int res, len, delta_len;

   273

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

   274

        if(topicLen == 0) {{

   275

            // zero topic len means null-terminated

   276

            topicLen = strlen(topicName);

   277

        }}

   278

   279

        len = topics[mid].topic_len;

   280

   281

        // keep track of length difference

   282

        delta_len = len - topicLen;

   283

   284

        // len = min(len, topicLen)

   285

        len = topicLen > len ? len : topicLen;

   286

   287

        // compare strings as far as possible

   288

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

   289

   290

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

   291

        if (res == 0)

   292

            // update res to continue bisection

   293

            res = delta_len;

   294

   295

        // Check if key is present at mid

   296

        if (res == 0)

   297

            goto found;

   298

   299

        // If key greater, ignore left half

   300

        if (res < 0)

   301

            low = mid + 1;

   302

   303

        // If key is smaller, ignore right half

   304

        else

   305

            high = mid - 1;

   306

    }}

   307

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

   308

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

   309

    goto exit;

   310

   311

found:

   312

   313

    is_json_type = topics[mid].is_json_type;

   314

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

   315

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

   316

            if(is_json_type){{

   317

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

   318

            }} else {{

   319

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

   320

            }}

   321

            *topics[mid].mqtt_pstate = CHANGED;

   322

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   323

        }}

   324

    }} else {{

   325

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

   326

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

   327

    }}

   328

exit:

   329

    MQTTClient_freeMessage(&message);

   330

    MQTTClient_free(topicName);

   331

    return 1;

   332

}}

   333

   334

#define INIT_NoAuth()                                                                             \

   335

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

   336

   337

#define INIT_x509(Verify, KeyStore, TrustStore)                                                   \

   338

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

   339

    ssl_opts.verify = Verify;                                                                     \

   340

    ssl_opts.keyStore = KeyStore;                                                                 \

   341

    ssl_opts.trustStore = TrustStore;                                                             \

   342

    conn_opts.ssl = &ssl_opts;

   343

   344

#define INIT_PSK(Secret, ID)                                                                      \

   345

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

   346

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

   347

    /* ssl_opts.ssl_psk_cb = TODO; */                                                             \

   348

    /* ssl_opts.ssl_psk_context = TODO; */                                                        \

   349

    conn_opts.ssl = &ssl_opts;

   350

   351

#define INIT_UserPassword(User, Password)                                                         \

   352

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

   353

    conn_opts.username = User;                                                                    \

   354

    conn_opts.password = Password;

   355

   356

#ifdef USE_MQTT_5

   357

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   358

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

   359

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

   360

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

   361

        MQTTResponse_free(response);

   362

#else

   363

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   364

        rc = MQTTClient_subscribe(client, #Topic, QoS);

   365

#endif

   366

   367

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   368

    {{                                                                                            \

   369

        int rc;                                                                                   \

   370

        _SUBSCRIBE(Topic, QoS)                                                                    \

   371

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   372

        {{                                                                                        \

   373

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

   374

        }}                                                                                        \

   375

    }}

   376

   377

   378

#ifdef USE_MQTT_5

   379

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

   380

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

   381

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

   382

        rc = response.reasonCode;                                                                 \

   383

        MQTTResponse_free(response);

   384

#else

   385

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

   386

        rc = MQTTClient_publish(client, #Topic, cstring_size,                                     \

   387

            cstring_ptr, QoS, Retained, NULL);

   388

#endif

   389

   390

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

   391

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

   392

   393

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

   394

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

   395

        if(res == 0) {{                                                                           \

   396

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

   397

        }}

   398

   399

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

   400

    {{                                                                                            \

   401

        int rc;                                                                                   \

   402

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

   403

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   404

        {{                                                                                        \

   405

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

   406

            /* TODO update status variable accordingly */                                         \

   407

        }}                                                                                        \

   408

    }}

   409

   410

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

   411

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

   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 publish '%s', return code %d\n", #Topic, rc);         \

   418

            /* TODO update status variable accordingly */                                         \

   419

        }} else {{                                                                                \

   420

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   421

        }}                                                                                        \

   422

    }}

   423

   424

static int _connect_mqtt(void)

   425

{{

   426

    int rc;

   427

   428

#ifdef USE_MQTT_5

   429

    MQTTProperties props = MQTTProperties_initializer;

   430

    MQTTProperties willProps = MQTTProperties_initializer;

   431

    MQTTResponse response = MQTTResponse_initializer;

   432

   433

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

   434

    rc = response.reasonCode;

   435

    MQTTResponse_free(response);

   436

#else

   437

    rc = MQTTClient_connect(client, &conn_opts);

   438

#endif

   439

   440

    if (rc != MQTTCLIENT_SUCCESS) {{

   441

        MQTT_is_disconnected = 1;

   442

        return rc;

   443

    }}else{{

   444

        MQTT_is_disconnected = 0;

   445

    }}

   446

   447

{init_pubsub}

   448

   449

    return MQTTCLIENT_SUCCESS;

   450

}}

   451

   452

static void *__MQTT_thread_proc(void *_unused) {{

   453

    int rc = 0;

   454

   455

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

   456

        int do_publish; 

   457

   458

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   459

   460

        if(MQTT_is_disconnected)

   461

        {{

   462

            /* TODO growing retry delay */

   463

            /* TODO max retry delay as config parameter */

   464

            sleep(5);

   465

            rc = _connect_mqtt();

   466

            if (rc == MQTTCLIENT_SUCCESS) {{

   467

                LogInfo("MQTT Reconnected\n");

   468

            }} else {{

   469

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

   470

            }}

   471

        }} 

   472

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   473

        {{

   474

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

   475

{publish_changes}

   476

            MQTT_any_pub_var_changed = 0;

   477

        }}

   478

   479

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   480

   481

        if(MQTT_stop_thread) break;

   482

    }}

   483

   484

    if(!MQTT_stop_thread){{

   485

        /* if thread exits outside of normal shutdown, report error*/

   486

        LogError("MQTT client thread exited unexpectedly, return code %d\n", rc);

   487

    }}