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

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

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

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

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

   264

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

   265

{{

   266

    int low = 0;

   267

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

   268

    int high = size - 1;

   269

    int mid;

   270

    int is_json_type;

   271

   272

    // bisect topic among subscribed topics

   273

    while (low <= high) {{

   274

        int res, len, delta_len;

   275

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

   276

        if(topicLen == 0) {{

   277

            // zero topic len means null-terminated

   278

            topicLen = strlen(topicName);

   279

        }}

   280

   281

        len = topics[mid].topic_len;

   282

   283

        // keep track of length difference

   284

        delta_len = len - topicLen;

   285

   286

        // len = min(len, topicLen)

   287

        len = topicLen > len ? len : topicLen;

   288

   289

        // compare strings as far as possible

   290

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

   291

   292

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

   293

        if (res == 0)

   294

            // update res to continue bisection

   295

            res = delta_len;

   296

   297

        // Check if key is present at mid

   298

        if (res == 0)

   299

            goto found;

   300

   301

        // If key greater, ignore left half

   302

        if (res < 0)

   303

            low = mid + 1;

   304

   305

        // If key is smaller, ignore right half

   306

        else

   307

            high = mid - 1;

   308

    }}

   309

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

   310

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

   311

    goto exit;

   312

   313

found:

   314

   315

    is_json_type = topics[mid].is_json_type;

   316

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

   317

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

   318

            if(is_json_type){{

   319

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

   320

            }} else {{

   321

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

   322

            }}

   323

            *topics[mid].mqtt_pstate = CHANGED;

   324

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   325

        }}

   326

    }} else {{

   327

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

   328

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

   329

    }}

   330

exit:

   331

    MQTTClient_freeMessage(&message);

   332

    MQTTClient_free(topicName);

   333

    return 1;

   334

}}

   335

   336

#define INIT_NoAuth()                                                                             \

   337

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

   338

   339

#define INIT_x509(Verify, KeyStore, TrustStore)                                                   \

   340

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

   341

    ssl_opts.verify = Verify;                                                                     \

   342

    ssl_opts.keyStore = KeyStore;                                                                 \

   343

    ssl_opts.trustStore = TrustStore;                                                             \

   344

    conn_opts.ssl = &ssl_opts;

   345

   346

#define INIT_PSK(Secret, ID)                                                                      \

   347

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

   348

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

   349

    /* ssl_opts.ssl_psk_cb = TODO; */                                                             \

   350

    /* ssl_opts.ssl_psk_context = TODO; */                                                        \

   351

    conn_opts.ssl = &ssl_opts;

   352

   353

#define INIT_UserPassword(User, Password)                                                         \

   354

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

   355

    conn_opts.username = User;                                                                    \

   356

    conn_opts.password = Password;

   357

   358

#ifdef USE_MQTT_5

   359

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   360

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

   361

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

   362

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

   363

        MQTTResponse_free(response);

   364

#else

   365

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   366

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

   367

#endif

   368

   369

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   370

    {{                                                                                            \

   371

        int rc;                                                                                   \

   372

        _SUBSCRIBE(Topic, QoS)                                                                    \

   373

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   374

        {{                                                                                        \

   375

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

   376

        }}                                                                                        \

   377

    }}

   378

   379

   380

#ifdef USE_MQTT_5

   381

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

   382

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

   383

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

   384

        rc = response.reasonCode;                                                                 \

   385

        MQTTResponse_free(response);

   386

#else

   387

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

   388

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

   389

            cstring_ptr, QoS, Retained, NULL);

   390

#endif

   391

   392

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

   393

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

   394

   395

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

   396

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

   397

        if(res == 0) {{                                                                           \

   398

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

   399

        }}

   400

   401

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

   402

    {{                                                                                            \

   403

        int rc;                                                                                   \

   404

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

   405

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   406

        {{                                                                                        \

   407

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

   408

            /* TODO update status variable accordingly */                                         \

   409

        }}                                                                                        \

   410

    }}

   411

   412

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

   413

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

   414

    {{                                                                                            \

   415

        int rc;                                                                                   \

   416

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

   417

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   418

        {{                                                                                        \

   419

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

   420

            /* TODO update status variable accordingly */                                         \

   421

        }} else {{                                                                                \

   422

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   423

        }}                                                                                        \

   424

    }}

   425

   426

static int _connect_mqtt(void)

   427

{{

   428

    int rc;

   429

   430

#ifdef USE_MQTT_5

   431

    MQTTProperties props = MQTTProperties_initializer;

   432

    MQTTProperties willProps = MQTTProperties_initializer;

   433

    MQTTResponse response = MQTTResponse_initializer;

   434

   435

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

   436

    rc = response.reasonCode;

   437

    MQTTResponse_free(response);

   438

#else

   439

    rc = MQTTClient_connect(client, &conn_opts);

   440

#endif

   441

   442

    if (rc != MQTTCLIENT_SUCCESS) {{

   443

        MQTT_is_disconnected = 1;

   444

        return rc;

   445

    }}else{{

   446

        MQTT_is_disconnected = 0;

   447

    }}

   448

   449

{init_pubsub}

   450

   451

    return MQTTCLIENT_SUCCESS;

   452

}}

   453

   454

static void *__MQTT_thread_proc(void *_unused) {{

   455

    int rc = 0;

   456

   457

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

   458

        int do_publish; 

   459

   460

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   461

   462

        if(MQTT_is_disconnected)

   463

        {{

   464

            /* TODO growing retry delay */

   465

            /* TODO max retry delay as config parameter */

   466

            sleep(5);

   467

            rc = _connect_mqtt();

   468

            if (rc == MQTTCLIENT_SUCCESS) {{

   469

                LogInfo("MQTT Reconnected\n");

   470

            }} else {{

   471

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

   472

            }}

   473

        }} 

   474

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   475

        {{

   476

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

   477

{publish_changes}

   478

            MQTT_any_pub_var_changed = 0;

   479

        }}

   480

   481

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   482

   483

        if(MQTT_stop_thread) break;

   484

    }}

   485

   486

    if(!MQTT_stop_thread){{

   487

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

   488