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

     9

#include "MQTTClient.h"

    10

#include "MQTTClientPersistence.h"

    11

    12

#define _Log(level, ...)                                                                          \

    13

    {{                                                                                            \

    14

        char mstr[256];                                                                           \

    15

        snprintf(mstr, 255, __VA_ARGS__);                                                         \

    16

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

    17

        printf(__VA_ARGS__);                                                                      \

    18

        fflush(stdout);                                                                           \

    19

    }}

    20

    21

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

    22

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

    23

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

    24

    25

// Selected debug level for paho stack

    26

// can be:

    27

// MQTTCLIENT_TRACE_PROTOCOL, MQTTCLIENT_TRACE_MAXIMUM, MQTTCLIENT_TRACE_ERROR

    28

#define MQTT_DEBUG_LEVEL MQTTCLIENT_TRACE_ERROR

    29

    30

void trace_callback(enum MQTTCLIENT_TRACE_LEVELS level, char* message)

    31

{{

    32

    if(level >= MQTT_DEBUG_LEVEL)

    33

    {{

    34

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

    35

                                (level > MQTTCLIENT_TRACE_MINIMUM) ? LOG_WARNING : 

    36

                                LOG_INFO;

    37

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

    38

    }}

    39

}}

    40

    41

#define CHANGED 1

    42

#define UNCHANGED 0

    43

    44

#define DECL_VAR(iec_type, C_type, c_loc_name)                                                     \

    45

static C_type PLC_##c_loc_name##_buf = 0;                                                          \

    46

static C_type MQTT_##c_loc_name##_buf = 0;                                                         \

    47

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

    48

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

    49

    50

{decl}

    51

    52

static MQTTClient client;

    53

#ifdef USE_MQTT_5

    54

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;

    55

#else

    56

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

    57

#endif

    58

    59

MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;

    60

    61

/* condition to quit publish thread */

    62

static int MQTT_stop_thread = 0;

    63

    64

/* condition to wakeup publish thread */

    65

static int MQTT_any_pub_var_changed = 0;

    66

    67

/* Keep track of connection state */

    68

static volatile int MQTT_is_disconnected = 1;

    69

    70

/* mutex to keep incoming PLC data consistent */

    71

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

    72

    73

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

    74

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

    75

    76

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

    77

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

    78

    79

/* thread that handles publish and reconnection */

    80

static pthread_t MQTT_thread;

    81

    82

#define INIT_TOPIC(topic, iec_type, c_loc_name)                                                    \

    83

{{#topic, &MQTT_##c_loc_name##_buf, &MQTT_##c_loc_name##_state, iec_type##_ENUM}},

    84

    85

static struct {{

    86

    const char *topic; //null terminated topic string

    87

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

    88

    int *mqtt_pchanged; // pointer to changed flag

    89

    __IEC_types_enum vartype;

    90

}} topics [] = {{

    91

{topics}

    92

}};

    93

    94

void __cleanup_{locstr}(void)

    95

{{

    96

    int rc;

    97

    98

    /* stop publish thread */

    99

    MQTT_stop_thread = 1;

   100

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

   101

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

   102

        pthread_cond_signal(&MQTT_thread_wakeup);

   103

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   104

    }}

   105

    pthread_join(MQTT_thread, NULL);

   106

   107

#ifdef USE_MQTT_5

   108

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

   109

#else

   110

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

   111

#endif

   112

    {{

   113

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

   114

    }}

   115

    MQTTClient_destroy(&client);

   116

}}

   117

   118

void connectionLost(void* context, char* reason)

   119

{{

   120

    int rc;

   121

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

   122

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

   123

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

   124

        MQTT_is_disconnected = 1;

   125

        pthread_cond_signal(&MQTT_thread_wakeup);

   126

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   127

    }}

   128

}}

   129

   130

   131

   132

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

   133

{{

   134

    int low = 0;

   135

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

   136

    int high = size - 1;

   137

    int mid;

   138

   139

    // bisect topic among subscribed topics

   140

    while (low <= high) {{

   141

        int res;

   142

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

   143

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

   144

   145

        // Check if key is present at mid

   146

        if (res == 0)

   147

            goto found;

   148

   149

        // If key greater, ignore left half

   150

        if (res < 0)

   151

            low = mid + 1;

   152

   153

        // If key is smaller, ignore right half

   154

        else

   155

            high = mid - 1;

   156

    }}

   157

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

   158

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

   159

    goto exit;

   160

   161

found:

   162

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

   163

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

   164

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

   165

            *topics[mid].mqtt_pchanged = 1;

   166

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   167

        }}

   168

    }} else {{

   169

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

   170

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

   171

    }}

   172

