pkg/paho-mqtt: add pkg paho-mqtt

pkg/paho-mqtt/Makefile

@@ -0,0 +1,10 @@
+PKG_NAME = paho-mqtt
+PKG_URL = https://github.com/eclipse/paho.mqtt.embedded-c.git
+PKG_VERSION = 29ab2aa29c5e47794284376d7f8386cfd54c3eed
+PKG_LICENSE = EDL
+
+include $(RIOTBASE)/pkg/pkg.mk
+
+all:
+	"$(MAKE)" -C $(PKG_SOURCE_DIR)/MQTTPacket/src/ -f $(CURDIR)/Makefile.$(PKG_NAME)-packet
+	"$(MAKE)" -C $(PKG_SOURCE_DIR)/MQTTClient-C/src/ -f $(CURDIR)/Makefile.$(PKG_NAME)

pkg/paho-mqtt/Makefile.dep

@@ -0,0 +1,4 @@
+USEMODULE += xtimer
+USEMODULE += paho-mqtt-contrib
+USEMODULE += paho-mqtt-packet
+USEMODULE += tsrb

pkg/paho-mqtt/Makefile.include

@@ -0,0 +1,8 @@
+INCLUDES += -I$(PKGDIRBASE)/paho-mqtt/MQTTClient-C/src/
+INCLUDES += -I$(PKGDIRBASE)/paho-mqtt/MQTTPacket/src
+INCLUDES += -I$(RIOTBASE)/pkg/paho-mqtt/include
+
+DIRS += $(RIOTBASE)/pkg/paho-mqtt/contrib
+
+#define to use MQTT Paho as a task using MQTTStartTask()
+CFLAGS += -DMQTT_TASK=1

pkg/paho-mqtt/Makefile.paho-mqtt

@@ -0,0 +1,10 @@
+MODULE = paho-mqtt
+
+RIOT_MQTT_IFACE_H = paho_mqtt.h
+
+CFLAGS += -DMQTTCLIENT_PLATFORM_HEADER="$(RIOT_MQTT_IFACE_H)"
+CFLAGS += -Wno-sign-compare
+CFLAGS += -Wno-unused-parameter
+CFLAGS += -Wno-address-of-packed-member
+
+include $(RIOTBASE)/Makefile.base

pkg/paho-mqtt/Makefile.paho-mqtt-packet

@@ -0,0 +1,5 @@
+MODULE = paho-mqtt-packet
+
+CFLAGS += -Wno-unused-parameter
+
+include $(RIOTBASE)/Makefile.base

pkg/paho-mqtt/contrib/Makefile

@@ -0,0 +1,3 @@
+MODULE = paho-mqtt-contrib
+
+include $(RIOTBASE)/Makefile.base

pkg/paho-mqtt/contrib/riot_iface.c

