ei_encode_switch_event_headers(ebuf, event);
}
+/* function to make rpc call to remote node to retrieve a pid -
+ calls module:function(Ref). The response comes back as
+ {rex, {Ref, Pid}}
+ */
+int ei_pid_from_rpc(struct ei_cnode_s *ec, int sockfd, erlang_ref *ref, char *module, char *function)
+{
+ ei_x_buff buf;
+ ei_x_new(&buf);
+ ei_x_encode_list_header(&buf, 1);
+ ei_init_ref(ec, ref);
+ ei_x_encode_ref(&buf, ref);
+ ei_x_encode_empty_list(&buf);
+
+ ei_rpc_to(ec, sockfd, module, function, buf.buff, buf.index);
+ ei_x_free(&buf);
+
+ return 0;
+}
/* function to spawn a process on a remote node */
int ei_spawn(struct ei_cnode_s *ec, int sockfd, erlang_ref *ref, char *module, char *function, int argc, char **argv)
static char *MARKER = "1";
+static switch_status_t handle_ref_tuple(listener_t *listener, erlang_msg *msg, ei_x_buff *buf, ei_x_buff *rbuf);
+
static void *SWITCH_THREAD_FUNC api_exec(switch_thread_t *thread, void *obj)
{
switch_bool_t r = SWITCH_TRUE;
return SWITCH_STATUS_SUCCESS;
}
-/* {handlecall,<uuid>,<handler process registered name>} */
-static switch_status_t handle_msg_handlecall(listener_t *listener, int arity, ei_x_buff *buf, ei_x_buff *rbuf)
+/* {handlecall,<uuid>,<handler process registered name>}
+ or
+ {handlecall,<uuid>} to send messages back to the sender
+ */
+static switch_status_t handle_msg_handlecall(listener_t *listener, erlang_msg *msg, int arity, ei_x_buff *buf, ei_x_buff *rbuf)
{
char reg_name[MAXATOMLEN];
char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1];
- if (arity != 3 ||
- ei_decode_string_or_binary(buf->buff, &buf->index, SWITCH_UUID_FORMATTED_LENGTH, uuid_str) ||
- ei_decode_atom(buf->buff, &buf->index, reg_name)) {
+ if (arity < 2 || arity > 3 ||
+ (arity==3 && ei_decode_atom(buf->buff, &buf->index, reg_name)) ||
+ ei_decode_string_or_binary(buf->buff, &buf->index, SWITCH_UUID_FORMATTED_LENGTH, uuid_str)) {
ei_x_encode_tuple_header(rbuf, 2);
ei_x_encode_atom(rbuf, "error");
ei_x_encode_atom(rbuf, "badarg");
switch_core_session_t *session;
if (!switch_strlen_zero(uuid_str) && (session = switch_core_session_locate(uuid_str))) {
/* create a new session list element and attach it to this listener */
- if (attach_call_to_registered_process(listener, reg_name, session)) {
+ if ((arity==2 && attach_call_to_pid(listener, &msg->from, session)) ||
+ (arity==3 && attach_call_to_registered_process(listener, reg_name, session))) {
ei_x_encode_atom(rbuf, "ok");
} else {
ei_x_encode_tuple_header(rbuf, 2);
return SWITCH_STATUS_SUCCESS;
}
+/* catch the response to ei_rpc_to (which comes back as {rex, {Ref, Pid}}
+ The {Ref,Pid} bit can be handled by handle_ref_tuple
+ */
+static switch_status_t handle_msg_rpcresponse(listener_t *listener, erlang_msg *msg, int arity, ei_x_buff *buf, ei_x_buff *rbuf)
+{
+ int type, size, arity2, tmpindex;
+
+ ei_get_type(buf->buff, &buf->index, &type, &size);
+ switch(type) {
+ case ERL_SMALL_TUPLE_EXT :
+ case ERL_LARGE_TUPLE_EXT :
+ tmpindex = buf->index;
+ ei_decode_tuple_header(buf->buff, &tmpindex, &arity2);
+ return handle_ref_tuple(listener,msg,buf,rbuf);
+ default:
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Unknown rpc response\n");
+ break;
+ }
+ /* no reply */
+ return SWITCH_STATUS_FALSE;
+}
+
static switch_status_t handle_msg_tuple(listener_t *listener, erlang_msg *msg, ei_x_buff *buf, ei_x_buff *rbuf)
{
char tupletag[MAXATOMLEN];
} else if (!strncmp(tupletag, "bind", MAXATOMLEN)) {
ret = handle_msg_bind(listener,msg,buf,rbuf);
} else if (!strncmp(tupletag, "handlecall", MAXATOMLEN)) {
- ret = handle_msg_handlecall(listener,arity,buf,rbuf);
+ ret = handle_msg_handlecall(listener,msg,arity,buf,rbuf);
+ } else if (!strncmp(tupletag, "rex", MAXATOMLEN)) {
+ ret = handle_msg_rpcresponse(listener,msg,arity,buf,rbuf);
} else {
ei_x_encode_tuple_header(rbuf, 2);
ei_x_encode_atom(rbuf, "error");
switch_core_hash_insert(listener->spawn_pid_hash, hash, pid);
}
- return SWITCH_STATUS_SUCCESS;
+ /* no reply */
+ return SWITCH_STATUS_FALSE;
}
l = lp;
lp = lp->next;
- /* test all of the sessions attached to this event in case
- one of them should receive it as well
+ /* test all of the sessions attached to this listener in case
+ one of them should receive the event as well
*/
send_event_to_attached_sessions(l,event);
ei_x_buff rbuf;
ei_x_new_with_version(&rbuf);
- switch_mutex_lock(listener->sock_mutex);
+ /* do we need the mutex when reading? */
+ /*switch_mutex_lock(listener->sock_mutex);*/
status = ei_xreceive_msg_tmo(listener->sockfd, &msg, &buf, 100);
- switch_mutex_unlock(listener->sock_mutex);
+ /*switch_mutex_unlock(listener->sock_mutex);*/
switch(status) {
case ERL_TICK :
ei_x_encode_atom(&rbuf, "error");
ei_x_encode_atom(&rbuf, "acldeny");
+ switch_mutex_lock(listener->sock_mutex);
ei_send(listener->sockfd, &msg.from, rbuf.buff, rbuf.index);
+ switch_mutex_unlock(listener->sock_mutex);
#ifdef EI_DEBUG
ei_x_print_msg(&rbuf, &msg.from, 1);
#endif
return session_element;
}
+session_elem_t* attach_call_to_pid(listener_t* listener, erlang_pid* pid, switch_core_session_t *session)
+{
+ /* create a session list element */
+ session_elem_t* session_element = NULL;
+ if (!(session_element = switch_core_alloc(switch_core_session_get_pool(session), sizeof(*session_element)))) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to allocate session element\n");
+ }
+ else {
+ if (SWITCH_STATUS_SUCCESS != switch_core_session_read_lock(session)) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to get session read lock\n");
+ }
+ else {
+ session_element->session = session;
+ session_element->process.type = ERLANG_PID;
+ memcpy(&session_element->process.pid, pid, sizeof(erlang_pid));
+ switch_set_flag(session_element, LFLAG_SESSION_ALIVE);
+ switch_clear_flag(session_element, LFLAG_OUTBOUND_INIT);
+ switch_queue_create(&session_element->event_queue, SWITCH_CORE_QUEUE_LEN, switch_core_session_get_pool(session));
+ switch_mutex_init(&session_element->flag_mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
+ /* attach the session to the listener */
+ add_session_elem_to_listener(listener,session_element);
+
+ ei_link(listener, ei_self(listener->ec), pid);
+ }
+ }
+ return session_element;
+}
session_elem_t* attach_call_to_spawned_process(listener_t* listener, char *module, char *function, switch_core_session_t *session)
{
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to get session read lock\n");
}
else {
- char *argv[1], hash[100];
+ char hash[100];
int i = 0;
session_element->session = session;
erlang_pid *pid;
ei_x_encode_atom(&rbuf, "new_pid");
ei_x_encode_ref(&rbuf, &ref);
ei_x_encode_pid(&rbuf, ei_self(listener->ec));
+ /* should lock with mutex? */
ei_reg_send(listener->ec, listener->sockfd, module, rbuf.buff, rbuf.index);
#ifdef EI_DEBUG
ei_x_print_reg_msg(&rbuf, module, 1);
#endif
ei_x_free(&rbuf);
} else {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "rpc call: %s:%s(Ref)\n", module, function);
+ /* should lock with mutex? */
+ ei_pid_from_rpc(listener->ec, listener->sockfd, &ref, module, function);
+ /*
+ char *argv[1];
ei_spawn(listener->ec, listener->sockfd, &ref, module, function, 0, argv);
+ */
}
ei_hash_ref(&ref, hash);
switch_clear_flag(session_element, LFLAG_OUTBOUND_INIT);
switch_clear_flag(session_element, LFLAG_WAITING_FOR_PID);
- ei_link(listener, ei_self(listener->ec), pid);
+ /* this hangs because it can never get hold of the socket mutex */
+ ei_link(listener, ei_self(listener->ec), pid);
}
}
return session_element;
void ei_link(listener_t *listener, erlang_pid *from, erlang_pid *to);
void ei_encode_switch_event_headers(ei_x_buff *ebuf, switch_event_t *event);
void ei_encode_switch_event_tag(ei_x_buff *ebuf, switch_event_t *event, char *tag);
+int ei_pid_from_rpc(struct ei_cnode_s *ec, int sockfd, erlang_ref *ref, char *module, char *function);
int ei_spawn(struct ei_cnode_s *ec, int sockfd, erlang_ref *ref, char *module, char *function, int argc, char **argv);
void ei_init_ref(struct ei_cnode_s *ec, erlang_ref *ref);
void ei_x_print_reg_msg(ei_x_buff *buf, char *dest, int send);
/* mod_erlang_event.c */
session_elem_t* attach_call_to_registered_process(listener_t* listener, char* reg_name, switch_core_session_t *session);
+session_elem_t* attach_call_to_pid(listener_t* listener, erlang_pid* pid, switch_core_session_t *session);
session_elem_t* attach_call_to_spawned_process(listener_t* listener, char *module, char *function, switch_core_session_t *session);
/* For Emacs:
projects / thirdparty / freeswitch.git / commitdiff
