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

    74

    75

#define scanf_fmt_BOOL   "%B"

    76

#define scanf_fmt_SINT   "%hhd"

    77

#define scanf_fmt_USINT  "%uhhd"

    78

#define scanf_fmt_INT    "%hd" 

    79

#define scanf_fmt_UINT   "%uhd"

    80

#define scanf_fmt_DINT   "%d" 

    81

#define scanf_fmt_UDINT  "%ud"

    82

#define scanf_fmt_LINT   "%ld"

    83

#define scanf_fmt_ULINT  "%uld"

    84

#define scanf_fmt_REAL   "%f"

    85

#define scanf_fmt_LREAL  "%Lf"

    86

#define scanf_fmt_STRING "%M"

    87

    88

#define scanf_fmt_separator ", "

    89

    92

    93

#define   scanf_arg_BOOL(name, data_ptr) &data_ptr->name

    94

#define   scanf_arg_SINT(name, data_ptr) &data_ptr->name

    95

#define  scanf_arg_USINT(name, data_ptr) &data_ptr->name

    96

#define    scanf_arg_INT(name, data_ptr) &data_ptr->name

    97

#define   scanf_arg_UINT(name, data_ptr) &data_ptr->name

    98

#define   scanf_arg_DINT(name, data_ptr) &data_ptr->name

    99

#define  scanf_arg_UDINT(name, data_ptr) &data_ptr->name

   100

#define   scanf_arg_LINT(name, data_ptr) &data_ptr->name

   101

#define  scanf_arg_ULINT(name, data_ptr) &data_ptr->name

   102

#define   scanf_arg_REAL(name, data_ptr) &data_ptr->name

   103

#define  scanf_arg_LREAL(name, data_ptr) &data_ptr->name

   104

#define scanf_arg_STRING(name, data_ptr) scan_string, &data_ptr->name 

   105

   106

#define scanf_args_separator ,

   107

   110

   111

#define   printf_arg_BOOL(name, data_ptr) data_ptr->name

   112

#define   printf_arg_SINT(name, data_ptr) data_ptr->name

   113

#define  printf_arg_USINT(name, data_ptr) data_ptr->name

   114

#define    printf_arg_INT(name, data_ptr) data_ptr->name

   115

#define   printf_arg_UINT(name, data_ptr) data_ptr->name

   116

#define   printf_arg_DINT(name, data_ptr) data_ptr->name

   117

#define  printf_arg_UDINT(name, data_ptr) data_ptr->name

   118

#define   printf_arg_LINT(name, data_ptr) data_ptr->name

   119

#define  printf_arg_ULINT(name, data_ptr) data_ptr->name

   120

#define   printf_arg_REAL(name, data_ptr) data_ptr->name

   121

#define  printf_arg_LREAL(name, data_ptr) data_ptr->name

   122

#define printf_arg_STRING(name, data_ptr) data_ptr->name.len, data_ptr->name.body 

   123

   124

#define printf_args_separator ,

   125

   128

   129

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

   130

	IEC_STRING *iecstr = (IEC_STRING*)user_data;

   131

	__strlen_t ieclen = len > STR_MAX_LEN ? STR_MAX_LEN : len;

   132

    printf("%.*s", len, str);

   133

    printf("%.*s", ieclen, str);

   134

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

   135

    iecstr->len = ieclen;

   136

}}

   137

   138

#define DECL_JSON_INPUT(C_type, c_loc_name) \

   139

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

   140

    C_type *struct_ptr = (C_type *)void_ptr; \

   141

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

   142

}}

   143

   144

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

   145

#define json_out_size 1<<12 // 4K

   146

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

   147

static int json_out_len = 0;

   148

   149

#define DECL_JSON_OUTPUT(C_type, c_loc_name) \

   150

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

   151

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

   152

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

   153

    if(json_out_len > json_out_size){{ \

   154

        json_out_len = 0; \

   155

        return -EOVERFLOW; \

   156

    }} \

   157

    return 0; \

   158

}}

   159

   160

{json_decl}

   161

   162

static MQTTClient client;

   163

#ifdef USE_MQTT_5

   164

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;

   165

#else

   166

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

   167

#endif

   168

   169

MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;

   170

   171

