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>

    11

    12

#include "MQTTClient.h"

    13

#include "MQTTClientPersistence.h"

    14

    17

#define _Log(level, ...)                                                                          \

    18

    {{                                                                                            \

    19

        char mstr[256];                                                                           \

    20

        snprintf(mstr, 255, __VA_ARGS__);                                                         \

    21

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

    22

        printf(__VA_ARGS__);                                                                      \

    23

        fflush(stdout);                                                                           \

    24

    }}

    25

    26

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

    27

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

    28

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

    29

    30

// Selected debug level for paho stack

    31

// can be:

    32

// MQTTCLIENT_TRACE_PROTOCOL, MQTTCLIENT_TRACE_MAXIMUM, MQTTCLIENT_TRACE_ERROR

    33

#define MQTT_DEBUG_LEVEL MQTTCLIENT_TRACE_ERROR

    34

    35

void trace_callback(enum MQTTCLIENT_TRACE_LEVELS level, char* message)

    36

{{

    37

    if(level >= MQTT_DEBUG_LEVEL)

    38

    {{

    39

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

    40

                                (level > MQTTCLIENT_TRACE_MINIMUM) ? LOG_WARNING : 

    41

                                LOG_INFO;

    42

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

    43

    }}

    44

}}

    45

    46

#define CHANGED 1

    47

#define UNCHANGED 0

    48

    49

#define DECL_VAR(iec_type, C_type, c_loc_name)                                                     \

    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

   106

static MQTTClient client;

   107

#ifdef USE_MQTT_5

   108

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;

   109

#else

   110

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;

   111

#endif

   112

   113

MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;

   114

   115

/* condition to quit publish thread */

   116

static int MQTT_stop_thread = 0;

   117

   118

/* condition to wakeup publish thread */

   119

static int MQTT_any_pub_var_changed = 0;

   120

   121

/* Keep track of connection state */

   122

static volatile int MQTT_is_disconnected = 1;

   123

   124

/* mutex to keep incoming PLC data consistent */

   125

static pthread_mutex_t MQTT_retrieve_mutex = PTHREAD_MUTEX_INITIALIZER;

   126

   127

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

   128

static pthread_mutex_t MQTT_thread_wakeup_mutex = PTHREAD_MUTEX_INITIALIZER;

   129

   130

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

   131

static pthread_cond_t MQTT_thread_wakeup = PTHREAD_COND_INITIALIZER;

   132

   133

/* thread that handles publish and reconnection */

   134

static pthread_t MQTT_thread;

   135

   143

   144

static struct {{

   145

    const char *topic; //null terminated topic string

   146

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

   147

    int *mqtt_pchanged; // pointer to changed flag

   153

}} topics [] = {{

   154

{topics}

   155

}};

   156

   157

void __cleanup_{locstr}(void)

   158

{{

   159

    int rc;

   160

   161

    /* stop publish thread */

   162

    MQTT_stop_thread = 1;

   163

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

   164

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

   165

        pthread_cond_signal(&MQTT_thread_wakeup);

   166

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   167

    }}

   168

    pthread_join(MQTT_thread, NULL);

   169

   170

#ifdef USE_MQTT_5

   171

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

   172

#else

   173

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

   174

#endif

   175

    {{

   176

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

   177

    }}

   178

    MQTTClient_destroy(&client);

   179

}}

   180

   181

void connectionLost(void* context, char* reason)

   182

{{

   183

    int rc;

   184

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

   185

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

   186

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

   187

        MQTT_is_disconnected = 1;

   188

        pthread_cond_signal(&MQTT_thread_wakeup);

   189

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   190

    }}

   191

}}

   192

   193

   194

   195

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

   196

