/* Callback when we get an message from the MQTT broker.  */
static size_t on_publish(struct mqtt_client_t *client_p, const char *topic_p, void *chin_p, size_t size)
/* Callback when we get an message from the MQTT broker.  */
static size_t on_publish(struct mqtt_client_t *client_p, const char *topic_p, void *chin_p, size_t size)
{
    uint8_t buf[16];
    int unread, toread;
    std_printf(OSTR("on_publish: %s = "), topic_p);
    for (unread = size; unread > 0; unread -= toread) {
        toread = MIN(unread, sizeof(buf) - 1);
        chan_read(chin_p, buf, toread);
        buf[toread] = '\0';
        std_printf(OSTR("%s"), &buf[0]);
    }
    std_printf(OSTR("\r\n"));
    message_count++;
    return (0);
}