CAMLprim value ocaml_gstreamer_buffer_of_data(value _ba, value _off, value _len)
{
  CAMLparam1(_ba);
  int bufoff = Int_val(_off);
  int buflen = Int_val(_len);
  GstBuffer *gstbuf;
  GstMapInfo map;
  gboolean bret;

  assert(buflen+bufoff <= Caml_ba_array_val(_ba)->dim[0]);

  caml_release_runtime_system();
  gstbuf = gst_buffer_new_and_alloc(buflen);
  bret = gst_buffer_map(gstbuf, &map, GST_MAP_WRITE);
  caml_acquire_runtime_system();

  if(!bret) caml_raise_constant(*caml_named_value("gstreamer_exn_failure"));
  memcpy(map.data, (unsigned char*)Caml_ba_data_val(_ba)+bufoff, buflen);

  caml_release_runtime_system();
  gst_buffer_unmap(gstbuf, &map);
  caml_acquire_runtime_system();

  CAMLreturn(value_of_buffer(gstbuf));
}

CAMLprim value ocaml_gstreamer_appsrc_push_buffer_data(value _as, value _buf)
{
  CAMLparam2(_as, _buf);
  int buflen = Caml_ba_array_val(_buf)->dim[0];
  appsrc *as = Appsrc_val(_as);
  GstBuffer *gstbuf;
  GstMapInfo map;
  GstFlowReturn ret;
  gboolean bret;

  caml_release_runtime_system();
  gstbuf = gst_buffer_new_and_alloc(buflen);
  bret = gst_buffer_map(gstbuf, &map, GST_MAP_WRITE);
  caml_acquire_runtime_system();

  if(!bret) caml_raise_constant(*caml_named_value("gstreamer_exn_failure"));
  memcpy(map.data, (unsigned char*)Caml_ba_data_val(_buf), buflen);

  caml_release_runtime_system();
  gst_buffer_unmap(gstbuf, &map);
  ret = gst_app_src_push_buffer(as->appsrc, gstbuf);
  caml_acquire_runtime_system();

  if (ret != GST_FLOW_OK) caml_raise_constant(*caml_named_value("gstreamer_exn_failure"));
  CAMLreturn(Val_unit);
}

CAMLprim value ocaml_gstreamer_appsink_pull_buffer(value _as, value string_mode)
{
  CAMLparam1(_as);
  CAMLlocal1(ans);
  appsink *as = Appsink_val(_as);
  GstSample *gstsample;
  GstBuffer *gstbuf;
  GstMapInfo map;
  intnat len;
  gboolean ret;

  caml_release_runtime_system();
  gstsample = gst_app_sink_pull_sample(as->appsink);
  caml_acquire_runtime_system();

  if (!gstsample)
    {
      if (gst_app_sink_is_eos(as->appsink))
        caml_raise_constant(*caml_named_value("gstreamer_exn_eos"));
      else
        caml_raise_constant(*caml_named_value("gstreamer_exn_failure"));
    }

  caml_release_runtime_system();
  gstbuf = gst_sample_get_buffer(gstsample);
  caml_acquire_runtime_system();

  if (!gstbuf) caml_raise_constant(*caml_named_value("gstreamer_exn_failure"));

  caml_release_runtime_system();
  ret = gst_buffer_map(gstbuf, &map, GST_MAP_READ);
  caml_acquire_runtime_system();

  if (!ret) caml_raise_constant(*caml_named_value("gstreamer_exn_failure"));

  len = map.size;
  if (string_mode == Val_false) {
    ans = caml_ba_alloc(CAML_BA_C_LAYOUT | CAML_BA_UINT8, 1, NULL, &len);
    memcpy(Caml_ba_data_val(ans), map.data, len);
  } else {
    ans = caml_alloc_string(len);
    memcpy(String_val(ans), map.data, len);
  }

  caml_release_runtime_system();
  gst_buffer_unmap(gstbuf, &map);
  gst_sample_unref(gstsample);
  caml_acquire_runtime_system();

  CAMLreturn(ans);
}