@@ -0,0 +1,231 @@
+/*
+ * Copyright (C) 2019  Javier FILEIV <javier.fileiv@gmail.com>
+ *
+ * This file is subject to the terms and conditions of the GNU Lesser
+ * General Public License v2.1. See the file LICENSE in the top level
+ * directory for more details.
+ */
+
+/**
+ * @brief       MQTT common RIOT interface functions
+ *
+ * @author      Javier FILEIV <javier.fileiv@gmail.com>
+ */
+#include <string.h>
+#include <errno.h>
+
+#ifdef MODULE_IPV6_ADDR
+#include "net/ipv6/addr.h"
+#endif
+#ifdef MODULE_IPV4_ADDR
+#include "net/ipv4/addr.h"
+#endif
+#include "net/sock/tcp.h"
+#include "paho_mqtt.h"
+#include "MQTTClient.h"
+#include "xtimer.h"
+#include "tsrb.h"
+#include "log.h"
+
+#define ENABLE_DEBUG                        (0)
+#include "debug.h"
+
+#define IP_MAX_LEN_ADDRESS                  (39)    /*IPv6 max length */
+
+#ifndef TSRB_MAX_SIZE
+#define TSRB_MAX_SIZE                       (1024)
+#endif
+
+#ifdef MODULE_LWIP
+static uint8_t buffer[TSRB_MAX_SIZE];
+static uint8_t _temp_buf[TSRB_MAX_SIZE];
+static tsrb_t tsrb_lwip_tcp;
+#endif
+
+#ifndef PAHO_MQTT_YIELD_MS
+#define PAHO_MQTT_YIELD_MS                  (10)
+#endif
+
+static int mqtt_read(struct Network *n, unsigned char *buf, int len,
+                     int timeout_ms)
+{
+    int _timeout;
+    int _len;
+    void *_buf;
+    int rc = -1;
+
+    if (IS_USED(MODULE_LWIP)) {
+        /* As LWIP doesn't support packet reading byte per byte and
+         * PAHO MQTT reads like that to decode it on the fly,
+         * we read TSRB_MAX_SIZE at once and keep them in a ring buffer.
+         */
+        _buf = _temp_buf;
+        _len = TSRB_MAX_SIZE;
+        _timeout = 0;
+    }
+    else {
+        _buf = buf;
+        _len = len;
+        _timeout = timeout_ms;
+    }
+
+    uint64_t send_tick = xtimer_now64().ticks64 +
+            xtimer_ticks_from_usec64(timeout_ms * US_PER_MS).ticks64;
+    do {
+        rc = sock_tcp_read(&n->sock, _buf, _len, _timeout);
+        if (rc == -EAGAIN) {
+            rc = 0;
+        }
+
+        if (IS_USED(MODULE_LWIP)) {
+            if (rc > 0) {
+                tsrb_add(&tsrb_lwip_tcp, _temp_buf, rc);
+            }
+
+            rc = tsrb_get(&tsrb_lwip_tcp, buf, len);
+        }
+    } while (rc < len && xtimer_now64().ticks64 < send_tick && rc >= 0);
+
+#ifdef ENABLE_DEBUG
+    if (IS_USED(MODULE_LWIP) && rc > 0) {
+        DEBUG("MQTT buf asked for %d, available to read %d\n",
+                rc, tsrb_avail(&tsrb_lwip_tcp));
+        for (int i = 0; i < rc; i++) {
+            DEBUG("0x%02X ", buf[i]);
+        }
+        DEBUG("\n");
+    }
+#endif
+
+    return rc;
+}
+
+static int mqtt_write(struct Network *n, unsigned char *buf, int len,
+                      int timeout_ms)
+{
+    /* timeout is controlled by upper layer in PAHO */
+    (void) timeout_ms;
+
+    return sock_tcp_write(&n->sock, buf, len);
+}
+
+void NetworkInit(Network *n)
+{
+    if (IS_USED(MODULE_LWIP)) {
+        tsrb_init(&tsrb_lwip_tcp, buffer, TSRB_MAX_SIZE);
+    }
+    n->mqttread = mqtt_read;
+    n->mqttwrite = mqtt_write;
+}
+
+int NetworkConnect(Network *n, char *addr_ip, int port)
+{
+    int ret =-1;
+    sock_tcp_ep_t remote = SOCK_IPV4_EP_ANY;
+    char _local_ip[IP_MAX_LEN_ADDRESS];
+
+    strncpy(_local_ip, addr_ip, sizeof(_local_ip));
+    if (IS_USED(MODULE_IPV4_ADDR) &&
+        ipv4_addr_from_str((ipv4_addr_t *)&remote.addr, _local_ip)) {
+            remote.port = port;
+    }
+
+    /* ipvN_addr_from_str modifies input buffer */
+    strncpy(_local_ip, addr_ip, sizeof(_local_ip));
+    if (IS_USED(MODULE_IPV6_ADDR) && (remote.port == 0)  &&
+        ipv6_addr_from_str((ipv6_addr_t *)&remote.addr, _local_ip)) {
+            remote.port = port;
+            remote.family = AF_INET6;
+    }
+
+    if (remote.port == 0) {
+        LOG_ERROR("Error: unable to parse destination address\n");
+        return ret;
+    }
+
+    ret = sock_tcp_connect(&n->sock, &remote, 0, 0);
+    return ret;
+}
+
+void NetworkDisconnect(Network *n)
+{
+    sock_tcp_disconnect(&n->sock);
+}
+
+void TimerInit(Timer *timer)
+{
+    timer->set_ticks.ticks64 = 0;
+    timer->ticks_timeout.ticks64 = 0;
+}
+
+char TimerIsExpired(Timer *timer)
+{
+    return (TimerLeftMS(timer) == 0);
+}
+
+void TimerCountdownMS(Timer *timer, unsigned int timeout_ms)
+{
+    timer->set_ticks = xtimer_now64();
+    timer->ticks_timeout = xtimer_ticks_from_usec64(timeout_ms * US_PER_MS);
+}
+
+void TimerCountdown(Timer *timer, unsigned int timeout_s)
+{
+    TimerCountdownMS(timer, timeout_s * MS_PER_SEC);
+}
+
+int TimerLeftMS(Timer *timer)
+{
+    xtimer_ticks64_t diff_ticks = xtimer_diff64(xtimer_now64(),
+            timer->set_ticks);  /* should be always greater than 0 */
+    if (xtimer_less64(diff_ticks, timer->ticks_timeout)) {
+        diff_ticks = xtimer_diff64(timer->ticks_timeout, diff_ticks);
+        return (xtimer_usec_from_ticks64(diff_ticks) / US_PER_MS);
+    }
+    return 0;
+}
+
+void MutexInit(Mutex *mutex)
+{
+    mutex_init(&mutex->lock);
+}
+
+int MutexLock(Mutex *mutex)
+{
+    mutex_lock(&mutex->lock);
+    return 0;
+}
+
+int MutexUnlock(Mutex *mutex)
+{
+    mutex_unlock(&mutex->lock);
+    return 0;
+}
+
+void *mqtt_riot_run(void *arg)
+{
+    MQTTClient *client = (MQTTClient *)arg;
+    assert(client);
+
+    while (1) {
+        int rc;
+        MutexLock(&client->mutex);
+        if ((rc = MQTTYield(client, PAHO_MQTT_YIELD_MS)) != 0) {
+            LOG_DEBUG("riot_iface: error while MQTTYield()(%d)\n", rc);
+        }
+        MutexUnlock(&client->mutex);
+        /* let other threads do their work */
+        xtimer_usleep(MQTT_YIELD_POLLING_MS * US_PER_MS);
+    }
+    return NULL;
+}
+
+int ThreadStart(Thread *thread, void (*fn)(void *), void *arg)
+{
+    (void) fn;
+    thread->pid = thread_create(thread->stack, sizeof(thread->stack),
+                                MQTT_THREAD_PRIORITY,
+                                THREAD_CREATE_STACKTEST, mqtt_riot_run, arg,
+                                "paho_mqtt_riot");
+    return thread->pid;
+}

