// AMQP
// ====

#include "amqp_common.h"

static Term amqp_consume_pack(Env e, IoWork* w) {
  LapineChannel* chan = (LapineChannel*)(intptr_t)w->hand;
  if (w->code != 0) {
    free(w->data);
    return io_tup(e, (Term)w->hand, io_fail(e, w->code, w->text));
  }
  Term body = io_str(e, w->data, w->size);
  Term tag = (Term)(w->made);
  free(w->data);
  Term pair = io_tup(e, body, tag);
  return io_tup(e, (Term)w->hand, io_done(e, pair));
}

static Term amqp_consume_more(Env e, IoWork* w) {
  LapineChannel* chan = (LapineChannel*)(intptr_t)w->hand;
  amqp_envelope_t envelope;
  amqp_maybe_release_bytes(chan->conn, &envelope.message);
  struct timeval timeout = {0, 0};
  amqp_rpc_reply_t reply = amqp_consume_message(chan->conn, &envelope, &timeout, 0);
  if (reply.reply_type == AMQP_RESPONSE_NORMAL) {
    w->size = envelope.message.body.len;
    w->data = io_mem(malloc(w->size + 1));
    memcpy(w->data, envelope.message.body.bytes, w->size);
    w->data[w->size] = '\0';
    w->made = (intptr_t)envelope.delivery_tag;
    amqp_destroy_envelope(&envelope);
    w->code = 0;
    return amqp_consume_pack(e, w);
  }
  if (reply.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION
    && reply.library_error == AMQP_STATUS_TIMEOUT) {
    amqp_socket_t* sock = amqp_get_socket(chan->conn);
    int fd = amqp_socket_get_sockfd(sock);
    return io_wait_on(w, fd, POLLIN, amqp_consume_more);
  }
  w->code = lapine_rpc_error(reply);
  w->text = "lapine: consume failed";
  return amqp_consume_pack(e, w);
}

Term amqp_consume_run(Env e, Term* f, IoWork* w) {
  LapineChannel* chan = (LapineChannel*)(intptr_t)f[0];
  if (chan == NULL || chan->conn == NULL) {
    return io_fail(e, EINVAL, "lapine: invalid channel");
  }
  uint64_t qn = 0;
  char* queue = io_cstr(e, f[1], &qn);
  if (!chan->consumer_ready) {
    amqp_bytes_t que = amqp_cstring_bytes(queue);
    amqp_rpc_reply_t reply = amqp_basic_consume(
      chan->conn,
      chan->channel,
      que,
      amqp_empty_bytes,
      0,
      1,
      0,
      amqp_empty_table);
    u32 err = lapine_rpc_error(reply);
    if (err != 0) {
      free(queue);
      return io_tup(e, f[0], io_fail(e, err, "lapine: basic.consume failed"));
    }
    chan->consumer_ready = 1;
    strncpy(chan->queue, queue, sizeof(chan->queue) - 1);
    chan->queue[sizeof(chan->queue) - 1] = '\0';
  }
  free(queue);
  if (lapine_socket_nonblock(chan->conn) < 0) {
    return io_tup(e, f[0], io_fail(e, (u32)errno, "lapine: socket setup failed"));
  }
  w->hand = (intptr_t)chan;
  w->code = 0;
  w->data = NULL;
  return amqp_consume_more(e, w);
}

static void __attribute__((constructor)) amqp_consume_use(void) {
  io_eff(CID_AMQP_CONSUME, amqp_consume_run, IO_READ);
}