{{

   197

    int low = 0;

   198

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

   199

    int high = size - 1;

   200

    int mid;

   202

   203

    // bisect topic among subscribed topics

   204

    while (low <= high) {{

   205

        int res;

   206

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

   207

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

   208

   209

        // Check if key is present at mid

   210

        if (res == 0)

   211

            goto found;

   212

   213

        // If key greater, ignore left half

   214

        if (res < 0)

   215

            low = mid + 1;

   216

   217

        // If key is smaller, ignore right half

   218

        else

   219

            high = mid - 1;

   220

    }}

   221

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

   222

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

   223

    goto exit;

   224

   225

found:

   229

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

   235

            *topics[mid].mqtt_pchanged = 1;

   236

            pthread_mutex_unlock(&MQTT_retrieve_mutex);

   237

        }}

   238

    }} else {{

   239

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

   240

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

   241

    }}

   242

exit:

   243

    MQTTClient_freeMessage(&message);

   244

    MQTTClient_free(topicName);

   245

    return 1;

   246

}}

   247

   248

#define INIT_NoAuth()                                                                             \

   249

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

   250

   251

#define INIT_x509(Verify, KeyStore, TrustStore)                                                   \

   253

    ssl_opts.verify = Verify;                                                                     \

   254

    ssl_opts.keyStore = KeyStore;                                                                 \

   255

    ssl_opts.trustStore = TrustStore;                                                             \

   256

    conn_opts.ssl = &ssl_opts;

   257

   258

#define INIT_PSK(Secret, ID)                                                                      \

   259

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

   260

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

   261

    /* ssl_opts.ssl_psk_cb = TODO; */                                                             \

   262

    /* ssl_opts.ssl_psk_context = TODO; */                                                        \

   263

    conn_opts.ssl = &ssl_opts;

   264

   265

#define INIT_UserPassword(User, Password)                                                         \

   266

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

   267

    conn_opts.username = User;                                                                    \

   268

    conn_opts.password = Password;

   269

   270

#ifdef USE_MQTT_5

   271

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   272

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

   273

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

   274

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

   275

        MQTTResponse_free(response);

   276

#else

   277

#define _SUBSCRIBE(Topic, QoS)                                                                    \

   278

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

   279

#endif

   280

   281

#define INIT_SUBSCRIPTION(Topic, QoS)                                                             \

   282

    {{                                                                                            \

   283

        int rc;                                                                                   \

   284

        _SUBSCRIBE(Topic, QoS)                                                                    \

   285

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   286

        {{                                                                                        \

   287

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

   288

        }}                                                                                        \

   289

    }}

   290

   291

   292

#ifdef USE_MQTT_5

   296

        rc = response.reasonCode;                                                                 \

   297

        MQTTResponse_free(response);

   298

#else

   302

#endif

   303

   314

    {{                                                                                            \

   315

        int rc;                                                                                   \

   317

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   318

        {{                                                                                        \

   319

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

   320

            /* TODO update status variable accordingly */                                         \

   321

        }}                                                                                        \

   322

    }}

   323

   325

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

   326

    {{                                                                                            \

   327

        int rc;                                                                                   \

   329

        if (rc != MQTTCLIENT_SUCCESS)                                                             \

   330

        {{                                                                                        \

   331

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

   332

            /* TODO update status variable accordingly */                                         \

   333

        }} else {{                                                                                \

   334

            MQTT_##c_loc_name##_state = UNCHANGED;                                                \

   335

        }}                                                                                        \

   336

    }}

   337

   338

static int _connect_mqtt(void)

   339

{{

   340

    int rc;

   341

   342

#ifdef USE_MQTT_5

   343

    MQTTProperties props = MQTTProperties_initializer;

   344

    MQTTProperties willProps = MQTTProperties_initializer;

   345

    MQTTResponse response = MQTTResponse_initializer;

   346

   347

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

   348

    rc = response.reasonCode;

   349

    MQTTResponse_free(response);

   350

#else

   351

    rc = MQTTClient_connect(client, &conn_opts);

   352

#endif

   353

   354

    if (rc != MQTTCLIENT_SUCCESS) {{

   355

        MQTT_is_disconnected = 1;

   356

        return rc;

   357

    }}else{{

   358

        MQTT_is_disconnected = 0;

   359

    }}

   360

   361

{init_pubsub}

   362

   363

    return MQTTCLIENT_SUCCESS;

   364

}}

   365

   366