/* condition to quit publish thread */

   172

static int MQTT_stop_thread = 0;

   173

   174

/* condition to wakeup publish thread */

   175

static int MQTT_any_pub_var_changed = 0;

   176

   177

/* Keep track of connection state */

   178

static volatile int MQTT_is_disconnected = 1;

   179

   180

/* mutex to keep incoming PLC data consistent */

   181

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

   182

   183

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

   184

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

   185

   186

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

   187

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

   188

   189

/* thread that handles publish and reconnection */

   190

static pthread_t MQTT_thread;

   191

   192

#define INIT_TOPIC(topic, iec_type, c_loc_name) \

   193

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

   194

   195

#define INIT_JSON_TOPIC(topic, iec_type, c_loc_name) \

   196

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

   197

   198

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

   199

   200

static struct {{

   201

    const char *topic; //null terminated topic string

   202

    const unsigned int topic_len;

   203

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

   204

    int *mqtt_pstate; // pointer to changed flag

   205

    int is_json_type;

   206

    union {{

   207

       __IEC_types_enum vartype;

   208

       json_parse_func_t json_parse_func;

   209

    }};

   210

}} topics [] = {{

   211

{topics}

   212

}};

   213

   214

void __cleanup_{locstr}(void)

   215

{{

   216

    int rc;

   217

   218

    /* stop publish thread */

   219

    MQTT_stop_thread = 1;

   220

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

   221

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

   222

        pthread_cond_signal(&MQTT_thread_wakeup);

   223

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   224

    }}

   225

    pthread_join(MQTT_thread, NULL);

   226

   227

#ifdef USE_MQTT_5

   228

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

   229

#else

   230

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

   231

#endif

   232

    {{

   233

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

   234

    }}

   235

    MQTTClient_destroy(&client);

   236

}}

   237

   238

void connectionLost(void* context, char* reason)

   239

{{

   240

    int rc;

   241

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

   242

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

   243

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

   244

        MQTT_is_disconnected = 1;

   245

        pthread_cond_signal(&MQTT_thread_wakeup);

   246

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   247

    }}

   248

}}

   249

   250

   251

   252

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

   253

{{

   254

    int low = 0;

   255

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

   256

    int high = size - 1;

   257

    int mid;

   258

    int is_json_type;

   259

   260

    // bisect topic among subscribed topics

   261

    while (low <= high) {{

   262

        int res, len, delta_len;

   263

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

   264

        if(topicLen == 0) {{

   265

            // zero topic len means null-terminated

   266

            topicLen = strlen(topicName);

   267

        }}

   268

   269

        len = topics[mid].topic_len;

   270

   271

        // keep track of length difference

   272

        delta_len = len - topicLen;

   273

   274

        // len = min(len, topicLen)

   275

        len = topicLen > len ? len : topicLen;

   276

   277

        // compare strings as far as possible

   278

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

   279

   280

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

   281

        if (res == 0)

   282

            // update res to continue bisection

   283

            res = delta_len;

   284

   285

        // Check if key is present at mid

   286

        if (res == 0)

   287

            goto found;

   288

   289

        // If key greater, ignore left half

   290

        if (res < 0)

   291

            low = mid + 1;

   292

   293

        // If key is smaller, ignore right half

   294

        else

   295

            high = mid - 1;

   296

    }}

   297

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

   298

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

   299

    goto exit;

   300

   301

found:

   302

   303

    is_json_type = topics[mid].is_json_type;

   304

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

   305

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

   306

            if(is_json_type){{

   307

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

   308

            }} else {{

   309

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

   310

            }}

   311

            *topics[mid].mqtt_pstate = CHANGED;

   312

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   313

        }}

   314

    }} else {{

   315

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

   316

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

   317

    }}

   318

exit:

   319

    MQTTClient_freeMessage(&message);

   320

    MQTTClient_free(topicName);

   321

    return 1;

   322

}}

   323

   324

#define INIT_NoAuth()                                                                             \

   325

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

   326

   327

#define INIT_x509(Verify, KeyStore, TrustStore)                                                   \

   328

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

   329

    ssl_opts.verify = Verify;                                                                     \

   330

    ssl_opts.keyStore = KeyStore;                                                                 \

   331

    ssl_opts.trustStore = TrustStore;                                                             \

   332

    conn_opts.ssl = &ssl_opts;

   333

   334

