main.mustache

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <random.h>

#include "shell.h"
#include "msg.h"
#include "net/emcute.h"
#include "net/ipv6/addr.h"

#ifndef EMCUTE_ID
#define EMCUTE_ID           ("gertrud")
#endif
#define EMCUTE_PORT         (1883U)
#define EMCUTE_PRIO         (THREAD_PRIORITY_MAIN - 1)

#define NUMOFSUBS           (16U)
#define TOPIC_MAXLEN        (64U)

static char stack[THREAD_STACKSIZE_DEFAULT];
static msg_t queue[8];

static emcute_sub_t subscriptions[NUMOFSUBS];
static char topics[NUMOFSUBS][TOPIC_MAXLEN];

static void *emcute_thread(void *arg)
{
    (void)arg;
    emcute_run(EMCUTE_PORT, EMCUTE_ID);
    return NULL;    /* should never be reached */
}

static unsigned get_qos(const char *str)
{
    int qos = atoi(str);
    switch (qos) {
        case 1:     return EMCUTE_QOS_1;
        case 2:     return EMCUTE_QOS_2;
        default:    return EMCUTE_QOS_0;
    }
}

{{>callback}}

static void connect (void)
{
   sock_udp_ep_t gw = { .family = AF_INET6, .port = EMCUTE_PORT };
   char *topic = NULL;
   char *message = NULL;
   size_t len = 0;

    /* parse address */
    if (ipv6_addr_from_str((ipv6_addr_t *)&gw.addr.ipv6, {{brokerAddress}} == NULL) {
        printf("error parsing IPv6 address\n");
        return ;
    }


    if (emcute_con(&gw, true, topic, message, len, 0) != EMCUTE_OK) {
        printf("error: unable to connect to [%s]:%i\n",{{brokerAddress}}, (int)gw.port);
        return ;
    }
    printf("Successfully connected to gateway at [%s]:%i\n",
        {{brokerAddress}}, (int)gw.port);
}

static void disconnect(void)
{
   int res = emcute_discon();
    if (res == EMCUTE_NOGW) {
        puts("error: not connected to any broker");
        return ;
    }
    else if (res != EMCUTE_OK) {
        puts("error: unable to disconnect");
        return ;
    }
    puts("Disconnect successful");
}

{{#operations}}{{#operation}} 
{{#isSubscribe}}
static int {{operationId}}(void)
{
   if (strlen({{topic}}) > TOPIC_MAXLEN) {
       puts("error: topic name exceeds maximum possible size");
       return 1;
   }
   {{#qos}}
   if (argc >= 3) {
       flags |= get_qos({{qos}});
   }{{/qos}}

   /* find empty subscription slot */
   unsigned i = 0;
   for (; (i < NUMOFSUBS) && (subscriptions[i].topic.id != 0); i++) {}
   if (i == NUMOFSUBS) {
        puts("error: no memory to store new subscriptions");
        return 1;
    }

    subscriptions[i].cb = {{operationId}}_onPub;
    strcpy(topics[i], {{topic}});
    subscriptions[i].topic.name = topics[i];
    if (emcute_sub(&subscriptions[i], flags) != EMCUTE_OK) {
        printf("error: unable to subscribe to %s\n", {{topic}});
        return 1;
    }

    printf("Now subscribed to %s\n",{{topic}});
    return 0;
}
{{/isSubscribe}}

{{^isSubscribe}}
static int {{operationId}}(const void* buf)
{
   emcute_topic_t t;
   unsigned flags = EMCUTE_QOS_0;

    /* parse QoS level */
    {{#qos}}
    if (argc >= 4) {
        flags |= get_qos({{qos}});
    }
    {{/qos}}

    printf("pub with topic: %s and name %s and flags 0x%02x\n",{{topic}}, buf, (int)flags);

    /* step 1: get topic id */
    t.name = {{topic}};
    if (emcute_reg(&t) != EMCUTE_OK) {
        puts("error: unable to obtain topic ID");
        return 1;
    }

    /* step 2: publish data */
    if (emcute_pub(&t, buf, strlen(buf), flags) != EMCUTE_OK) {
        printf("error: unable to publish data to topic '%s [%i]'\n",
                t.name, (int)t.id);
        return 1;
    }

    printf("Published %i bytes to topic '%s [%i]'\n",
            (int)strlen(buf), t.name, t.id);

    return 0;
}
{{/isSubscribe}}
{{/operation}}{{/operations}}
 
 int main(void)
{
    puts("MQTT-SN IoT application\n");
    puts("Type 'help' to get started. Have a look at the README.md for more"
         "information.");

    /* the main thread needs a msg queue to be able to run `ping6`*/
    msg_init_queue(queue, ARRAY_SIZE(queue));

    /* initialize our subscription buffers */
    memset(subscriptions, 0, (NUMOFSUBS * sizeof(emcute_sub_t)));

    /* start the emcute thread */
    thread_create(stack, sizeof(stack), EMCUTE_PRIO, 0,
                  emcute_thread, NULL, "emcute");
                  
    char line_buf[SHELL_DEFAULT_BUFSIZE];
    shell_run(NULL, line_buf, SHELL_DEFAULT_BUFSIZE);
    
    /* calls function to connect to broker*/
    connect(); 
    
    /*all pub-sub functions to be called here*/
    
    /*disconnect from broker before closing the application*/
    disconnect();
    
    return 0;
}