exit:

   173

    MQTTClient_freeMessage(&message);

   174

    MQTTClient_free(topicName);

   175

    return 1;

   176

}}

   177

   178

#define INIT_NoAuth()                                                                             \

   179

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

   180

   181

#define INIT_x509(Verify, KeyStore, TrustStore)                                                   \

   182

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

   183

    ssl_opts.verify = Verify;                                                                     \

   184

    ssl_opts.keyStore = KeyStore;                                                                 \

   185

    ssl_opts.trustStore = TrustStore;                                                             \

   186

    conn_opts.ssl = &ssl_opts;

   187

   188

#define INIT_PSK(Secret, ID)                                                                      \

   189

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

   190

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

   191

    /* ssl_opts.ssl_psk_cb = TODO; */                                                             \

   192

    /* ssl_opts.ssl_psk_context = TODO; */                                                        \

   193

    conn_opts.ssl = &ssl_opts;

   194

   195

#define INIT_UserPassword(User, Password)                                                         \

   196

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

   197

    conn_opts.username = User;                                                                    \

   198

    conn_opts.password = Password;

   199

   200

#ifdef USE_MQTT_5

   201

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   202

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

   203

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

   204

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

   205

        MQTTResponse_free(response);

   206

#else

   207

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   208

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

   209

#endif

   210

   211

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   212

    {{                                                                                            \

   213

        int rc;                                                                                   \

   214

        _SUBSCRIBE(Topic, QoS)                                                                    \

   215

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   216

        {{                                                                                        \

   217

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

   218

        }}                                                                                        \

   219

    }}

   220

   221

   222

#ifdef USE_MQTT_5

   223

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

   224

        MQTTResponse response = MQTTClient_publish5(client, #Topic, sizeof(C_type),               \

   225

            &MQTT_##c_loc_name##_buf, QoS, Retained, NULL, NULL);                                 \

   226

        rc = response.reasonCode;                                                                 \

   227

        MQTTResponse_free(response);

   228

#else

   229

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

   230

        rc = MQTTClient_publish(client, #Topic, sizeof(C_type),                                   \

   231

            &PLC_##c_loc_name##_buf, QoS, Retained, NULL);

   232

#endif

   233

   234

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

   235

    {{                                                                                            \

   236

        int rc;                                                                                   \

   237

        _PUBLISH(Topic, QoS, C_type, c_loc_name, Retained)                                        \

   238

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   239

        {{                                                                                        \

   240

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

   241

            /* TODO update status variable accordingly */                                         \

   242

        }}                                                                                        \

   243

    }}

   244

   245

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

   246

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

   247

    {{                                                                                            \

   248

        int rc;                                                                                   \

   249

        _PUBLISH(Topic, QoS, C_type, c_loc_name, Retained)                                        \

   250

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   251

        {{                                                                                        \

   252

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

   253

            /* TODO update status variable accordingly */                                         \

   254

        }} else {{                                                                                \

   255

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   256

        }}                                                                                        \

   257

    }}

   258

   259

static int _connect_mqtt(void)

   260

{{

   261

    int rc;

   262

   263

#ifdef USE_MQTT_5

   264

    MQTTProperties props = MQTTProperties_initializer;

   265

    MQTTProperties willProps = MQTTProperties_initializer;

   266

    MQTTResponse response = MQTTResponse_initializer;

   267

   268

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

   269

    rc = response.reasonCode;

   270

    MQTTResponse_free(response);

   271

#else

   272

    rc = MQTTClient_connect(client, &conn_opts);

   273

#endif

   274

   275

    if (rc != MQTTCLIENT_SUCCESS) {{

   276

        MQTT_is_disconnected = 1;

   277

        return rc;

   278

    }}else{{

   279

        MQTT_is_disconnected = 0;

   280

    }}

   281

   282

{init_pubsub}

   283

   284

    return MQTTCLIENT_SUCCESS;

   285

}}

   286

   287

static void *__MQTT_thread_proc(void *_unused) {{

   288

    int rc = 0;

   289

   290

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

   291

        int do_publish; 

   292

   293

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   294

   295

        if(MQTT_is_disconnected)

   296

        {{

   297

            /* TODO growing retry delay */

   298

            /* TODO max retry delay as config parameter */

   299

            sleep(5);

   300

            rc = _connect_mqtt();

   301

            if (rc == MQTTCLIENT_SUCCESS) {{

   302

                LogInfo("MQTT Reconnected\n");

   303

            }} else {{

   304

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

   305

            }}

   306

        }} 

   307

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   308

        {{

   309

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

   310

{publish_changes}

   311

            MQTT_any_pub_var_changed = 0;

   312

        }}

   313

   314

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   315

   316

        if(MQTT_stop_thread) break;

   317

    }}

   318

   319

    if(!MQTT_stop_thread){{

   320

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

   321

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

   322

    }}

   323

}}

   324

   325

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

   326

