#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include <lwmqtt/posix.h>

#define COMMAND_TIMEOUT 5000

lwmqtt_posix_network_t network = {0};

lwmqtt_posix_timer_t timer1, timer2;

lwmqtt_client_t client;

pthread_mutex_t mutex;

static void message_arrived(lwmqtt_client_t *_client, void *ref, lwmqtt_string_t topic, lwmqtt_message_t msg) {
  printf("message_arrived: %.*s => %.*s (%d)\n", (int)topic.len, topic.data, (int)msg.payload_len, (char *)msg.payload,
         (int)msg.payload_len);
}

static void *thread(void *_) {
  // loop forever
  for (;;) {
    // block until data is available
    bool available = false;
    lwmqtt_err_t err = lwmqtt_posix_network_select(&network, &available, COMMAND_TIMEOUT);
    if (err != LWMQTT_SUCCESS) {
      printf("failed lwmqtt_posix_network_select: %d\n", err);
      exit(1);
    }

    // lock mutex
    pthread_mutex_lock(&mutex);

    // process incoming data if available
    if (available) {
      // get available bytes
      size_t available_bytes = 0;
      err = lwmqtt_posix_network_peek(&network, &available_bytes);
      if (err != LWMQTT_SUCCESS) {
        printf("failed lwmqtt_posix_network_peek: %d\n", err);
        exit(1);
      }

      // yield client only if there is still data to read since select might unblock because of incoming ack packets
      // that are already handled until we get to this point
      if (available_bytes > 0) {
        err = lwmqtt_yield(&client, available_bytes, COMMAND_TIMEOUT);
        if (err != LWMQTT_SUCCESS) {
          printf("failed lwmqtt_yield: %d\n", err);
          exit(1);
        }
      }
    }

    // keep connection alive
    err = lwmqtt_keep_alive(&client, COMMAND_TIMEOUT);
    if (err != LWMQTT_SUCCESS) {
      printf("failed lwmqtt_keep_alive: %d\n", err);
      exit(1);
    }

    // unlock mutex
    pthread_mutex_unlock(&mutex);
  }
}

int main(void) {
  // initialize client
  lwmqtt_init(&client, malloc(512), 512, malloc(512), 512);

  // configure client
  lwmqtt_set_network(&client, &network, lwmqtt_posix_network_read, lwmqtt_posix_network_write);
  lwmqtt_set_timers(&client, &timer1, &timer2, lwmqtt_posix_timer_set, lwmqtt_posix_timer_get);
  lwmqtt_set_callback(&client, NULL, message_arrived);

  // connect to broker
  lwmqtt_err_t err = lwmqtt_posix_network_connect(&network, "public.cloud.shiftr.io", 1883);
  if (err != LWMQTT_SUCCESS) {
    printf("failed lwmqtt_posix_network_connect: %d\n", err);
    exit(1);
  }

  // set options
  lwmqtt_connect_options_t options = lwmqtt_default_connect_options;
  options.client_id = lwmqtt_string("lwmqtt");
  options.clean_session = false;
  options.username = lwmqtt_string("public");
  options.password = lwmqtt_string("public");
  options.keep_alive = 10;

  // send connect packet
  err = lwmqtt_connect(&client, &options, NULL, COMMAND_TIMEOUT);
  if (err != LWMQTT_SUCCESS) {
    printf("failed lwmqtt_connect: %d (%d)\n", err, options.return_code);
    exit(1);
  }

  // log
  printf("connected! (session present: %d)\n", options.session_present);

  // subscribe to topic
  err = lwmqtt_subscribe_one(&client, lwmqtt_string("hello"), LWMQTT_QOS0, COMMAND_TIMEOUT);
  if (err != LWMQTT_SUCCESS) {
    printf("failed lwmqtt_subscribe: %d\n", err);
    exit(1);
  }

  // initialize mutex
  int r = pthread_mutex_init(&mutex, NULL);
  if (r < 0) {
    printf("failed pthread_mutex_init: %d\n", r);
    exit(0);
  }

  // run background thread
  pthread_t t;
  r = pthread_create(&t, NULL, thread, NULL);
  if (r < 0) {
    printf("failed pthread_create: %d\n", r);
    exit(1);
  }

  // loop forever
  for (;;) {
    // sleep 1s
    usleep(1000 * 1000);

    // prepare message
    lwmqtt_message_t msg = {.qos = LWMQTT_QOS0, .retained = false, .payload = (uint8_t *)("world"), .payload_len = 5};

    // lock mutex
    pthread_mutex_lock(&mutex);

    // publish message
    err = lwmqtt_publish(&client, NULL, lwmqtt_string("hello"), msg, COMMAND_TIMEOUT);
    if (err != LWMQTT_SUCCESS) {
      printf("failed lwmqtt_publish: %d\n", err);
      exit(1);
    }

    // unlock mutex
    pthread_mutex_unlock(&mutex);
  }
}
