Hi!
I am trying to merge the basic/button sample with the mqtt-simple sample.
I added
CONFIG_GPIO=yto the prj.conf of mqtt-simple sample.
My main.c looks like this:
#include <zephyr.h>
#include <stdio.h>
#include <drivers/uart.h>
#include <string.h>
#include <drivers/gpio.h>
#include <net/mqtt.h>
#include <net/socket.h>
#include <lte_lc.h>
#if defined(CONFIG_LWM2M_CARRIER)
#include <lwm2m_carrier.h>
#endif
/* Buffers for MQTT client. */
static u8_t rx_buffer[CONFIG_MQTT_MESSAGE_BUFFER_SIZE];
static u8_t tx_buffer[CONFIG_MQTT_MESSAGE_BUFFER_SIZE];
static u8_t payload_buf[CONFIG_MQTT_PAYLOAD_BUFFER_SIZE];
/* The mqtt client struct */
static struct mqtt_client client;
/* MQTT Broker details. */
static struct sockaddr_storage broker;
/* Connected flag */
static bool connected;
/* File descriptor */
static struct pollfd fds;
/* GPIO callback */
static struct gpio_callback gpio_cb;
/* Button pressed flag */
bool btn_pressed;
/* change this to use another GPIO port */
#ifndef DT_ALIAS_SW0_GPIOS_CONTROLLER
#ifdef SW0_GPIO_NAME
#define DT_ALIAS_SW0_GPIOS_CONTROLLER SW0_GPIO_NAME
#else
#error SW0_GPIO_NAME or DT_ALIAS_SW0_GPIOS_CONTROLLER needs to be set in board.h
#endif
#endif
#define PORT DT_ALIAS_SW0_GPIOS_CONTROLLER
/* change this to use another GPIO pin */
#ifdef DT_ALIAS_SW0_GPIOS_PIN
#define PIN DT_ALIAS_SW0_GPIOS_PIN
#else
#error DT_ALIAS_SW0_GPIOS_PIN needs to be set in board.h
#endif
/* change to use another GPIO pin interrupt config */
#ifdef DT_ALIAS_SW0_GPIOS_FLAGS
#define EDGE (DT_ALIAS_SW0_GPIOS_FLAGS | GPIO_INT_EDGE)
#else
/*
* If DT_ALIAS_SW0_GPIOS_FLAGS not defined used default EDGE value.
* Change this to use a different interrupt trigger
*/
#define EDGE (GPIO_INT_EDGE | GPIO_INT_ACTIVE_LOW)
#endif
/* change this to enable pull-up/pull-down */
#ifndef DT_ALIAS_SW0_GPIOS_FLAGS
#ifdef DT_ALIAS_SW0_GPIOS_PIN_PUD
#define DT_ALIAS_SW0_GPIOS_FLAGS DT_ALIAS_SW0_GPIOS_PIN_PUD
#else
#define DT_ALIAS_SW0_GPIOS_FLAGS 0
#endif
#endif
#define PULL_UP DT_ALIAS_SW0_GPIOS_FLAGS
void button_pressed(struct device *gpiob, struct gpio_callback *cb,
u32_t pins)
{
btn_pressed = true;
printk("Button pressed at %" PRIu32 "\n", k_cycle_get_32());
}
#if defined(CONFIG_BSD_LIBRARY)
/**@brief Recoverable BSD library error. */
void bsd_recoverable_error_handler(uint32_t err)
{
printk("bsdlib recoverable error: %u\n", (unsigned int)err);
}
#endif /* defined(CONFIG_BSD_LIBRARY) */
#if defined(CONFIG_LWM2M_CARRIER)
K_SEM_DEFINE(carrier_registered, 0, 1);
void lwm2m_carrier_event_handler(const lwm2m_carrier_event_t *event)
{
switch (event->type) {
case LWM2M_CARRIER_EVENT_BSDLIB_INIT:
printk("LWM2M_CARRIER_EVENT_BSDLIB_INIT\n");
break;
case LWM2M_CARRIER_EVENT_CONNECT:
printk("LWM2M_CARRIER_EVENT_CONNECT\n");
break;
case LWM2M_CARRIER_EVENT_DISCONNECT:
printk("LWM2M_CARRIER_EVENT_DISCONNECT\n");
break;
case LWM2M_CARRIER_EVENT_READY:
printk("LWM2M_CARRIER_EVENT_READY\n");
k_sem_give(&carrier_registered);
break;
case LWM2M_CARRIER_EVENT_FOTA_START:
printk("LWM2M_CARRIER_EVENT_FOTA_START\n");
break;
case LWM2M_CARRIER_EVENT_REBOOT:
printk("LWM2M_CARRIER_EVENT_REBOOT\n");
break;
}
}
#endif /* defined(CONFIG_LWM2M_CARRIER) */
/**@brief Function to print strings without null-termination
*/
static void data_print(u8_t *prefix, u8_t *data, size_t len)
{
char buf[len + 1];
memcpy(buf, data, len);
buf[len] = 0;
printk("%s%s\n", prefix, buf);
}
/**@brief Function to publish data on the configured topic
*/
static int data_publish(struct mqtt_client *c, enum mqtt_qos qos,
u8_t *data, size_t len)
{
struct mqtt_publish_param param;
param.message.topic.qos = qos;
param.message.topic.topic.utf8 = CONFIG_MQTT_PUB_TOPIC;
param.message.topic.topic.size = strlen(CONFIG_MQTT_PUB_TOPIC);
param.message.payload.data = data;
param.message.payload.len = len;
param.message_id = sys_rand32_get();
param.dup_flag = 0;
param.retain_flag = 0;
data_print("Publishing: ", data, len);
printk("to topic: %s len: %u\n",
CONFIG_MQTT_PUB_TOPIC,
(unsigned int)strlen(CONFIG_MQTT_PUB_TOPIC));
return mqtt_publish(c, ¶m);
}
/**@brief Function to subscribe to the configured topic
*/
static int subscribe(void)
{
struct mqtt_topic subscribe_topic = {
.topic = {
.utf8 = CONFIG_MQTT_SUB_TOPIC,
.size = strlen(CONFIG_MQTT_SUB_TOPIC)
},
.qos = MQTT_QOS_1_AT_LEAST_ONCE
};
const struct mqtt_subscription_list subscription_list = {
.list = &subscribe_topic,
.list_count = 1,
.message_id = 1234
};
printk("Subscribing to: %s len %u\n", CONFIG_MQTT_SUB_TOPIC,
(unsigned int)strlen(CONFIG_MQTT_SUB_TOPIC));
return mqtt_subscribe(&client, &subscription_list);
}
/**@brief Function to read the published payload.
*/
static int publish_get_payload(struct mqtt_client *c, size_t length)
{
u8_t *buf = payload_buf;
u8_t *end = buf + length;
if (length > sizeof(payload_buf)) {
return -EMSGSIZE;
}
while (buf < end) {
int ret = mqtt_read_publish_payload(c, buf, end - buf);
if (ret < 0) {
int err;
if (ret != -EAGAIN) {
return ret;
}
printk("mqtt_read_publish_payload: EAGAIN\n");
err = poll(&fds, 1, K_SECONDS(CONFIG_MQTT_KEEPALIVE));
if (err > 0 && (fds.revents & POLLIN) == POLLIN) {
continue;
} else {
return -EIO;
}
}
if (ret == 0) {
return -EIO;
}
buf += ret;
}
return 0;
}
/**@brief MQTT client event handler
*/
void mqtt_evt_handler(struct mqtt_client *const c,
const struct mqtt_evt *evt)
{
int err;
switch (evt->type) {
case MQTT_EVT_CONNACK:
if (evt->result != 0) {
printk("MQTT connect failed %d\n", evt->result);
break;
}
connected = true;
printk("[%s:%d] MQTT client connected!\n", __func__, __LINE__);
subscribe();
break;
case MQTT_EVT_DISCONNECT:
printk("[%s:%d] MQTT client disconnected %d\n", __func__,
__LINE__, evt->result);
connected = false;
break;
case MQTT_EVT_PUBLISH: {
const struct mqtt_publish_param *p = &evt->param.publish;
printk("[%s:%d] MQTT PUBLISH result=%d len=%d\n", __func__,
__LINE__, evt->result, p->message.payload.len);
err = publish_get_payload(c, p->message.payload.len);
if (err >= 0) {
data_print("Received: ", payload_buf,
p->message.payload.len);
/* Echo back received data */
data_publish(&client, MQTT_QOS_1_AT_LEAST_ONCE,
payload_buf, p->message.payload.len);
} else {
printk("mqtt_read_publish_payload: Failed! %d\n", err);
printk("Disconnecting MQTT client...\n");
err = mqtt_disconnect(c);
if (err) {
printk("Could not disconnect: %d\n", err);
}
}
} break;
case MQTT_EVT_PUBACK:
if (evt->result != 0) {
printk("MQTT PUBACK error %d\n", evt->result);
break;
}
printk("[%s:%d] PUBACK packet id: %u\n", __func__, __LINE__,
evt->param.puback.message_id);
break;
case MQTT_EVT_SUBACK:
if (evt->result != 0) {
printk("MQTT SUBACK error %d\n", evt->result);
break;
}
printk("[%s:%d] SUBACK packet id: %u\n", __func__, __LINE__,
evt->param.suback.message_id);
break;
default:
printk("[%s:%d] default: %d\n", __func__, __LINE__,
evt->type);
break;
}
}
/**@brief Resolves the configured hostname and
* initializes the MQTT broker structure
*/
static void broker_init(void)
{
int err;
struct addrinfo *result;
struct addrinfo *addr;
struct addrinfo hints = {
.ai_family = AF_INET,
.ai_socktype = SOCK_STREAM
};
err = getaddrinfo(CONFIG_MQTT_BROKER_HOSTNAME, NULL, &hints, &result);
if (err) {
printk("ERROR: getaddrinfo failed %d\n", err);
return;
}
addr = result;
err = -ENOENT;
/* Look for address of the broker. */
while (addr != NULL) {
/* IPv4 Address. */
if (addr->ai_addrlen == sizeof(struct sockaddr_in)) {
struct sockaddr_in *broker4 =
((struct sockaddr_in *)&broker);
char ipv4_addr[NET_IPV4_ADDR_LEN];
broker4->sin_addr.s_addr =
((struct sockaddr_in *)addr->ai_addr)
->sin_addr.s_addr;
broker4->sin_family = AF_INET;
broker4->sin_port = htons(CONFIG_MQTT_BROKER_PORT);
inet_ntop(AF_INET, &broker4->sin_addr.s_addr,
ipv4_addr, sizeof(ipv4_addr));
printk("IPv4 Address found %s\n", ipv4_addr);
break;
} else {
printk("ai_addrlen = %u should be %u or %u\n",
(unsigned int)addr->ai_addrlen,
(unsigned int)sizeof(struct sockaddr_in),
(unsigned int)sizeof(struct sockaddr_in6));
}
addr = addr->ai_next;
break;
}
/* Free the address. */
freeaddrinfo(result);
}
/**@brief Initialize the MQTT client structure
*/
static void client_init(struct mqtt_client *client)
{
mqtt_client_init(client);
broker_init();
/* MQTT client configuration */
client->broker = &broker;
client->evt_cb = mqtt_evt_handler;
client->client_id.utf8 = (u8_t *)CONFIG_MQTT_CLIENT_ID;
client->client_id.size = strlen(CONFIG_MQTT_CLIENT_ID);
client->password = NULL;
client->user_name = NULL;
client->protocol_version = MQTT_VERSION_3_1_1;
/* MQTT buffers configuration */
client->rx_buf = rx_buffer;
client->rx_buf_size = sizeof(rx_buffer);
client->tx_buf = tx_buffer;
client->tx_buf_size = sizeof(tx_buffer);
/* MQTT transport configuration */
client->transport.type = MQTT_TRANSPORT_NON_SECURE;
}
/**@brief Initialize the file descriptor structure used by poll.
*/
static int fds_init(struct mqtt_client *c)
{
if (c->transport.type == MQTT_TRANSPORT_NON_SECURE) {
fds.fd = c->transport.tcp.sock;
} else {
#if defined(CONFIG_MQTT_LIB_TLS)
fds.fd = c->transport.tls.sock;
#else
return -ENOTSUP;
#endif
}
fds.events = POLLIN;
return 0;
}
/**@brief Configures modem to provide LTE link. Blocks until link is
* successfully established.
*/
static void modem_configure(void)
{
#if defined(CONFIG_LTE_LINK_CONTROL)
if (IS_ENABLED(CONFIG_LTE_AUTO_INIT_AND_CONNECT)) {
/* Do nothing, modem is already turned on
* and connected.
*/
} else {
#if defined(CONFIG_LWM2M_CARRIER)
/* Wait for the LWM2M_CARRIER to configure the modem and
* start the connection.
*/
printk("Waitng for carrier registration...\n");
k_sem_take(&carrier_registered, K_FOREVER);
printk("Registered!\n");
#else /* defined(CONFIG_LWM2M_CARRIER) */
int err;
printk("LTE Link Connecting ...\n");
err = lte_lc_init_and_connect();
__ASSERT(err == 0, "LTE link could not be established.");
printk("LTE Link Connected!\n");
#endif /* defined(CONFIG_LWM2M_CARRIER) */
}
#endif /* defined(CONFIG_LTE_LINK_CONTROL) */
}
void main(void)
{
struct device *gpiob;
gpiob = device_get_binding(PORT);
if (!gpiob) {
printk("error\n");
return;
}
gpio_pin_configure(gpiob, PIN,
GPIO_DIR_IN | GPIO_INT | PULL_UP | EDGE);
gpio_init_callback(&gpio_cb, button_pressed, BIT(PIN));
gpio_add_callback(gpiob, &gpio_cb);
gpio_pin_enable_callback(gpiob, PIN);
int err;
printk("The MQTT simple sample started\n");
modem_configure();
client_init(&client);
err = mqtt_connect(&client);
if (err != 0) {
printk("ERROR: mqtt_connect %d\n", err);
return;
}
err = fds_init(&client);
if (err != 0) {
printk("ERROR: fds_init %d\n", err);
return;
}
while (1) {
err = poll(&fds, 1, mqtt_keepalive_time_left(&client));
if (err < 0) {
printk("ERROR: poll %d\n", errno);
break;
}
err = mqtt_live(&client);
if ((err != 0) && (err != -EAGAIN)) {
printk("ERROR: mqtt_live %d\n", err);
break;
}
if ((fds.revents & POLLIN) == POLLIN) {
err = mqtt_input(&client);
if (err != 0) {
printk("ERROR: mqtt_input %d\n", err);
break;
}
}
if ((fds.revents & POLLERR) == POLLERR) {
printk("POLLERR\n");
break;
}
if ((fds.revents & POLLNVAL) == POLLNVAL) {
printk("POLLNVAL\n");
break;
}
if (btn_pressed) {
btn_pressed = false;
data_publish(&client, MQTT_QOS_1_AT_LEAST_ONCE, "test", strlen("test"));
}
printk("end of while\n");
}
printk("Disconnecting MQTT client...\n");
err = mqtt_disconnect(&client);
if (err) {
printk("Could not disconnect MQTT client. Error: %d\n", err);
}
}
After merging, i added the "btn_pressed" flag as i read in the forum that individual use of "data_publish" has to be at the end of the while loop, after the other mqtt functions had been called (ref: here ).
Behaviour: If i only have the data_publish function without the IF-statement, it publishes like crazy to the server. On the other hand, inside the IF-statement, it only publishes once per minute, in spite of how many times the button has been pressed. I assume that is directly connected to the keep-alive time which by default is 60 second.
What should i do?