pkg/paho-mqtt/doc.txt

@@ -0,0 +1,9 @@
+/**
+ * @defgroup pkg_paho_mqtt   PAHO MQTT framework
+ * @ingroup  pkg
+ * @brief    The Eclipse Paho project provides open-source client implementations of MQTT for embedded systems
+ * @see      https://github.com/eclipse/paho.mqtt.embedded-c
+ *
+ * The Eclipse Paho project provides open-source client
+ * implementations of MQTT.
+ */

pkg/paho-mqtt/include/paho_mqtt.h

@@ -0,0 +1,198 @@
+/*
+ * Copyright (C) 2019  Javier FILEIV <javier.fileiv@gmail.com>
+ *
+ * This file is subject to the terms and conditions of the GNU Lesser
+ * General Public License v2.1. See the file LICENSE in the top level
+ * directory for more details.
+ *
+ */
+
+/**
+ * @addtogroup  pkg_paho_mqtt
+ * @{
+ *
+ * @file
+ * @brief       Network MQTT interface definitions
+ *
+ * @author      Javier FILEIV <javier.fileiv@gmail.com>
+ */
+#ifndef PAHO_MQTT_H
+#define PAHO_MQTT_H
+
+#include "mutex.h"
+#include "xtimer.h"
+#include "thread.h"
+#include "net/sock/tcp.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @brief  RIOT's mqtt paho thread priority
+ */
+#ifndef MQTT_THREAD_PRIORITY
+#define MQTT_THREAD_PRIORITY            (THREAD_PRIORITY_MAIN - 1)
+#endif
+
+/**
+ * @brief  RIOT's mqtt paho thread stack size
+ */
+
+#ifndef MQTT_THREAD_STACKSIZE
+#define MQTT_THREAD_STACKSIZE           (THREAD_STACKSIZE_LARGE)
+#endif
+/**
+ * @brief  MQTT thread YIELD polling time in msecs
+ */
+#ifndef MQTT_YIELD_POLLING_MS
+#define MQTT_YIELD_POLLING_MS           (30)
+#endif
+
+/**
+ * @brief  struct to get time references within mqtt paho
+ *
+ */
+typedef struct {
+    xtimer_ticks64_t set_ticks;             /**< timeout ticks */
+    xtimer_ticks64_t ticks_timeout;         /**< timeout in ticks */
+} Timer;
+
+/**
+ * @brief  Initialize timer struct
+ *
+ * @param timer timer to init
+ */
+void TimerInit(Timer *timer);
+
+/**
+ * @brief  is timer expired?
+ *
+ * @param   timer   timer to check
+ *
+ * @return 1 if timer expired, 0 otherwise
+ */
+char TimerIsExpired(Timer *timer);
+
+/**
+ * @brief  start timer set to milli seconds
+ *
+ * @param   timer   timer to start
+ * @param   msecs   time to set in msecs
+ */
+void TimerCountdownMS(Timer *timer, unsigned int msecs);
+
+/**
+ * @brief  start timer set to seconds
+ *
+ * @param   timer   timer to start
+ * @param   secs   time to set in secs
+ */
+void TimerCountdown(Timer *timer, unsigned int secs);
+
+/**
+ * @brief  Returns millisecs left in timer
+ *
+ * @param   timer   timer to check
+ *
+ * @return  msecs left
+ */
+int TimerLeftMS(Timer *timer);
+
+/**
+ * @brief  Network struct within mqtt paho
+ */
+typedef struct Network {
+    sock_tcp_t sock;    /**< socket number */
+    /**
+     * @brief read internal function
+     */
+    int (*mqttread) (struct Network*, unsigned char*, int, int);
+    /**
+     * @brief write internal function
+     */
+    int (*mqttwrite) (struct Network*, unsigned char*, int, int);
+} Network;
+
+/**
+ * @brief  Initialize network struct
+ *
+ * @param   n   network struct
+ */
+void NetworkInit(Network *n);
+
+/**
+ * @brief  Connect network to host
+ *
+ * @param   n   network struct
+ * @param   address_ip  IP address to connect to
+ * @param   port_number port to connect to
+ *
+ * @return  0 if success, !=0 otherwise
+ */
+int NetworkConnect(Network *n, char* address_ip, int port_number);
+
+/**
+ * @brief  Disconnect network
+ *
+ * @param   n       network struct
+ */
+void NetworkDisconnect(Network *n);
+
+/**
+ * @brief  Mutex struct within mqtt paho
+ */
+typedef struct {
+    mutex_t lock;   /**< MQTT thread mutex*/
+} Mutex;
+
+/**
+ * @brief  Initialize mutex struct
+ *
+ * @param  mutex pointer
+ */
+void MutexInit(Mutex *mutex);
+
+/**
+ * @brief  Locks mutex struct
+ *
+ * @param  mutex pointer
+ *
+ * @return  0 if success, !=0 otherwise
+ */
+int MutexLock(Mutex *mutex);
+
+/**
+ * @brief  Unlocks mutex struct
+ *
+ * @param  mutex pointer
+ *
+ * @return  0 if success, !=0 otherwise
+ */
+int MutexUnlock(Mutex *mutex);
+
+/**
+ * @brief  Thread struct within mqtt paho
+ */
+typedef struct {
+    char stack[MQTT_THREAD_STACKSIZE];      /**< stack for MQTT thread*/
+    kernel_pid_t pid;                       /**< MQTT thread pid*/
+} Thread;
+
+/**
+ * @brief  Start new thread
+ *
+ * @param  thread to start
+ * @param  fn  pointer function to execute
+ * @param  arg arguments to pass to that fn
+ *
+ * @return  0 if success, !=0 otherwise
+ */
+int ThreadStart(Thread *thread, void (*fn)(void *), void *arg);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PAHO_MQTT_H */
+/** @} */