// AMQP
// ====

#include "amqp_common.h"

Term amqp_publish_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 en = 0;
  uint64_t kn = 0;
  uint64_t bn = 0;
  char* exchange = io_cstr(e, f[1], &en);
  char* key = io_cstr(e, f[2], &kn);
  char* body = io_cstr(e, f[3], &bn);
  amqp_bytes_t payload;
  payload.len = bn;
  payload.bytes = body;
  int ok = amqp_basic_publish(
    chan->conn,
    chan->channel,
    amqp_cstring_bytes(exchange),
    amqp_cstring_bytes(key),
    0,
    0,
    NULL,
    payload);
  free(exchange);
  free(key);
  free(body);
  if (!ok) {
    return io_fail(e, (u32)errno, "lapine: publish failed");
  }
  return io_tup(e, f[0], io_done(e, term_pak(CID_UNIT, 0)));
}

static void __attribute__((constructor)) amqp_publish_use(void) {
  io_eff(CID_AMQP_PUBLISH, amqp_publish_run, 0);
}