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

/* condition to quit publish thread */

    60

static int MQTT_stop_thread = 0;

    61

    62

/* condition to wakeup publish thread */

    63

static int MQTT_any_pub_var_changed = 0;

    64

    65

/* Keep track of connection state */

    66

static volatile int MQTT_is_disconnected = 1;

    67

    68

/* mutex to keep incoming PLC data consistent */

    69

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

    70

    71

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

    72

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

    73

    74

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

    75

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

    76

    77

/* thread that handles publish and reconnection */

    78

static pthread_t MQTT_thread;

    79

    80

#define INIT_TOPIC(topic, iec_type, c_loc_name)                                                    \

    81

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

    82

    83

static struct {{

    84

    const char *topic; //null terminated topic string

    85

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

    86

    int *mqtt_pchanged; // pointer to changed flag

    87

    __IEC_types_enum vartype;

    88

}} topics [] = {{

    89

{topics}

    90

}};

    91

    92

void __cleanup_{locstr}(void)

    93

{{

    94

    int rc;

    95

    96

    /* stop publish thread */

    97

    MQTT_stop_thread = 1;

    98

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

    99

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

   100

        pthread_cond_signal(&MQTT_thread_wakeup);

   101

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   102

    }}

   103

    pthread_join(MQTT_thread, NULL);

   104

   105

#ifdef USE_MQTT_5

   106

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

   107

#else

   108

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

   109

#endif

   110

    {{

   111

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

   112

    }}

   113

    MQTTClient_destroy(&client);

   114

}}

   115

   116

void connectionLost(void* context, char* reason)

   117

{{

   118

    int rc;

   119

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

   120

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

   121

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

   122

        MQTT_is_disconnected = 1;

   123

        pthread_cond_signal(&MQTT_thread_wakeup);

   124

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   125

    }}

   126

}}

   127

   128

   129

   130

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

   131

{{

   132

    int low = 0;

   133

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

   134

    int high = size - 1;

   135

    int mid;

   136

   137

    // bisect topic among subscribed topics

   138

    while (low <= high) {{

   139

        int res;

   140

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

   141

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

   142

   143

        // Check if key is present at mid

   144

        if (res == 0)

   145

            goto found;

   146

   147

        // If key greater, ignore left half

   148

        if (res < 0)

   149

            low = mid + 1;

   150

   151

        // If key is smaller, ignore right half

   152

        else

   153

            high = mid - 1;

   154

    }}

   155

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

   156

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

   157

    goto exit;

   158

   159

found:

   160

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

   161

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

   162

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

   163

            *topics[mid].mqtt_pchanged = 1;

   164

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   165

        }}

   166

    }} else {{

   167

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

   168

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

   169

    }}

   170

exit:

   171

    MQTTClient_freeMessage(&message);

   172

    MQTTClient_free(topicName);

   173

    return 1;

   174

}}

   175

   176

#define INIT_NoAuth()                                                                             \

   177

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

   178

   179

#define INIT_x509(PrivateKey, Certificate)                                                        \

   180

    LogInfo("MQTT Init x509 %s,%s\n", PrivateKey, Certificate);

   181

    /* TODO */

   182

   183

#define INIT_UserPassword(User, Password)                                                         \

   184

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

   185

    conn_opts.username = User;                                                                    \

   186

    conn_opts.password = Password;

   187

   188

#ifdef USE_MQTT_5

   189

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   190

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

   191

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

   192

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

   193

        MQTTResponse_free(response);

   194

#else

   195

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   196

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

   197

#endif

   198

   199

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   200

    {{                                                                                            \

   201

        int rc;                                                                                   \

   202

        _SUBSCRIBE(Topic, QoS)                                                                    \

   203

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   204

        {{                                                                                        \

   205

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

   206

        }}                                                                                        \

   207

    }}

   208

   209

   210

#ifdef USE_MQTT_5

   211

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

   212

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

   213

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

   214

        rc = response.reasonCode;                                                                 \

   215

        MQTTResponse_free(response);

   216

#else

   217

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

   218

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

   219

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

   220

#endif

   221

   222

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

   223

    {{                                                                                            \

   224

        int rc;                                                                                   \

   225

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

   226

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   227

        {{                                                                                        \

   228

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

   229

            /* TODO update status variable accordingly */                                         \

   230

        }}                                                                                        \

   231

    }}

   232

   233

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

   234

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

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

   241

            /* TODO update status variable accordingly */                                         \

   242

        }} else {{                                                                                \

   243

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   244

        }}                                                                                        \

   245

    }}

   246

   247

static int _connect_mqtt(void)

   248