CAMLprim value stub_sem_wait(value sem) {
  CAMLparam1(sem);
  CAMLlocal2(result, perrno);
  int rc, lerrno;
  sem_t *s;

  s = *Sem_val(sem);
  if (NULL == s) {
    lerrno = EINVAL;
    goto ERROR;
  }

  caml_release_runtime_system();
  rc = sem_wait(s);
  lerrno = errno;
  caml_acquire_runtime_system();

  if (0 != rc) {
    goto ERROR;
  }

  result = caml_alloc(1, 0); // Result.Ok
  Store_field(result, 0, Val_unit);
  goto END;

ERROR:
  perrno = caml_alloc(2, 0);
  Store_field(perrno, 0, eunix); // `EUnix
  Store_field(perrno, 1, unix_error_of_code(lerrno));
  result = caml_alloc(1, 1); // Result.Error
  Store_field(result, 0, perrno);

END:
  CAMLreturn(result);
}

ssize_t unix_unistd_readlink(const char *path, char *buf, size_t bufsiz) {
  ssize_t retval;
  caml_release_runtime_system();
  retval = readlink(path, buf, bufsiz);
  caml_acquire_runtime_system();
  return retval;
}

int unix_unistd_close(int fd) {
  int retval;
  caml_release_runtime_system();
  retval = close(fd);
  caml_acquire_runtime_system();
  return retval;
}

int unix_unistd_seteuid(uid_t uid) {
  int retval;
  caml_release_runtime_system();
  retval = seteuid(uid);
  caml_acquire_runtime_system();
  return retval;
}

int unix_unistd_chown(const char *path, uid_t owner, gid_t group) {
  int retval;
  caml_release_runtime_system();
  retval = chown(path, owner, group);
  caml_acquire_runtime_system();
  return retval;
}

int unix_unistd_truncate(const char *path, off_t length) {
  int retval;
  caml_release_runtime_system();
  retval = truncate(path, length);
  caml_acquire_runtime_system();
  return retval;
}

CAMLprim value
stub_ba_send(value fd, value val_buf, value val_ofs, value val_len)
{
  CAMLparam4(fd, val_buf, val_ofs, val_len);
  int ret = 0;
#ifdef WIN32
  char *data = (char*)Caml_ba_data_val(val_buf) + Long_val(val_ofs);
  size_t c_len = Int_val(val_len);
  SOCKET s = Socket_val(fd);
  DWORD err = 0;

  caml_release_runtime_system();
  ret = send(s, data, c_len, 0);
  if (ret == SOCKET_ERROR) err = WSAGetLastError();
  caml_acquire_runtime_system();

  if (err) {
    win32_maperr(err);
    uerror("read", Nothing);
  }
#else
  caml_failwith("AF_HYPERV only available on Windows");
#endif
  CAMLreturn(Val_int(ret));
}

CAMLprim value ocaml_gstreamer_bus_pop_filtered(value _bus, value _filter)
{
  CAMLparam2(_bus, _filter);
  CAMLlocal1(ans);
  GstBus *bus = Bus_val(_bus);
  GstMessageType filter = 0;
  GstMessage *msg;
  int i;

  for(i = 0; i < Wosize_val(_filter); i++)
    filter |= message_type_of_int(Int_val(Field(_filter, i)));

  caml_release_runtime_system();
  msg = gst_bus_pop_filtered(bus, filter);
  caml_acquire_runtime_system();

  if(!msg)
    ans = Val_int(0);
  else
    {
      ans = caml_alloc_tuple(1);
      Store_field(ans, 0, value_of_message(msg));
    }

  CAMLreturn(ans);
}

CAMLprim value ocaml_gstreamer_init(value _argv)
{
  CAMLparam1(_argv);
  char **argv = NULL;
  int argc = 0;
  int len, i;

  if (Is_block(_argv))
    {
      _argv = Field(_argv, 0);
      argc = Wosize_val(_argv);
      argv = malloc(argc*sizeof(char*));
      for(i = 0; i < argc; i++)
        {
          len = caml_string_length(Field(_argv,i));
          argv[i] = malloc(len+1);
          memcpy(argv[i], String_val(Field(_argv,i)), len+1);
        }
    }

  caml_release_runtime_system();
  gst_init(&argc, &argv);
  for(i = 0; i < argc; i++)
    free(argv[i]);
  free(argv);
  caml_acquire_runtime_system();

  CAMLreturn(Val_unit);
}

int unix_fcntl_open_none(const char *path, int oflag) {
    int retval;
    caml_release_runtime_system();
    retval = open(path, oflag);
    caml_acquire_runtime_system();
    return retval;
}