static void *__MQTT_thread_proc(void *_unused) {{

   367

    int rc = 0;

   368

   369

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

   370

        int do_publish; 

   371

   372

        pthread_cond_wait(&MQTT_thread_wakeup, &MQTT_thread_wakeup_mutex);

   373

   374

        if(MQTT_is_disconnected)

   375

        {{

   376

            /* TODO growing retry delay */

   377

            /* TODO max retry delay as config parameter */

   378

            sleep(5);

   379

            rc = _connect_mqtt();

   380

            if (rc == MQTTCLIENT_SUCCESS) {{

   381

                LogInfo("MQTT Reconnected\n");

   382

            }} else {{

   383

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

   384

            }}

   385

        }} 

   386

        if(!MQTT_is_disconnected && MQTT_any_pub_var_changed)

   387

        {{

   388

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

   389

{publish_changes}

   390

            MQTT_any_pub_var_changed = 0;

   391

        }}

   392

   393

        pthread_mutex_unlock(&MQTT_thread_wakeup_mutex);

   394

   395

        if(MQTT_stop_thread) break;

   396

    }}

   397

   398

    if(!MQTT_stop_thread){{

   399

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

   400

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

   401

    }}

   402

}}

   403

   404

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

   405

{{

   406

    char *uri = "{uri}";

   407

    char *clientID = "{clientID}";

   408

    int rc;

   409

   410

    MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;

   411

   412

#ifdef USE_MQTT_5

   413

    conn_opts.MQTTVersion = MQTTVERSION_5;

   414

    conn_opts.cleanstart = 1;

   415

   416

    createOpts.MQTTVersion = MQTTVERSION_5;

   417

#else

   418

    conn_opts.cleansession = 1;

   419

#endif

   420

   421

    MQTTClient_setTraceCallback(trace_callback);

   422

    MQTTClient_setTraceLevel(MQTT_DEBUG_LEVEL);

   423

   424

    rc = MQTTClient_createWithOptions(

   425

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

   426

    if (rc != MQTTCLIENT_SUCCESS)

   427

    {{

   428

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

   429

        goto exit_error;

   430

    }}

   431

   432

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

   433

    if (rc != MQTTCLIENT_SUCCESS)

   434

    {{

   435

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

   436

        goto exit_error;

   437

    }}

   438

   439

{init}

   440

   441

    rc = _connect_mqtt();

   442

	if (rc == MQTTCLIENT_SUCCESS) {{

   443

		LogInfo("MQTT Connected\n");

   444

	}} else {{

   445

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

   446

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

   447

    }}

   448

   449

    /* start MQTT thread */

   450

    MQTT_stop_thread = 0;

   451

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

   452

    if (rc != 0) {{

   453

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

   454

        goto exit_error;

   455

    }}

   456

   457

    return 0;

   458

   459

exit_error:

   460

    MQTTClient_destroy(&client);

   461

    return rc;

   462

}}

   463

   464

#define READ_VALUE(c_loc_name, C_type) \

   465

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

   466

        /* TODO care about endianess */ \

   467

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

   468

        MQTT_##c_loc_name##_state = UNCHANGED; \

   469

    }}

   470

   471

void __retrieve_{locstr}(void)

   472

{{

   473

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

   474

{retrieve}

   475

        pthread_mutex_unlock(&MQTT_retrieve_mutex);

   476

    }}

   477

}}

   478

   479

#define WRITE_VALUE(c_loc_name, C_type) \

   480

    /* TODO care about endianess */ \

   482

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

   483

        MQTT_##c_loc_name##_state = CHANGED; \

   484

        MQTT_any_pub_var_changed = 1; \

   485

    }}

   486

   487

void __publish_{locstr}(void)