{{

   327

    char *uri = "{uri}";

   328

    char *clientID = "{clientID}";

   329

    int rc;

   330

   331

    MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;

   332

   333

#ifdef USE_MQTT_5

   334

    conn_opts.MQTTVersion = MQTTVERSION_5;

   335

    conn_opts.cleanstart = 1;

   336

   337

    createOpts.MQTTVersion = MQTTVERSION_5;

   338

#else

   339

    conn_opts.cleansession = 1;

   340

#endif

   341

   342

    MQTTClient_setTraceCallback(trace_callback);

   343

    MQTTClient_setTraceLevel(MQTT_DEBUG_LEVEL);

   344

   345

    rc = MQTTClient_createWithOptions(

   346

        &client, uri, clientID, MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);

   347

    if (rc != MQTTCLIENT_SUCCESS)

   348

    {{

   349

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

   350

        goto exit_error;

   351

    }}

   352

   353

    rc = MQTTClient_setCallbacks(client, NULL, connectionLost, messageArrived, NULL);

   354

    if (rc != MQTTCLIENT_SUCCESS)

   355

    {{

   356

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

   357

        goto exit_error;

   358

    }}

   359

   360

{init}

   361

   362

    rc = _connect_mqtt();

   363

	if (rc == MQTTCLIENT_SUCCESS) {{

   364

		LogInfo("MQTT Connected\n");

   365

	}} else {{

   366

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

   367

        // Connect error at init is fine, publish thread will retry later

   368

    }}

   369

   370

    /* start MQTT thread */

   371

    MQTT_stop_thread = 0;

   372

    rc = pthread_create(&MQTT_thread, NULL, &__MQTT_thread_proc, NULL);

   373

    if (rc != 0) {{

   374

        LogError("MQTT cannot create thread, return code %d\n", rc);

   375

        goto exit_error;

   376

    }}

   377

   378

    return 0;

   379

   380

exit_error:

   381

    MQTTClient_destroy(&client);

   382

    return rc;

   383

}}

   384

   385

#define READ_VALUE(c_loc_name, C_type) \

   386

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

   387

        /* TODO care about endianess */ \

   388

        PLC_##c_loc_name##_buf = MQTT_##c_loc_name##_buf; \

   389

        MQTT_##c_loc_name##_state = UNCHANGED; \

   390

    }}

   391

   392

void __retrieve_{locstr}(void)

   393

{{

   394

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

   395

{retrieve}

   396

        pthread_mutex_unlock(&MQTT_retrieve_mutex);

   397

    }}

   398

}}

   399

   400

#define WRITE_VALUE(c_loc_name, C_type) \

   401

    /* TODO care about endianess */ \

   402

    if(MQTT_##c_loc_name##_buf != PLC_##c_loc_name##_buf){{ \

   403

        MQTT_##c_loc_name##_buf = PLC_##c_loc_name##_buf; \

   404

        MQTT_##c_loc_name##_state = CHANGED; \

   405

        MQTT_any_pub_var_changed = 1; \

   406

    }}

   407

   408

void __publish_{locstr}(void)

   409

{{

   410

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

   411

        MQTT_any_pub_var_changed = 0;

   412

        /* copy PLC_* variables to MQTT_*, and mark those who changed */

   413

{publish}

   414

        /* if any change detcted, unblock publish thread */

   415

        if(MQTT_any_pub_var_changed || MQTT_is_disconnected){{

   416

            pthread_cond_signal(&MQTT_thread_wakeup);

   417

        }}

   418

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   419

    }} else {{

   420

        /* TODO if couldn't lock mutex set status variable accordingly */ 

   421

    }}

   422

}}

   423