int unix_fcntl_open_perms(const char *path, int oflag, mode_t perms) {
    int retval;
    caml_release_runtime_system();
    retval = open(path, oflag, perms);
    caml_acquire_runtime_system();
    return retval;
}

CAMLprim value recv_stub(value socket, value rcv_option) {
    CAMLparam2 (socket, rcv_option);
    CAMLlocal1 (message);

    void *sock = Socket_val(socket)->wrapped;

    zmq_msg_t request;
    int result = zmq_msg_init (&request);
    stub_raise_if (result == -1);

    caml_release_runtime_system();
    result = zmq_recvmsg(sock, &request, Int_val(rcv_option));
    caml_acquire_runtime_system();

    stub_raise_if (result == -1);

    size_t size = zmq_msg_size (&request);
    if (size == 0) {
        message = EMPTY_STRING;
    } else {
        message = caml_alloc_string(size);
        memcpy (String_val(message), zmq_msg_data (&request), size);
    }
    result = zmq_msg_close(&request);
    stub_raise_if (result == -1);
    CAMLreturn (message);
}

CAMLprim value stub_asl_new_msg() {
  CAMLparam0();
  caml_release_runtime_system();
  aslmsg msg = asl_new(ASL_TYPE_MSG);
  caml_acquire_runtime_system();
  CAMLreturn(alloc_message(msg));
}

CAMLprim value stub_launch_activate_socket(value name) {
  CAMLparam1(name);
  CAMLlocal1(result);
  const char *c_name = caml_strdup(String_val(name));
  int *listening_fds = NULL;
  size_t n_listening_fds = 0;
  int err;

  caml_release_runtime_system();
  err = launch_activate_socket(c_name, &listening_fds, &n_listening_fds);
  caml_acquire_runtime_system();

  caml_stat_free((void*)c_name);

  switch (err) {
    case 0:
      result = caml_alloc_tuple(n_listening_fds);
      for (int i = 0; i < n_listening_fds; i++) {
        Store_field(result, i, Val_int(*(listening_fds + i)));
      }
      break;
    default:
      unix_error(err, "launch_activate_socket", name);
      break;
  }
  CAMLreturn(result);
}

CAMLprim value stub_vsock_accept(value sock){
  CAMLparam1(sock);
  CAMLlocal1(result);
  int lsock = Int_val(sock);
  int csock = -1;
#ifdef AF_VSOCK
  struct sockaddr_vm sac;
  socklen_t socklen = sizeof(sac);

  caml_release_runtime_system();
  csock = accept(lsock, (struct sockaddr *)&sac, &socklen);
  caml_acquire_runtime_system();
  if (csock == -1) {
    uerror("accept", Nothing);
  }

  result = caml_alloc_tuple(3);
  Store_field(result, 0, Val_int(csock));
  Store_field(result, 1, Val_int(sac.svm_cid));
  Store_field(result, 2, Val_int(sac.svm_port));
#else
  caml_failwith("AF_VSOCK not available");
#endif
  CAMLreturn(result);
}

int unix_unistd_symlink(const char *target, const char *linkpath) {
  int retval;
  caml_release_runtime_system();
  retval = symlink(target, linkpath);
  caml_acquire_runtime_system();
  return retval;
}

CAMLprim value caml_zmq_poll(value poll, value timeout) {
    CAMLparam2 (poll, timeout);
    CAMLlocal2 (events, some);
    int n = CAML_ZMQ_Poll_val(poll)->num_elems;
    zmq_pollitem_t *items = CAML_ZMQ_Poll_val(poll)->poll_items;
    int tm = Int_val(timeout);

    caml_release_runtime_system();
    int num_event_sockets = zmq_poll(items, n, tm);
    caml_acquire_runtime_system();

    caml_zmq_raise_if(num_event_sockets == -1);
    events = caml_alloc(n, 0);

    int i;
    for(i = 0; i < n; i++) {
        if (!((items[i].revents & ZMQ_POLLIN) || (items[i].revents & ZMQ_POLLOUT))) {
          Store_field(events, i, Val_int(0)); /* None */
        } else {
          some = caml_alloc(1, 0);
          Store_field(some, 0, CAML_ZMQ_Val_mask(items[i].revents));
          Store_field(events, i, some);
        }
    }

    CAMLreturn (events);
}