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

/* mutex to keep incoming PLC data consistent */

    66

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

    67

    68

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

    69

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

    70

    71

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

    72

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

    73

    74

/* thread that handles publish and reconnection */

    75

static pthread_t MQTT_thread;

    76

    77

#define INIT_TOPIC(topic, iec_type, c_loc_name)                                                    \

    78

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

    79

    80

static struct {{

    81

    const char *topic; //null terminated topic string

    82

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

    83

    int *mqtt_pchanged; // pointer to changed flag

    84

    __IEC_types_enum vartype;

    85

}} topics [] = {{

    86

{topics}

    87

}};

    88

    89

void __cleanup_{locstr}(void)

    90

{{

    91

    int rc;

    92

    93

    /* stop publish thread */

    94

    MQTT_stop_thread = 1;

    95

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

    96

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

    97

        pthread_cond_signal(&MQTT_thread_wakeup);

    98

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

    99

    }}

   100

    pthread_join(MQTT_thread, NULL);

   101

   102

#ifdef USE_MQTT_5

   103

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

   104

#else

   105

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

   106

#endif

   107

    {{

   108

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

   109

    }}

   110

    MQTTClient_destroy(&client);

   111

}}

   112

   113

void connectionLost(void* context, char* reason)

   114

{{

   115

    int rc;

   116

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

   117

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

   118

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

   119

        pthread_cond_signal(&MQTT_thread_wakeup);

   120

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   121

    }}

   122

}}

   123

   124

   125

   126

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

   127

{{

   128

    int low = 0;

   129

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

   130

    int high = size - 1;

   131

    int mid;

   132

   133

    // bisect topic among subscribed topics

   134

    while (low <= high) {{

   135

        int res;

   136

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

   137

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

   138

   139

        // Check if key is present at mid

   140

        if (res == 0)

   141

            goto found;

   142

   143

        // If key greater, ignore left half

   144

        if (res < 0)

   145

            low = mid + 1;

   146

   147

        // If key is smaller, ignore right half

   148

        else

   149

            high = mid - 1;

   150

    }}

   151

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

   152

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

   153

    goto exit;

   154

   155

found:

   156

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

   157

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

   158

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

   159

            *topics[mid].mqtt_pchanged = 1;

   160

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   161

        }}

   162

    }} else {{

   163

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

   164

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

   165

    }}

   166

exit:

   167

    MQTTClient_freeMessage(&message);

   168

    MQTTClient_free(topicName);

   169

    return 1;

   170

}}

   171

   172

#define INIT_NoAuth()                                                                             \

   173

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

   174

   175

#define INIT_x509(PrivateKey, Certificate)                                                        \

   176

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

   177

    /* TODO */

   178

   179

#define INIT_UserPassword(User, Password)                                                         \

   180

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

   181

    conn_opts.username = User;                                                                    \

   182

    conn_opts.password = Password;

   183

   184

#ifdef USE_MQTT_5

   185

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   186

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

   187

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

   188

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

   189

        MQTTResponse_free(response);

   190

#else

   191

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   192

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

   193

#endif

   194

   195

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   196

    {{                                                                                            \

   197

        int rc;                                                                                   \

   198

        _SUBSCRIBE(Topic, QoS)                                                                    \

   199

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   200

        {{                                                                                        \

   201

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

   202

        }}                                                                                        \

   203

    }}

   204

   205

   206

#ifdef USE_MQTT_5

   207

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

   208

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

   209

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

   210

        rc = response.reasonCode;                                                                 \

   211

        MQTTResponse_free(response);

   212

#else

   213

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

   214

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

   215

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

   216

#endif

   217

   218

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

   219

    {{                                                                                            \

   220

        int rc;                                                                                   \

   221

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

   222

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   223

        {{                                                                                        \

   224

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

   225

            /* TODO update status variable accordingly */                                         \

   226

        }}                                                                                        \

   227

    }}

   228

   229

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

   230

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

   231

    {{                                                                                            \

   232

        int rc;                                                                                   \

   233

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

   234

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   235

        {{                                                                                        \

   236

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

   237

            /* TODO update status variable accordingly */                                         \

   238

        }} else {{                                                                                \

   239

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   240

        }}                                                                                        \

   241

    }}

   242

   243

static int _connect_mqtt(void)

   244

