Merge pull request #17 from membraneframework/nif

Add NIF wrapper alongside existing C node wrapper

Author: mickel8

README.md

@@ -23,13 +23,21 @@ end
 ## Usage
 
 Basically this library works similarly to [libnice].
+
+`ExLibnice` can work both as CNode and as NIF.
+By default `CNode` implementation is used however, user can change it by passing proper option while starting `ExLibnice` (see below) or by `config.exs`:
+```elixir
+config :ex_libnice, impl: :NIF
+```
+
 Example flow can look in the following way (this is not complete i.e. runnable example).
 
 Listed functions must be invoked on both peers.
 ```elixir
 # Init ExLibnice
 {:ok, pid} =
   ExLibnice.start_link(
+    impl: NIF,
     parent: self(),
     stun_servers: [
       %{server_addr: {64, 233, 161, 127}, server_port: 19_302},

bundlex.exs

@@ -14,7 +14,7 @@ defmodule ExLibnice.BundlexProject do
         deps: [unifex: :unifex],
         pkg_configs: ["nice"],
         libs: ["pthread"],
-        interface: :cnode,
+        interface: [:nif, :cnode],
         preprocessor: Unifex
       ]
     ]

c_src/ex_libnice/native.c

@@ -33,9 +33,13 @@ UNIFEX_TERM init(UnifexEnv *env, char **stun_servers, unsigned int stun_servers_
   Refer to: https://gitlab.freedesktop.org/libnice/libnice/-/issues/120
   */
   g_object_set (G_OBJECT (state->agent), "ice-trickle", TRUE, NULL);
-  state->env = env;
+  state->env = unifex_alloc_env(env);
+  if (!unifex_self(env, &state->reply_to)) {
+    return unifex_raise(env, "failed to create native state");
+  };
   state->min_port = min_port;
   state->max_port = max_port;
+
   NiceAgent *agent = state->agent;
 
   int parse_res = parse_args(agent, stun_servers, stun_servers_length, controlling_mode);
@@ -69,13 +73,14 @@ UNIFEX_TERM init(UnifexEnv *env, char **stun_servers, unsigned int stun_servers_
     return unifex_raise(env, "failed to create main loop thread");
   }
 
-  return init_result_ok(env, state);
+  UNIFEX_TERM ret = init_result_ok(env, state);
+  unifex_release_state(env, state);
+  return ret;
 }
 
 static void *main_loop_thread_func(void *user_data) {
   GMainLoop *loop = (GMainLoop *)user_data;
   g_main_loop_run(loop);
-  g_main_loop_unref(loop);
   return NULL;
 }
 
@@ -84,7 +89,7 @@ static void cb_candidate_gathering_done(NiceAgent *agent, guint stream_id,
   UNIFEX_UNUSED(agent);
   UNIFEX_UNUSED(stream_id);
   State *state = (State *)user_data;
-  send_candidate_gathering_done(state->env, *state->env->reply_to, 0, stream_id);
+  send_candidate_gathering_done(state->env, state->reply_to, 0, stream_id);
 }
 
 static void cb_component_state_changed(NiceAgent *agent, guint stream_id,
@@ -93,9 +98,9 @@ static void cb_component_state_changed(NiceAgent *agent, guint stream_id,
   UNIFEX_UNUSED(agent);
   State *state = (State *)user_data;
   if(component_state == NICE_COMPONENT_STATE_FAILED) {
-    send_component_state_failed(state->env, *state->env->reply_to, 0, stream_id, component_id);
+    send_component_state_failed(state->env, state->reply_to, 0, stream_id, component_id);
   } else if(component_state == NICE_COMPONENT_STATE_READY) {
-    send_component_state_ready(state->env, *state->env->reply_to, 0, stream_id, component_id);
+    send_component_state_ready(state->env, state->reply_to, 0, stream_id, component_id);
   }
 }
 
@@ -104,7 +109,7 @@ static void cb_new_candidate_full(NiceAgent *agent, NiceCandidate *candidate,
   State *state = (State *)user_data;
   gchar *candidate_sdp_str =
       nice_agent_generate_local_candidate_sdp(agent, candidate);
-  send_new_candidate_full(state->env, *state->env->reply_to, 0, candidate_sdp_str);
+  send_new_candidate_full(state->env, state->reply_to, 0, candidate_sdp_str);
   g_free(candidate_sdp_str);
 }
 
@@ -119,7 +124,7 @@ static void cb_new_remote_candidate_full(NiceAgent *agent, NiceCandidate *candid
   */
   gchar *candidate_sdp_str =
       nice_agent_generate_local_candidate_sdp(agent, candidate);
-  send_new_remote_candidate_full(state->env, *state->env->reply_to, 0, candidate_sdp_str);
+  send_new_remote_candidate_full(state->env, state->reply_to, 0, candidate_sdp_str);
   g_free(candidate_sdp_str);
 }
 
@@ -128,7 +133,7 @@ static void cb_new_selected_pair(NiceAgent *agent, guint stream_id,
                                  gchar *rfoundation, gpointer user_data) {
   UNIFEX_UNUSED(agent);
   State *state = (State *)user_data;
-  send_new_selected_pair(state->env, *state->env->reply_to, 0, stream_id, component_id,
+  send_new_selected_pair(state->env, state->reply_to, 0, stream_id, component_id,
                          lfoundation, rfoundation);
 }
 
@@ -138,27 +143,27 @@ static void cb_recv(NiceAgent *_agent, guint stream_id, guint component_id,
   State *state = (State *)user_data;
   UnifexPayload *payload = unifex_payload_alloc(state->env, UNIFEX_PAYLOAD_BINARY, len);
   memcpy(payload->data, buf, len);
-  send_ice_payload(state->env, *state->env->reply_to, 0, stream_id, component_id, payload);
+  send_ice_payload(state->env, state->reply_to, 0, stream_id, component_id, payload);
   unifex_payload_release(payload);
 }
 
 UNIFEX_TERM add_stream(UnifexEnv *env, UnifexState *state,
                        unsigned int n_components, char *name) {
   guint stream_id = nice_agent_add_stream(state->agent, n_components);
   if (stream_id == 0) {
-    return add_stream_result_error_failed_to_add_stream(env);
+    return add_stream_result_error_failed_to_add_stream(env, state);
   }
 
   if(!attach_recv(state, stream_id, n_components)) {
-    return add_stream_result_error_failed_to_attach_recv(env);
+    return add_stream_result_error_failed_to_attach_recv(env, state);
   }
 
   // set name if one was specified
   if (strcmp(name, "") == 0) {
-    return add_stream_result_ok(env, stream_id);
+    return add_stream_result_ok(env, stream_id, state);
   } else {
     if (!nice_agent_set_stream_name(state->agent, stream_id, name)) {
-      return add_stream_result_error_invalid_stream_or_duplicate_name(env);
+      return add_stream_result_error_invalid_stream_or_duplicate_name(env, state);
     }
   }
 
@@ -167,7 +172,7 @@ UNIFEX_TERM add_stream(UnifexEnv *env, UnifexState *state,
     nice_agent_set_port_range(state->agent, stream_id, i, state->min_port, state->max_port);
   }
 
-  return add_stream_result_ok(env, stream_id);
+  return add_stream_result_ok(env, stream_id, state);
 }
 
 static gboolean attach_recv(UnifexState *state, guint stream_id, guint n_components) {
@@ -182,7 +187,7 @@ static gboolean attach_recv(UnifexState *state, guint stream_id, guint n_compone
 
 UNIFEX_TERM remove_stream(UnifexEnv *env, UnifexState *state, unsigned int stream_id) {
   nice_agent_remove_stream(state->agent, stream_id);
-  return remove_stream_result_ok(env);
+  return remove_stream_result_ok(env, state);
 }
 
 UNIFEX_TERM set_relay_info(UnifexEnv *env, UnifexState *state, unsigned int stream_id,
@@ -196,57 +201,57 @@ UNIFEX_TERM set_relay_info(UnifexEnv *env, UnifexState *state, unsigned int stre
   } else if(strcmp(relay_type, "tls") == 0) {
     nice_relay_type = NICE_RELAY_TYPE_TURN_TLS;
   } else {
-    return set_relay_info_result_error_bad_relay_type(env);
+    return set_relay_info_result_error_bad_relay_type(env, state);
   }
 
   if(!nice_agent_set_relay_info(state->agent, stream_id, component_id, server_ip, server_port,
                                 username, password, nice_relay_type)) {
-    return set_relay_info_result_error_failed_to_set_turn(env);
+    return set_relay_info_result_error_failed_to_set_turn(env, state);
   }
-  return set_relay_info_result_ok(env);
+  return set_relay_info_result_ok(env, state);
 }
 
 UNIFEX_TERM forget_relays(UnifexEnv *env, UnifexState *state, unsigned int stream_id,
                           unsigned int component_id) {
   if(!nice_agent_forget_relays(state->agent, stream_id, component_id)) {
-    return forget_relays_result_error_component_not_found(env);
+    return forget_relays_result_error_component_not_found(env, state);
   }
-  return forget_relays_result_ok(env);
+  return forget_relays_result_ok(env, state);
 }
 
 UNIFEX_TERM generate_local_sdp(UnifexEnv *env, UnifexState *state) {
   gchar *local_sdp = nice_agent_generate_local_sdp(state->agent);
-  return generate_local_sdp_result_ok(env, local_sdp);
+  return generate_local_sdp_result_ok(env, local_sdp, state);
 }
 
 UNIFEX_TERM parse_remote_sdp(UnifexEnv *env, UnifexState *state, char *remote_sdp) {
   int cand_added_num = nice_agent_parse_remote_sdp(state->agent, remote_sdp);
 	if (cand_added_num < 0) {
-    return parse_remote_sdp_result_error_failed_to_parse_sdp(env);
+    return parse_remote_sdp_result_error_failed_to_parse_sdp(env, state);
   }
-  return parse_remote_sdp_result_ok(env, cand_added_num);
+  return parse_remote_sdp_result_ok(env, cand_added_num, state);
 }
 
 UNIFEX_TERM gather_candidates(UnifexEnv *env, State *state, unsigned int stream_id) {
   g_networking_init();
   if(!nice_agent_gather_candidates(state->agent, stream_id)) {
-    return gather_candidates_result_error_invalid_stream_or_allocation(env);
+    return gather_candidates_result_error_invalid_stream_or_allocation(env, state);
   }
-  return gather_candidates_result_ok(env);
+  return gather_candidates_result_ok(env, state);
 }
 
 UNIFEX_TERM peer_candidate_gathering_done(UnifexEnv *env, State *state, unsigned int stream_id) {
   if(!nice_agent_peer_candidate_gathering_done(state->agent, stream_id)) {
-    return peer_candidate_gathering_done_result_error_stream_not_found(env);
+    return peer_candidate_gathering_done_result_error_stream_not_found(env, state);
   }
-  return peer_candidate_gathering_done_result_ok(env);
+  return peer_candidate_gathering_done_result_ok(env, state);
 }
 
 UNIFEX_TERM get_local_credentials(UnifexEnv *env, State *state, unsigned int stream_id) {
   gchar *ufrag = NULL;
   gchar *pwd = NULL;
   if (!nice_agent_get_local_credentials(state->agent, stream_id, &ufrag, &pwd)) {
-    return get_local_credentials_result_error_failed_to_get_credentials(env);
+    return get_local_credentials_result_error_failed_to_get_credentials(env, state);
   }
   const size_t lenufrag = strlen(ufrag);
   const size_t lenpwd = strlen(pwd);
@@ -256,7 +261,7 @@ UNIFEX_TERM get_local_credentials(UnifexEnv *env, State *state, unsigned int str
   memcpy(credentials + lenufrag + 1, pwd, lenpwd + 1);
   g_free(ufrag);
   g_free(pwd);
-  UNIFEX_TERM ret = get_local_credentials_result_ok(env, credentials); 
+  UNIFEX_TERM ret = get_local_credentials_result_ok(env, credentials, state);
   unifex_free(credentials);
   return ret;
 }
@@ -267,51 +272,50 @@ UNIFEX_TERM set_remote_credentials(UnifexEnv *env, State *state,
   char *pwd = NULL;
   parse_credentials(credentials, &ufrag, &pwd);
   if (!nice_agent_set_remote_credentials(state->agent, stream_id, ufrag, pwd)) {
-    return set_remote_credentials_result_error_failed_to_set_credentials(env);
+    return set_remote_credentials_result_error_failed_to_set_credentials(env, state);
   }
-  return set_remote_credentials_result_ok(env);
+  return set_remote_credentials_result_ok(env, state);
 }
 
 UNIFEX_TERM set_remote_candidate(UnifexEnv *env, State *state,
                                   char *candidate, unsigned int stream_id, unsigned int component_id) {
   NiceCandidate *cand = nice_agent_parse_remote_candidate_sdp(state->agent, stream_id, candidate);
   if (cand == NULL) {
-    return set_remote_candidate_result_error_failed_to_parse_sdp_string(env);
+    return set_remote_candidate_result_error_failed_to_parse_sdp_string(env, state);
   }
   GSList *cands = NULL;
   cands = g_slist_append(cands, cand);
   if (nice_agent_set_remote_candidates(state->agent, stream_id, component_id, cands) < 0) {
-    return set_remote_candidate_result_error_failed_to_set(env);
+    return set_remote_candidate_result_error_failed_to_set(env, state);
   }
-  return set_remote_candidate_result_ok(env);
+  return set_remote_candidate_result_ok(env, state);
 }
 
 UNIFEX_TERM restart(UnifexEnv *env, State *state) {
   if(nice_agent_restart(state->agent)) {
-    return restart_result_ok(env);
+    return restart_result_ok(env, state);
   }
-  return restart_result_error_failed_to_restart(env);
+  return restart_result_error_failed_to_restart(env, state);
 }
 
 UNIFEX_TERM restart_stream(UnifexEnv *env, State *state, unsigned int stream_id) {
   if(nice_agent_restart_stream(state->agent, stream_id)) {
-    return restart_stream_result_ok(env);
+    return restart_stream_result_ok(env, state);
   }
-  return restart_stream_result_error_failed_to_restart(env);
+  return restart_stream_result_error_failed_to_restart(env, state);
 }
 
 UNIFEX_TERM send_payload(UnifexEnv *env, State *state, unsigned int stream_id,
                          unsigned int component_id, UnifexPayload *payload) {
   if(nice_agent_send(state->agent, stream_id, component_id, payload->size, (char *)payload->data) < 0) {
-    return send_payload_result_error_failed_to_send(env);
+    return send_payload_result_error_failed_to_send(env, state);
   }
-  return send_payload_result_ok(env);
+  return send_payload_result_ok(env, state);
 }
 
 void handle_destroy_state(UnifexEnv *env, State *state) {
   UNIFEX_UNUSED(env);
   g_main_loop_quit(state->gloop);
-  pthread_kill(state->gloop_tid, SIGKILL);
   if (state->gloop) {
     g_main_loop_unref(state->gloop);
     state->gloop = NULL;
@@ -320,4 +324,7 @@ void handle_destroy_state(UnifexEnv *env, State *state) {
     g_object_unref(state->agent);
     state->agent = NULL;
   }
+  if (state->env) {
+    unifex_free_env(state->env);
+  }
 }

c_src/ex_libnice/native.h

@@ -13,6 +13,7 @@ struct State {
   pthread_t gloop_tid;
   unsigned int min_port;
   unsigned int max_port;
+  UnifexPid reply_to;
 };
 
 #include "_generated/native.h"