{{

   249

    int rc;

   250

   251

#ifdef USE_MQTT_5

   252

    MQTTProperties props = MQTTProperties_initializer;

   253

    MQTTProperties willProps = MQTTProperties_initializer;

   254

    MQTTResponse response = MQTTResponse_initializer;

   255

   256

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

   257

    rc = response.reasonCode;

   258

    MQTTResponse_free(response);

   259

#else

   260

    rc = MQTTClient_connect(client, &conn_opts);

   261

#endif

   262

   263

    if (rc != MQTTCLIENT_SUCCESS) {{

   264

        MQTT_is_disconnected = 1;

   265

        return rc;

   266

    }}else{{

   267

        MQTT_is_disconnected = 0;

   268

    }}

   269

   270

{init_pubsub}

   271

   272

    return MQTTCLIENT_SUCCESS;

   273

}}

   274

   275

static void *__MQTT_thread_proc(void *_unused) {{

   276

    int rc = 0;

   277

   278

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

   279

        int do_publish; 

   280

   281

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   282

   283

        if(MQTT_is_disconnected)

   284

        {{

   285

            /* TODO growing retry delay */

   286

            /* TODO max retry delay as config parameter */

   287

            sleep(5);

   288

            rc = _connect_mqtt();

   289

            if (rc == MQTTCLIENT_SUCCESS) {{

   290

                LogInfo("MQTT Reconnected\n");

   291

            }} else {{

   292

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

   293

            }}

   294

        }} 

   295

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   296

        {{

   297

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

   298

{publish_changes}

   299

            MQTT_any_pub_var_changed = 0;

   300

        }}

   301

   302

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   303

   304

        if(MQTT_stop_thread) break;

   305

    }}

   306

   307

    if(!MQTT_stop_thread){{

   308

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

   309

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

   310

    }}

   311

}}

   312

   313

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

   314

{{

   315

    char *uri = "{uri}";

   316

    char *clientID = "{clientID}";

   317

    int rc;

   318

   319

    MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;

   320

   321

#ifdef USE_MQTT_5

   322

    conn_opts.MQTTVersion = MQTTVERSION_5;

   323

    conn_opts.cleanstart = 1;

   324

   325

    createOpts.MQTTVersion = MQTTVERSION_5;

   326

#else

   327

    conn_opts.cleansession = 1;

   328

#endif

   329

   330

    MQTTClient_setTraceCallback(trace_callback);

   331

    MQTTClient_setTraceLevel(MQTT_DEBUG_LEVEL);

   332

   333

    rc = MQTTClient_createWithOptions(

   334

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

   335

    if (rc != MQTTCLIENT_SUCCESS)

   336

    {{

   337

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

   338

        goto exit_error;

   339

    }}

   340

   341

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

   342

    if (rc != MQTTCLIENT_SUCCESS)

   343

    {{

   344

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

   345

        goto exit_error;

   346

    }}

   347

   348

{init}

   349

   350

    rc = _connect_mqtt();

   351

	if (rc == MQTTCLIENT_SUCCESS) {{

   352

		LogInfo("MQTT Connected\n");

   353

	}} else {{

   354

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

   355

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

   356

    }}

   357

   358

    /* start MQTT thread */

   359

    MQTT_stop_thread = 0;

   360

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

   361

    if (rc != 0) {{

   362

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

   363

        goto exit_error;

   364

    }}

   365

   366

    return 0;

   367

   368

exit_error:

   369

    MQTTClient_destroy(&client);

   370

    return rc;

   371

}}

   372

   373

#define READ_VALUE(c_loc_name, C_type) \

   374

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

   375

        /* TODO care about endianess */ \

   376

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

   377

        MQTT_##c_loc_name##_state = UNCHANGED; \

   378

    }}

   379

   380

void __retrieve_{locstr}(void)

   381

{{

   382

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

   383

{retrieve}

   384

        pthread_mutex_unlock(&MQTT_retrieve_mutex);

   385

    }}

   386

}}

   387

   388

#define WRITE_VALUE(c_loc_name, C_type) \

   389

    /* TODO care about endianess */ \

   390

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

   391

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

   392

        MQTT_##c_loc_name##_state = CHANGED; \

   393

        MQTT_any_pub_var_changed = 1; \

   394

    }}

   395

   396

void __publish_{locstr}(void)

   397

{{

   398

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

   399

        MQTT_any_pub_var_changed = 0;

   400

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

   401

{publish}

   402

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

   403

        if(MQTT_any_pub_var_changed || MQTT_is_disconnected){{

   404

            pthread_cond_signal(&MQTT_thread_wakeup);

   405

        }}

   406

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   407

    }} else {{

   408

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

   409

    }}

   410

}}

   411