{{

   245

    int rc;

   246

   247

#ifdef USE_MQTT_5

   248

    MQTTProperties props = MQTTProperties_initializer;

   249

    MQTTProperties willProps = MQTTProperties_initializer;

   250

    MQTTResponse response = MQTTResponse_initializer;

   251

   252

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

   253

    rc = response.reasonCode;

   254

    MQTTResponse_free(response);

   255

#else

   256

    rc = MQTTClient_connect(client, &conn_opts);

   257

#endif

   258

   259

    if (rc != MQTTCLIENT_SUCCESS) {{

   260

        return rc;

   261

    }}

   262

   263

{init_pubsub}

   264

   265

    return MQTTCLIENT_SUCCESS;

   266

}}

   267

   268

static void *__MQTT_thread_proc(void *_unused) {{

   269

    int rc = 0;

   270

   271

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

   272

        int do_publish; 

   273

        int is_connected;

   274

   275

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   276

        is_connected = MQTTClient_isConnected(client);

   277

        do_publish = MQTT_any_pub_var_changed && is_connected; 

   278

   279

        if(do_publish)

   280

        {{

   281

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

   282

{publish_changes}

   283

            MQTT_any_pub_var_changed = 0;

   284

        }}

   285

   286

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   287

   288

        if(!is_connected) {{

   289

            rc = _connect_mqtt();

   290

            if (rc == MQTTCLIENT_SUCCESS) {{

   291

                LogInfo("MQTT Reconnected\n");

   292

            }} else {{

   293

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

   294

                sleep(5);

   295

            }}

   296

        }}

   297

   298

        if(MQTT_stop_thread) break;

   299

    }}

   300

   301

    if(!MQTT_stop_thread){{

   302

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

   303

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

   304

    }}

   305

}}

   306

   307

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

   308

{{

   309

    char *uri = "{uri}";

   310

    char *clientID = "{clientID}";

   311

    int rc;

   312

   313

    MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;

   314

   315

#ifdef USE_MQTT_5

   316

    conn_opts.MQTTVersion = MQTTVERSION_5;

   317

    conn_opts.cleanstart = 1;

   318

   319

    createOpts.MQTTVersion = MQTTVERSION_5;

   320

#else

   321

    conn_opts.cleansession = 1;

   322

#endif

   323

   324

    MQTTClient_setTraceCallback(trace_callback);

   325

    MQTTClient_setTraceLevel(MQTT_DEBUG_LEVEL);

   326

   327

    rc = MQTTClient_createWithOptions(

   328

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

   329

    if (rc != MQTTCLIENT_SUCCESS)

   330

    {{

   331

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

   332

        goto exit_error;

   333

    }}

   334

   335

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

   336

    if (rc != MQTTCLIENT_SUCCESS)

   337

    {{

   338

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

   339

        goto exit_error;

   340

    }}

   341

   342

{init}

   343

   344

    rc = _connect_mqtt();

   345

	if (rc == MQTTCLIENT_SUCCESS) {{

   346

		LogInfo("MQTT Connected\n");

   347

	}} else {{

   348

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

   349

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

   350

    }}

   351

   352

    /* start MQTT thread */

   353

    MQTT_stop_thread = 0;

   354

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

   355

    if (rc != 0) {{

   356

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

   357

        goto exit_error;

   358

    }}

   359

   360

    return 0;

   361

   362

exit_error:

   363

    MQTTClient_destroy(&client);

   364

    return rc;

   365

}}

   366

   367

#define READ_VALUE(c_loc_name, C_type) \

   368

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

   369

        /* TODO care about endianess */ \

   370

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

   371

        MQTT_##c_loc_name##_state = UNCHANGED; \

   372

    }}

   373

   374

void __retrieve_{locstr}(void)

   375

{{

   376

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

   377

{retrieve}

   378

        pthread_mutex_unlock(&MQTT_retrieve_mutex);

   379

    }}

   380

}}

   381

   382

#define WRITE_VALUE(c_loc_name, C_type) \

   383

    /* TODO care about endianess */ \

   384

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

   385

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

   386

        MQTT_##c_loc_name##_state = CHANGED; \

   387

        MQTT_any_pub_var_changed = 1; \

   388

    }}

   389

   390

void __publish_{locstr}(void)

   391

{{

   392

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

   393

        MQTT_any_pub_var_changed = 0;

   394

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

   395

{publish}

   396

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

   397

        if(MQTT_any_pub_var_changed){{

   398

            pthread_cond_signal(&MQTT_thread_wakeup);

   399

        }}

   400

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   401

    }} else {{

   402

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

   403

    }}

   404

}}

   405