#define INIT_PSK(Secret, ID)                                                                      \

   335

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

   336

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

   337

    /* ssl_opts.ssl_psk_cb = TODO; */                                                             \

   338

    /* ssl_opts.ssl_psk_context = TODO; */                                                        \

   339

    conn_opts.ssl = &ssl_opts;

   340

   341

#define INIT_UserPassword(User, Password)                                                         \

   342

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

   343

    conn_opts.username = User;                                                                    \

   344

    conn_opts.password = Password;

   345

   346

#ifdef USE_MQTT_5

   347

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   348

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

   349

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

   350

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

   351

        MQTTResponse_free(response);

   352

#else

   353

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   354

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

   355

#endif

   356

   357

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   358

    {{                                                                                            \

   359

        int rc;                                                                                   \

   360

        _SUBSCRIBE(Topic, QoS)                                                                    \

   361

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   362

        {{                                                                                        \

   363

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

   364

        }}                                                                                        \

   365

    }}

   366

   367

   368

#ifdef USE_MQTT_5

   369

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

   370

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

   371

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

   372

        rc = response.reasonCode;                                                                 \

   373

        MQTTResponse_free(response);

   374

#else

   375

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

   376

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

   377

            cstring_ptr, QoS, Retained, NULL);

   378

#endif

   379

   380

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

   381

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

   382

   383

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

   384

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

   385

        if(res == 0) {{                                                                           \

   386

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

   387

        }}

   388

   389

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

   390

    {{                                                                                            \

   391

        int rc;                                                                                   \

   392

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

   393

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   394

        {{                                                                                        \

   395

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

   396

            /* TODO update status variable accordingly */                                         \

   397

        }}                                                                                        \

   398

    }}

   399

   400

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

   401

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

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

   408

            /* TODO update status variable accordingly */                                         \

   409

        }} else {{                                                                                \

   410

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   411

        }}                                                                                        \

   412

    }}

   413

   414

static int _connect_mqtt(void)

   415

{{

   416

    int rc;

   417

   418

#ifdef USE_MQTT_5

   419

    MQTTProperties props = MQTTProperties_initializer;

   420

    MQTTProperties willProps = MQTTProperties_initializer;

   421

    MQTTResponse response = MQTTResponse_initializer;

   422

   423

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

   424

    rc = response.reasonCode;

   425

    MQTTResponse_free(response);

   426

#else

   427

    rc = MQTTClient_connect(client, &conn_opts);

   428

#endif

   429

   430

    if (rc != MQTTCLIENT_SUCCESS) {{

   431

        MQTT_is_disconnected = 1;

   432

        return rc;

   433

    }}else{{

   434

        MQTT_is_disconnected = 0;

   435

    }}

   436

   437

{init_pubsub}

   438

   439

    return MQTTCLIENT_SUCCESS;

   440

}}

   441

   442

static void *__MQTT_thread_proc(void *_unused) {{

   443

    int rc = 0;

   444

   445

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

   446

        int do_publish; 

   447

   448

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   449

   450

        if(MQTT_is_disconnected)

   451

        {{

   452

            /* TODO growing retry delay */

   453

            /* TODO max retry delay as config parameter */

   454

            sleep(5);

   455

            rc = _connect_mqtt();

   456

            if (rc == MQTTCLIENT_SUCCESS) {{

   457

                LogInfo("MQTT Reconnected\n");

   458

            }} else {{

   459

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

   460

            }}

   461

        }} 

   462

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   463

        {{

   464

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

   465

{publish_changes}

   466

            MQTT_any_pub_var_changed = 0;

   467

        }}

   468

   469

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   470

   471

        if(MQTT_stop_thread) break;

   472

    }}

   473

   474

    if(!MQTT_stop_thread){{

   475

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

   476

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

   477

    }}

   478

}}

   479

   480

int __init_{locstr}(int argc,char **argv)

   481

{{

   482

    char *uri = "{uri}";

   483

    char *clientID = "{clientID}";

   484

    int rc;

   485

   486

    MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;

   487

   488