Migrate PeerStatus events to stasis, add stasis endpoints, and add chan_pjsip device...
authorJoshua Colp <jcolp@digium.com>
Sat, 22 Jun 2013 12:40:16 +0000 (12:40 +0000)
committerJoshua Colp <jcolp@digium.com>
Sat, 22 Jun 2013 12:40:16 +0000 (12:40 +0000)
(closes issue ASTERISK-21489)
(closes issue ASTERISK-21503)

Review: https://reviewboard.asterisk.org/r/2601/

git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@392538 65c4cc65-6c06-0410-ace0-fbb531ad65f3

12 files changed:
channels/chan_gulp.c
channels/chan_iax2.c
channels/chan_sip.c
channels/chan_skinny.c
include/asterisk/manager.h
include/asterisk/res_sip.h
include/asterisk/stasis_endpoints.h
main/manager.c
main/manager_endpoints.c [new file with mode: 0644]
main/stasis_endpoints.c
res/res_sip.c
res/res_sip/sip_configuration.c

index 967349e..9e939a0 100644 (file)
@@ -53,6 +53,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/musiconhold.h"
 #include "asterisk/causes.h"
 #include "asterisk/taskprocessor.h"
+#include "asterisk/stasis_endpoints.h"
+#include "asterisk/stasis_channels.h"
 
 #include "asterisk/res_sip.h"
 #include "asterisk/res_sip_session.h"
@@ -82,6 +84,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 static const char desc[] = "Gulp SIP Channel";
 static const char channel_type[] = "Gulp";
 
+static unsigned int chan_idx;
+
 /*!
  * \brief Positions of various media
  */
@@ -125,6 +129,7 @@ static struct ast_frame *gulp_read(struct ast_channel *ast);
 static int gulp_write(struct ast_channel *ast, struct ast_frame *f);
 static int gulp_indicate(struct ast_channel *ast, int condition, const void *data, size_t datalen);
 static int gulp_fixup(struct ast_channel *oldchan, struct ast_channel *newchan);
+static int gulp_devicestate(const char *data);
 
 /*! \brief PBX interface structure for channel registration */
 static struct ast_channel_tech gulp_tech = {
@@ -143,6 +148,7 @@ static struct ast_channel_tech gulp_tech = {
        .exception = gulp_read,
        .indicate = gulp_indicate,
        .fixup = gulp_fixup,
+       .devicestate = gulp_devicestate,
        .properties = AST_CHAN_TP_WANTSJITTER | AST_CHAN_TP_CREATESJITTER
 };
 
@@ -422,8 +428,8 @@ static struct ast_channel *gulp_new(struct ast_sip_session *session, int state,
                return NULL;
        }
 
-       if (!(chan = ast_channel_alloc(1, state, S_OR(session->id.number.str, ""), S_OR(session->id.name.str, ""), "", "", "", linkedid, 0, "Gulp/%s-%.*s", ast_sorcery_object_get_id(session->endpoint),
-               (int)session->inv_session->dlg->call_id->id.slen, session->inv_session->dlg->call_id->id.ptr))) {
+       if (!(chan = ast_channel_alloc(1, state, S_OR(session->id.number.str, ""), S_OR(session->id.name.str, ""), "", "", "", linkedid, 0, "Gulp/%s-%08x", ast_sorcery_object_get_id(session->endpoint),
+               ast_atomic_fetchadd_int((int *)&chan_idx, +1)))) {
                ao2_cleanup(pvt);
                return NULL;
        }
@@ -461,6 +467,8 @@ static struct ast_channel *gulp_new(struct ast_sip_session *session, int state,
        ast_channel_exten_set(chan, S_OR(exten, "s"));
        ast_channel_priority_set(chan, 1);
 
+       ast_endpoint_add_channel(session->endpoint->persistent, chan);
+
        return chan;
 }
 
@@ -623,6 +631,68 @@ static int gulp_fixup(struct ast_channel *oldchan, struct ast_channel *newchan)
        return 0;
 }
 
+/*! \brief Function called to get the device state of an endpoint */
+static int gulp_devicestate(const char *data)
+{
+       RAII_VAR(struct ast_sip_endpoint *, endpoint, ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "endpoint", data), ao2_cleanup);
+       enum ast_device_state state = AST_DEVICE_UNKNOWN;
+       RAII_VAR(struct ast_endpoint_snapshot *, endpoint_snapshot, NULL, ao2_cleanup);
+       RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
+       struct ast_devstate_aggregate aggregate;
+       int num, inuse = 0;
+
+       if (!endpoint) {
+               return AST_DEVICE_INVALID;
+       }
+
+       endpoint_snapshot = ast_endpoint_latest_snapshot(ast_endpoint_get_tech(endpoint->persistent),
+               ast_endpoint_get_resource(endpoint->persistent), 1);
+
+       if (endpoint_snapshot->state == AST_ENDPOINT_OFFLINE) {
+               state = AST_DEVICE_UNAVAILABLE;
+       } else if (endpoint_snapshot->state == AST_ENDPOINT_ONLINE) {
+               state = AST_DEVICE_NOT_INUSE;
+       }
+
+       if (!endpoint_snapshot->num_channels || !(caching_topic = ast_channel_topic_all_cached())) {
+               return state;
+       }
+
+       ast_devstate_aggregate_init(&aggregate);
+
+       ao2_ref(caching_topic, +1);
+
+       for (num = 0; num < endpoint_snapshot->num_channels; num++) {
+               RAII_VAR(struct stasis_message *, msg, stasis_cache_get_extended(caching_topic, ast_channel_snapshot_type(),
+                       endpoint_snapshot->channel_ids[num], 1), ao2_cleanup);
+               struct ast_channel_snapshot *snapshot;
+
+               if (!msg) {
+                       continue;
+               }
+
+               snapshot = stasis_message_data(msg);
+
+               if (snapshot->state == AST_STATE_DOWN) {
+                       ast_devstate_aggregate_add(&aggregate, AST_DEVICE_NOT_INUSE);
+               } else if (snapshot->state == AST_STATE_RINGING) {
+                       ast_devstate_aggregate_add(&aggregate, AST_DEVICE_RINGING);
+               } else if ((snapshot->state == AST_STATE_UP) || (snapshot->state == AST_STATE_RING) ||
+                       (snapshot->state == AST_STATE_BUSY)) {
+                       ast_devstate_aggregate_add(&aggregate, AST_DEVICE_INUSE);
+                       inuse++;
+               }
+       }
+
+       if (endpoint->devicestate_busy_at && (inuse == endpoint->devicestate_busy_at)) {
+               state = AST_DEVICE_BUSY;
+       } else if (ast_devstate_aggregate_result(&aggregate) != AST_DEVICE_INVALID) {
+               state = ast_devstate_aggregate_result(&aggregate);
+       }
+
+       return state;
+}
+
 struct indicate_data {
        struct ast_sip_session *session;
        int condition;
@@ -731,6 +801,7 @@ static int gulp_indicate(struct ast_channel *ast, int condition, const void *dat
                } else {
                        res = -1;
                }
+               ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "Gulp/%s", ast_sorcery_object_get_id(session->endpoint));
                break;
        case AST_CONTROL_BUSY:
                if (ast_channel_state(ast) != AST_STATE_UP) {
@@ -1326,7 +1397,7 @@ static int gulp_incoming_request(struct ast_sip_session *session, struct pjsip_r
                return 0;
        }
 
-       if (!(session->channel = gulp_new(session, AST_STATE_DOWN, session->exten, NULL, NULL, NULL))) {
+       if (!(session->channel = gulp_new(session, AST_STATE_RING, session->exten, NULL, NULL, NULL))) {
                if (pjsip_inv_end_session(session->inv_session, 503, NULL, &packet) == PJ_SUCCESS) {
                        ast_sip_session_send_response(session, packet);
                }
@@ -1335,7 +1406,6 @@ static int gulp_incoming_request(struct ast_sip_session *session, struct pjsip_r
                return -1;
        }
 
-       ast_setstate(session->channel, AST_STATE_RING);
        res = ast_pbx_start(session->channel);
 
        switch (res) {
index 486af52..0f5c8fc 100644 (file)
@@ -101,6 +101,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/data.h"
 #include "asterisk/netsock2.h"
 #include "asterisk/security_events.h"
+#include "asterisk/stasis_endpoints.h"
 #include "asterisk/bridging.h"
 
 #include "iax2/include/iax2.h"
@@ -552,6 +553,8 @@ struct iax2_peer {
 
        struct ast_acl_list *acl;
        enum calltoken_peer_enum calltoken_required;    /*!< Is calltoken validation required or not, can be YES, NO, or AUTO */
+
+       struct ast_endpoint *endpoint; /*!< Endpoint structure for this peer */
 };
 
 #define IAX2_TRUNK_PREFACE (sizeof(struct iax_frame) + sizeof(struct ast_iax2_meta_hdr) + sizeof(struct ast_iax2_meta_trunk_hdr))
@@ -8563,6 +8566,7 @@ static void unlink_peer(struct iax2_peer *peer)
 static void __expire_registry(const void *data)
 {
        struct iax2_peer *peer = (struct iax2_peer *) data;
+       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 
        if (!peer)
                return;
@@ -8576,7 +8580,11 @@ static void __expire_registry(const void *data)
        ast_debug(1, "Expiring registration for peer '%s'\n", peer->name);
        if (ast_test_flag64((&globalflags), IAX_RTUPDATE) && (ast_test_flag64(peer, IAX_TEMPONLY|IAX_RTCACHEFRIENDS)))
                realtime_update_peer(peer->name, &peer->addr, 0);
-       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: IAX2\r\nPeer: IAX2/%s\r\nPeerStatus: Unregistered\r\nCause: Expired\r\n", peer->name);
+       ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_OFFLINE);
+       blob = ast_json_pack("{s: s, s: s}",
+               "peer_status", "Unregistered",
+               "cause", "Expired");
+       ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
        /* modify entry in peercnts table as _not_ registered */
        peercnt_modify(0, 0, &peer->addr);
        /* Reset the address */
@@ -8701,6 +8709,8 @@ static int update_registry(struct sockaddr_in *sin, int callno, char *devtype, i
        }
 
        if (ast_sockaddr_cmp(&p->addr, &sockaddr)) {
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                if (iax2_regfunk) {
                        iax2_regfunk(p->name, 1);
                }
@@ -8716,17 +8726,26 @@ static int update_registry(struct sockaddr_in *sin, int callno, char *devtype, i
                        ast_db_put("IAX/Registry", p->name, data);
                        ast_verb(3, "Registered IAX2 '%s' (%s) at %s:%d\n", p->name,
                                            ast_test_flag(&iaxs[callno]->state, IAX_STATE_AUTHENTICATED) ? "AUTHENTICATED" : "UNAUTHENTICATED", ast_inet_ntoa(sin->sin_addr), ntohs(sin->sin_port));
-                       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: IAX2\r\nPeer: IAX2/%s\r\nPeerStatus: Registered\r\nAddress: %s\r\nPort: %d\r\n", p->name, ast_inet_ntoa(sin->sin_addr), ntohs(sin->sin_port));
+                       ast_endpoint_set_state(p->endpoint, AST_ENDPOINT_ONLINE);
+                       blob = ast_json_pack("{s: s, s: s, s: i}",
+                               "peer_status", "Registered",
+                               "address", ast_inet_ntoa(sin->sin_addr),
+                               "port", ntohs(sin->sin_port));
                        register_peer_exten(p, 1);
                        ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "IAX2/%s", p->name); /* Activate notification */
                } else if (!ast_test_flag64(p, IAX_TEMPONLY)) {
                        ast_verb(3, "Unregistered IAX2 '%s' (%s)\n", p->name,
                                            ast_test_flag(&iaxs[callno]->state, IAX_STATE_AUTHENTICATED) ? "AUTHENTICATED" : "UNAUTHENTICATED");
-                       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: IAX2\r\nPeer: IAX2/%s\r\nPeerStatus: Unregistered\r\n", p->name);
+                       ast_endpoint_set_state(p->endpoint, AST_ENDPOINT_OFFLINE);
+                       blob = ast_json_pack("{s: s}",
+                               "peer_status", "Unregistered");
                        register_peer_exten(p, 0);
                        ast_db_del("IAX/Registry", p->name);
                        ast_devstate_changed(AST_DEVICE_UNAVAILABLE, AST_DEVSTATE_CACHABLE, "IAX2/%s", p->name); /* Activate notification */
                }
+
+               ast_endpoint_blob_publish(p->endpoint, ast_endpoint_state_type(), blob);
+
                /* Update the host */
                /* Verify that the host is really there */
                iax2_poke_peer(p, callno);
@@ -10759,20 +10778,28 @@ static int socket_process_helper(struct iax2_thread *thread)
                                log_jitterstats(fr->callno);
 
                                if (iaxs[fr->callno]->peerpoke) {
+                                       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
                                        peer = iaxs[fr->callno]->peerpoke;
                                        if ((peer->lastms < 0)  || (peer->historicms > peer->maxms)) {
                                                if (iaxs[fr->callno]->pingtime <= peer->maxms) {
                                                        ast_log(LOG_NOTICE, "Peer '%s' is now REACHABLE! Time: %d\n", peer->name, iaxs[fr->callno]->pingtime);
-                                                       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: IAX2\r\nPeer: IAX2/%s\r\nPeerStatus: Reachable\r\nTime: %d\r\n", peer->name, iaxs[fr->callno]->pingtime); 
+                                                       ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_ONLINE);
+                                                       blob = ast_json_pack("{s: s, s: i}",
+                                                               "peer_status", "Reachable",
+                                                               "time", iaxs[fr->callno]->pingtime);
                                                        ast_devstate_changed(AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE, "IAX2/%s", peer->name); /* Activate notification */
                                                }
                                        } else if ((peer->historicms > 0) && (peer->historicms <= peer->maxms)) {
                                                if (iaxs[fr->callno]->pingtime > peer->maxms) {
                                                        ast_log(LOG_NOTICE, "Peer '%s' is now TOO LAGGED (%d ms)!\n", peer->name, iaxs[fr->callno]->pingtime);
-                                                       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: IAX2\r\nPeer: IAX2/%s\r\nPeerStatus: Lagged\r\nTime: %d\r\n", peer->name, iaxs[fr->callno]->pingtime); 
+                                                       ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_ONLINE);
+                                                       blob = ast_json_pack("{s: s, s: i}",
+                                                               "peer_status", "Lagged",
+                                                               "time", iaxs[fr->callno]->pingtime);
                                                        ast_devstate_changed(AST_DEVICE_UNAVAILABLE, AST_DEVSTATE_CACHABLE, "IAX2/%s", peer->name); /* Activate notification */
                                                }
                                        }
+                                       ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
                                        peer->lastms = iaxs[fr->callno]->pingtime;
                                        if (peer->smoothing && (peer->lastms > -1))
                                                peer->historicms = (iaxs[fr->callno]->pingtime + peer->historicms) / 2;
@@ -11886,8 +11913,14 @@ static void __iax2_poke_noanswer(const void *data)
        int callno;
 
        if (peer->lastms > -1) {
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                ast_log(LOG_NOTICE, "Peer '%s' is now UNREACHABLE! Time: %d\n", peer->name, peer->lastms);
-               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: IAX2\r\nPeer: IAX2/%s\r\nPeerStatus: Unreachable\r\nTime: %d\r\n", peer->name, peer->lastms);
+               ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_OFFLINE);
+               blob = ast_json_pack("{s: s, s: i}",
+                       "peer_status", "Unreachable",
+                       "time", peer->lastms);
+               ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
                ast_devstate_changed(AST_DEVICE_UNAVAILABLE, AST_DEVSTATE_CACHABLE, "IAX2/%s", peer->name); /* Activate notification */
        }
        if ((callno = peer->callno) > 0) {
@@ -12307,6 +12340,8 @@ static void peer_destructor(void *obj)
        peer->mwi_event_sub = stasis_unsubscribe(peer->mwi_event_sub);
 
        ast_string_field_free_memory(peer);
+
+       ast_endpoint_shutdown(peer->endpoint);
 }
 
 /*! \brief Create peer structure based on configuration */
@@ -12340,6 +12375,9 @@ static struct iax2_peer *build_peer(const char *name, struct ast_variable *v, st
                peer->addr.len = sizeof(struct sockaddr_in);
                if (ast_string_field_init(peer, 32))
                        peer = peer_unref(peer);
+               if (!(peer->endpoint = ast_endpoint_create("IAX2", name))) {
+                       peer = peer_unref(peer);
+               }
        }
 
        if (peer) {
index fbd7f1c..c207e24 100644 (file)
@@ -15724,6 +15724,7 @@ static void set_socket_transport(struct sip_socket *socket, int transport)
 static int expire_register(const void *data)
 {
        struct sip_peer *peer = (struct sip_peer *)data;
+       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 
        if (!peer) {            /* Hmmm. We have no peer. Weird. */
                return 0;
@@ -15743,7 +15744,11 @@ static int expire_register(const void *data)
                peer->socket.ws_session = NULL;
        }
 
-       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: SIP\r\nPeer: SIP/%s\r\nPeerStatus: Unregistered\r\nCause: Expired\r\n", peer->name);
+       ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_OFFLINE);
+       blob = ast_json_pack("{s: s, s: s}",
+               "peer_status", "Unregistered",
+               "cause", "Expired");
+       ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
        register_peer_exten(peer, FALSE);       /* Remove regexten */
        ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "SIP/%s", peer->name);
 
@@ -15988,6 +15993,7 @@ static enum parse_register_result parse_register_contact(struct sip_pvt *pvt, st
        int start = 0;
        int wildcard_found = 0;
        int single_binding_found = 0;
+       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 
        ast_copy_string(contact, __get_header(req, "Contact", &start), sizeof(contact));
 
@@ -16174,7 +16180,12 @@ static enum parse_register_result parse_register_contact(struct sip_pvt *pvt, st
                }
                ast_db_put("SIP/Registry", peer->name, data);
        }
-       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: SIP\r\nPeer: SIP/%s\r\nPeerStatus: Registered\r\nAddress: %s\r\n", peer->name,  ast_sockaddr_stringify(&peer->addr));
+
+       ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_ONLINE);
+       blob = ast_json_pack("{s: s, s: s}",
+               "peer_status", "Registered",
+               "address", ast_sockaddr_stringify(&peer->addr));
+       ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
 
        /* Is this a new IP address for us? */
        if (ast_sockaddr_cmp(&peer->addr, &oldsin)) {
@@ -17178,6 +17189,7 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                /* Create peer if we have autocreate mode enabled */
                peer = temp_peer(name);
                if (peer) {
+                       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
                        ao2_t_link(peers, peer, "link peer into peer table");
                        if (!ast_sockaddr_isnull(&peer->addr)) {
                                ao2_t_link(peers_by_ip, peer, "link peer into peers-by-ip table");
@@ -17206,7 +17218,11 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                                ast_string_field_set(p, fullcontact, peer->fullcontact);
                                /* Say OK and ask subsystem to retransmit msg counter */
                                transmit_response_with_date(p, "200 OK", req);
-                               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: SIP\r\nPeer: SIP/%s\r\nPeerStatus: Registered\r\nAddress: %s\r\n", peer->name, ast_sockaddr_stringify(addr));
+                               ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_ONLINE);
+                               blob = ast_json_pack("{s: s, s: s}",
+                                       "peer_status", "Registered",
+                                       "address", ast_sockaddr_stringify(addr));
+                               ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
                                send_mwi = 1;
                                res = 0;
                                break;
@@ -17225,6 +17241,8 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "SIP/%s", peer->name);
        }
        if (res < 0) {
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                switch (res) {
                case AUTH_SECRET_FAILED:
                        /* Wrong password in authentication. Go away, don't try again until you fixed it */
@@ -17232,14 +17250,12 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                        if (global_authfailureevents) {
                                const char *peer_addr = ast_strdupa(ast_sockaddr_stringify_addr(addr));
                                const char *peer_port = ast_strdupa(ast_sockaddr_stringify_port(addr));
-                               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus",
-                                             "ChannelType: SIP\r\n"
-                                             "Peer: SIP/%s\r\n"
-                                             "PeerStatus: Rejected\r\n"
-                                             "Cause: AUTH_SECRET_FAILED\r\n"
-                                             "Address: %s\r\n"
-                                             "Port: %s\r\n",
-                                             name, peer_addr, peer_port);
+
+                               blob = ast_json_pack("{s: s, s: s, s: s, s: s}",
+                                       "peer_status", "Rejected",
+                                       "cause", "AUTH_SECRET_FAILED",
+                                       "address", peer_addr,
+                                       "port", peer_port);
                        }
                        break;
                case AUTH_USERNAME_MISMATCH:
@@ -17255,16 +17271,12 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                                if (global_authfailureevents) {
                                        const char *peer_addr = ast_strdupa(ast_sockaddr_stringify_addr(addr));
                                        const char *peer_port = ast_strdupa(ast_sockaddr_stringify_port(addr));
-                                       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus",
-                                                     "ChannelType: SIP\r\n"
-                                                     "Peer: SIP/%s\r\n"
-                                                     "PeerStatus: Rejected\r\n"
-                                                     "Cause: %s\r\n"
-                                                     "Address: %s\r\n"
-                                                     "Port: %s\r\n",
-                                                     name,
-                                                     res == AUTH_PEER_NOT_DYNAMIC ? "AUTH_PEER_NOT_DYNAMIC" : "URI_NOT_FOUND",
-                                                     peer_addr, peer_port);
+
+                                       blob = ast_json_pack("{s: s, s: s, s: s, s: s}",
+                                               "peer_status", "Rejected",
+                                               "cause", res == AUTH_PEER_NOT_DYNAMIC ? "AUTH_PEER_NOT_DYNAMIC" : "URI_NOT_FOUND",
+                                               "address", peer_addr,
+                                               "port", peer_port);
                                }
                        } else {
                                /* URI not found */
@@ -17273,30 +17285,24 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                                        if (global_authfailureevents) {
                                                const char *peer_addr = ast_strdupa(ast_sockaddr_stringify_addr(addr));
                                                const char *peer_port = ast_strdupa(ast_sockaddr_stringify_port(addr));
-                                               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus",
-                                                       "ChannelType: SIP\r\n"
-                                                       "Peer: SIP/%s\r\n"
-                                                       "PeerStatus: Rejected\r\n"
-                                                       "Cause: AUTH_PEER_NOT_DYNAMIC\r\n"
-                                                       "Address: %s\r\n"
-                                                       "Port: %s\r\n",
-                                                       name, peer_addr, peer_port);
+
+                                               blob = ast_json_pack("{s: s, s: s, s: s, s: s}",
+                                                       "peer_status", "Rejected",
+                                                       "cause", "AUTH_PEER_NOT_DYNAMIC",
+                                                       "address", peer_addr,
+                                                       "port", peer_port);
                                        }
                                } else {
                                        transmit_response(p, "404 Not found", &p->initreq);
                                        if (global_authfailureevents) {
                                                const char *peer_addr = ast_strdupa(ast_sockaddr_stringify_addr(addr));
                                                const char *peer_port = ast_strdupa(ast_sockaddr_stringify_port(addr));
-                                               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus",
-                                                       "ChannelType: SIP\r\n"
-                                                       "Peer: SIP/%s\r\n"
-                                                       "PeerStatus: Rejected\r\n"
-                                                       "Cause: %s\r\n"
-                                                       "Address: %s\r\n"
-                                                       "Port: %s\r\n",
-                                                       name,
-                                                       (res == AUTH_USERNAME_MISMATCH) ? "AUTH_USERNAME_MISMATCH" : "URI_NOT_FOUND",
-                                                       peer_addr, peer_port);
+
+                                               blob = ast_json_pack("{s: s, s: s, s: s, s: s}",
+                                                       "peer_status", "Rejected",
+                                                       "cause", (res == AUTH_USERNAME_MISMATCH) ? "AUTH_USERNAME_MISMATCH" : "URI_NOT_FOUND",
+                                                       "address", peer_addr,
+                                                       "port", peer_port);
                                        }
                                }
                        }
@@ -17305,6 +17311,8 @@ static enum check_auth_result register_verify(struct sip_pvt *p, struct ast_sock
                default:
                        break;
                }
+
+               ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
        }
        if (peer) {
                sip_unref_peer(peer, "register_verify: sip_unref_peer: tossing stack peer pointer at end of func");
@@ -23804,6 +23812,8 @@ static void handle_response_peerpoke(struct sip_pvt *p, int resp, struct sip_req
        if (statechanged) {
                const char *s = is_reachable ? "Reachable" : "Lagged";
                char str_lastms[20];
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                snprintf(str_lastms, sizeof(str_lastms), "%d", pingtime);
 
                ast_log(LOG_NOTICE, "Peer '%s' is now %s. (%dms / %dms)\n",
@@ -23812,9 +23822,11 @@ static void handle_response_peerpoke(struct sip_pvt *p, int resp, struct sip_req
                if (sip_cfg.peer_rtupdate) {
                        ast_update_realtime(ast_check_realtime("sipregs") ? "sipregs" : "sippeers", "name", peer->name, "lastms", str_lastms, SENTINEL);
                }
-               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus",
-                       "ChannelType: SIP\r\nPeer: SIP/%s\r\nPeerStatus: %s\r\nTime: %d\r\n",
-                       peer->name, s, pingtime);
+               ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_ONLINE);
+               blob = ast_json_pack("{s: s, s: i}",
+                       "peer_status", s,
+                       "time", pingtime);
+               ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
                if (is_reachable && sip_cfg.regextenonqualify)
                        register_peer_exten(peer, TRUE);
        }
@@ -29094,11 +29106,17 @@ static int sip_poke_noanswer(const void *data)
        peer->pokeexpire = -1;
 
        if (peer->lastms > -1) {
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                ast_log(LOG_NOTICE, "Peer '%s' is now UNREACHABLE!  Last qualify: %d\n", peer->name, peer->lastms);
                if (sip_cfg.peer_rtupdate) {
                        ast_update_realtime(ast_check_realtime("sipregs") ? "sipregs" : "sippeers", "name", peer->name, "lastms", "-1", SENTINEL);
                }
-               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: SIP\r\nPeer: SIP/%s\r\nPeerStatus: Unreachable\r\nTime: %d\r\n", peer->name, -1);
+               ast_endpoint_set_state(peer->endpoint, AST_ENDPOINT_OFFLINE);
+               blob = ast_json_pack("{s: s, s: s}",
+                       "peer_status", "Unreachable",
+                       "time", "-1");
+               ast_endpoint_blob_publish(peer->endpoint, ast_endpoint_state_type(), blob);
                if (sip_cfg.regextenonqualify) {
                        register_peer_exten(peer, FALSE);
                }
index ad51edf..9bdc0f2 100644 (file)
@@ -80,6 +80,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/event.h"
 #include "asterisk/indications.h"
 #include "asterisk/linkedlists.h"
+#include "asterisk/stasis_endpoints.h"
 #include "asterisk/bridging.h"
 
 /*** DOCUMENTATION
@@ -1585,6 +1586,7 @@ struct skinny_device {
        struct skinny_line *activeline;
        struct ast_format_cap *cap;
        struct ast_format_cap *confcap;
+       struct ast_endpoint *endpoint;
        AST_LIST_HEAD(, skinny_line) lines;
        AST_LIST_HEAD(, skinny_speeddial) speeddials;
        AST_LIST_HEAD(, skinny_serviceurl) serviceurls;
@@ -1687,7 +1689,7 @@ static struct skinny_line *skinny_line_destroy(struct skinny_line *l)
        ast_free(l);
        return NULL;
 }
-static struct skinny_device *skinny_device_alloc(void)
+static struct skinny_device *skinny_device_alloc(const char *dname)
 {
        struct skinny_device *d;
        if (!(d = ast_calloc(1, sizeof(*d)))) {
@@ -1696,18 +1698,23 @@ static struct skinny_device *skinny_device_alloc(void)
 
        d->cap = ast_format_cap_alloc_nolock();
        d->confcap = ast_format_cap_alloc_nolock();
-       if (!d->cap || !d->confcap) {
+       d->endpoint = ast_endpoint_create("Skinny", dname);
+       if (!d->cap || !d->confcap || !d->endpoint) {
                d->cap = ast_format_cap_destroy(d->cap);
                d->confcap = ast_format_cap_destroy(d->confcap);
                ast_free(d);
                return NULL;
        }
+
+       ast_copy_string(d->name, dname, sizeof(d->name));
+
        return d;
 }
 static struct skinny_device *skinny_device_destroy(struct skinny_device *d)
 {
        d->cap = ast_format_cap_destroy(d->cap);
        d->confcap = ast_format_cap_destroy(d->confcap);
+       ast_endpoint_shutdown(d->endpoint);
        ast_free(d);
        return NULL;
 }
@@ -2244,6 +2251,8 @@ static int skinny_register(struct skinny_req *req, struct skinnysession *s)
                ast_sockaddr_from_sin(&addr, &s->sin);
                if (!d->session && !strcasecmp(req->data.reg.name, d->id)
                                && ast_apply_ha(d->ha, &addr)) {
+                       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                        s->device = d;
                        d->type = letohl(req->data.reg.type);
                        d->protocolversion = letohl(req->data.reg.protocolVersion);
@@ -2277,7 +2286,6 @@ static int skinny_register(struct skinny_req *req, struct skinnysession *s)
                                l->instance = instance;
                                l->newmsgs = ast_app_has_voicemail(l->mailbox, NULL);
                                set_callforwards(l, NULL, SKINNY_CFWD_ALL|SKINNY_CFWD_BUSY|SKINNY_CFWD_NOANSWER);
-                               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: Skinny\r\nPeer: Skinny/%s@%s\r\nPeerStatus: Registered\r\n", l->name, d->name);
                                register_exten(l);
                                /* initialize MWI on line and device */
                                mwi_event_cb(l, NULL, NULL, NULL);
@@ -2287,6 +2295,9 @@ static int skinny_register(struct skinny_req *req, struct skinnysession *s)
                                ast_devstate_changed(AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE, "Skinny/%s", l->name);
                                --instance;
                        }
+                       ast_endpoint_set_state(d->endpoint, AST_ENDPOINT_ONLINE);
+                       blob = ast_json_pack("{s: s}", "peer_status", "Registered");
+                       ast_endpoint_blob_publish(d->endpoint, ast_endpoint_state_type(), blob);
                        break;
                }
        }
@@ -2306,6 +2317,7 @@ static int skinny_unregister(struct skinny_req *req, struct skinnysession *s)
        d = s->device;
 
        if (d) {
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
                d->session = NULL;
 
                AST_LIST_TRAVERSE(&d->speeddials, sd, list) {
@@ -2317,11 +2329,14 @@ static int skinny_unregister(struct skinny_req *req, struct skinnysession *s)
                                ast_format_cap_remove_all(l->cap);
                                ast_parse_allow_disallow(&l->prefs, l->cap, "all", 0);
                                l->instance = 0;
-                               manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: Skinny\r\nPeer: Skinny/%s@%s\r\nPeerStatus: Unregistered\r\n", l->name, d->name);
                                unregister_exten(l);
                                ast_devstate_changed(AST_DEVICE_UNAVAILABLE, AST_DEVSTATE_CACHABLE, "Skinny/%s", l->name);
                        }
                }
+
+               ast_endpoint_set_state(d->endpoint, AST_ENDPOINT_OFFLINE);
+               blob = ast_json_pack("{s: s}", "peer_status", "Unregistered");
+               ast_endpoint_blob_publish(d->endpoint, ast_endpoint_state_type(), blob);
        }
 
        return -1; /* main loop will destroy the session */
@@ -8220,7 +8235,7 @@ static struct skinny_device *config_device(const char *dname, struct ast_variabl
                }
        }
 
-       if (!(d = skinny_device_alloc())) {
+       if (!(d = skinny_device_alloc(dname))) {
                ast_verb(1, "Unable to allocate memory for device %s.\n", dname);
                AST_LIST_UNLOCK(&devices);
                return NULL;
@@ -8613,6 +8628,8 @@ static int unload_module(void)
        AST_LIST_LOCK(&sessions);
        /* Destroy all the interfaces and free their memory */
        while((s = AST_LIST_REMOVE_HEAD(&sessions, list))) {
+               RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
                d = s->device;
                AST_LIST_TRAVERSE(&d->lines, l, list){
                        ast_mutex_lock(&l->lock);
@@ -8627,9 +8644,11 @@ static int unload_module(void)
                                l->mwi_event_sub = stasis_unsubscribe(l->mwi_event_sub);
                        }
                        ast_mutex_unlock(&l->lock);
-                       manager_event(EVENT_FLAG_SYSTEM, "PeerStatus", "ChannelType: Skinny\r\nPeer: Skinny/%s@%s\r\nPeerStatus: Unregistered\r\n", l->name, d->name);
                        unregister_exten(l);
                }
+               ast_endpoint_set_state(d->endpoint, AST_ENDPOINT_OFFLINE);
+               blob = ast_json_pack("{s: s}", "peer_status", "Unregistered");
+               ast_endpoint_blob_publish(d->endpoint, ast_endpoint_state_type(), blob);
                if (s->fd > -1)
                        close(s->fd);
                pthread_cancel(s->t);
index e808041..c44d6d0 100644 (file)
@@ -464,6 +464,14 @@ int manager_mwi_init(void);
 int manager_bridging_init(void);
 
 /*!
+ * \brief Initialize support for AMI endpoint events.
+ * \return 0 on success.
+ * \return non-zero on error.
+ * \since 12
+ */
+int manager_endpoints_init(void);
+
+/*!
  * \since 12
  * \brief Get the \ref stasis_message_type for generic messages
  *
index aee27aa..b48ed9f 100644 (file)
@@ -30,6 +30,8 @@
 #include "asterisk/sorcery.h"
 /* Needed for ast_dnsmgr */
 #include "asterisk/dnsmgr.h"
+/* Needed for ast_endpoint */
+#include "asterisk/endpoints.h"
 /* Needed for pj_sockaddr */
 #include <pjlib.h>
 
@@ -326,6 +328,10 @@ struct ast_sip_endpoint {
        unsigned int send_rpid;
        /*! Should unsolicited MWI be aggregated into a single NOTIFY? */
        unsigned int aggregate_mwi;
+       /*! Pointer to the persistent Asterisk endpoint */
+       struct ast_endpoint *persistent;
+       /*! The number of channels at which busy device state is returned */
+       unsigned int devicestate_busy_at;
 };
 
 /*!
index 10abcd7..6d7f8db 100644 (file)
@@ -79,6 +79,45 @@ struct ast_endpoint_blob {
 };
 
 /*!
+ * \since 12
+ * \brief Creates a \ref ast_endpoint_blob message.
+ *
+ * The given \a blob should be treated as immutable and not modified after it is
+ * put into the message.
+ *
+ * \param endpoint Endpoint blob is associated with.
+ * \param type Message type for this blob.
+ * \param blob JSON object representing the data, or \c NULL for no data. If
+ *             \c NULL, ast_json_null() is put into the object.
+ *
+ * \return \ref ast_endpoint_blob message.
+ * \return \c NULL on error
+ */
+struct stasis_message *ast_endpoint_blob_create(struct ast_endpoint *endpoint,
+       struct stasis_message_type *type, struct ast_json *blob);
+
+/*!
+ * \since 12
+ * \brief Creates and publishes a \ref ast_endpoint_blob message.
+ *
+ * The given \a blob should be treated as immutable and not modified after it is
+ * put into the message.
+ *
+ * \param endpoint Endpoint blob is associated with.
+ * \param type Message type for this blob.
+ * \param blob JSON object representing the data, or \c NULL for no data. If
+ *             \c NULL, ast_json_null() is put into the object.
+ */
+void ast_endpoint_blob_publish(struct ast_endpoint *endpoint, struct stasis_message_type *type,
+       struct ast_json *blob);
+
+/*!
+ * \brief Message type for endpoint state changes.
+ * \since 12
+ */
+struct stasis_message_type *ast_endpoint_state_type(void);
+
+/*!
  * \brief Message type for \ref ast_endpoint_snapshot.
  * \since 12
  */
index d4960d6..a1b0c7f 100644 (file)
@@ -7771,6 +7771,10 @@ static int __init_manager(int reload, int by_external_config)
                if (manager_bridging_init()) {
                        return -1;
                }
+               if (manager_endpoints_init()) {
+                       ast_log(AST_LOG_ERROR, "Failed to initialize manager endpoints handling\n");
+                       return -1;
+               }
        }
 
        if (!registered) {
diff --git a/main/manager_endpoints.c b/main/manager_endpoints.c
new file mode 100644 (file)
index 0000000..1a36424
--- /dev/null
@@ -0,0 +1,104 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2013, Digium, Inc.
+ *
+ * Joshua Colp <jcolp@digium.com>
+ * David M. Lee, II <dlee@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+/*! \file
+ *
+ * \brief The Asterisk Management Interface - AMI (endpoint handling)
+ *
+ * \author Joshua Colp <jcolp@digium.com>
+ * \author David M. Lee, II <dlee@digium.com>
+ *
+  */
+
+#include "asterisk.h"
+
+ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
+
+#include "asterisk/callerid.h"
+#include "asterisk/channel.h"
+#include "asterisk/manager.h"
+#include "asterisk/stasis_message_router.h"
+#include "asterisk/pbx.h"
+#include "asterisk/stasis_endpoints.h"
+
+static struct stasis_message_router *endpoint_router;
+
+/*! \brief The \ref stasis subscription returned by the forwarding of the endpoint topic
+ * to the manager topic
+ */
+static struct stasis_subscription *topic_forwarder;
+
+static void manager_endpoints_shutdown(void)
+{
+       stasis_message_router_unsubscribe_and_join(endpoint_router);
+       endpoint_router = NULL;
+
+       stasis_unsubscribe(topic_forwarder);
+       topic_forwarder = NULL;
+}
+
+int manager_endpoints_init(void)
+{
+       struct stasis_topic *manager_topic;
+       struct stasis_topic *endpoint_topic;
+       struct stasis_message_router *message_router;
+       int ret = 0;
+
+       if (endpoint_router) {
+               /* Already initialized */
+               return 0;
+       }
+
+       ast_register_atexit(manager_endpoints_shutdown);
+
+       manager_topic = ast_manager_get_topic();
+       if (!manager_topic) {
+               return -1;
+       }
+       message_router = ast_manager_get_message_router();
+       if (!message_router) {
+               return -1;
+       }
+       endpoint_topic = stasis_caching_get_topic(ast_endpoint_topic_all_cached());
+       if (!endpoint_topic) {
+               return -1;
+       }
+
+       topic_forwarder = stasis_forward_all(endpoint_topic, manager_topic);
+       if (!topic_forwarder) {
+               return -1;
+       }
+
+       endpoint_router = stasis_message_router_create(endpoint_topic);
+
+       if (!endpoint_router) {
+               return -1;
+       }
+
+       /* If somehow we failed to add any routes, just shut down the whole
+        * thing and fail it.
+        */
+       if (ret) {
+               manager_endpoints_shutdown();
+               return -1;
+       }
+
+       return 0;
+}
+
index d5347cb..90d9685 100644 (file)
@@ -35,12 +35,139 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/stasis.h"
 #include "asterisk/stasis_endpoints.h"
 
+/*** DOCUMENTATION
+       <managerEvent language="en_US" name="PeerStatus">
+               <managerEventInstance class="EVENT_FLAG_SYSTEM">
+                       <synopsis>Raised when the state of a peer changes.</synopsis>
+                       <syntax>
+                               <parameter name="ChannelType">
+                                       <para>The channel technology of the peer.</para>
+                               </parameter>
+                               <parameter name="Peer">
+                                       <para>The name of the peer (including channel technology).</para>
+                               </parameter>
+                               <parameter name="PeerStatus">
+                                       <para>New status of the peer.</para>
+                                       <enumlist>
+                                               <enum name="Unknown"/>
+                                               <enum name="Registered"/>
+                                               <enum name="Unregistered"/>
+                                               <enum name="Rejected"/>
+                                               <enum name="Lagged"/>
+                                       </enumlist>
+                               </parameter>
+                               <parameter name="Cause">
+                                       <para>The reason the status has changed.</para>
+                               </parameter>
+                               <parameter name="Address">
+                                       <para>New address of the peer.</para>
+                               </parameter>
+                               <parameter name="Port">
+                                       <para>New port for the peer.</para>
+                               </parameter>
+                               <parameter name="Time">
+                                       <para>Time it takes to reach the peer and receive a response.</para>
+                               </parameter>
+                       </syntax>
+               </managerEventInstance>
+       </managerEvent>
+***/
+
+static struct ast_manager_event_blob *peerstatus_to_ami(struct stasis_message *msg);
+
 STASIS_MESSAGE_TYPE_DEFN(ast_endpoint_snapshot_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_endpoint_state_type,
+       .to_ami = peerstatus_to_ami,
+);
 
 static struct stasis_topic *endpoint_topic_all;
 
 static struct stasis_caching_topic *endpoint_topic_all_cached;
 
+static struct ast_manager_event_blob *peerstatus_to_ami(struct stasis_message *msg)
+{
+       struct ast_endpoint_blob *obj = stasis_message_data(msg);
+       RAII_VAR(struct ast_str *, peerstatus_event_string, ast_str_create(64), ast_free);
+       const char *value;
+
+       /* peer_status is the only *required* thing */
+       if (!(value = ast_json_string_get(ast_json_object_get(obj->blob, "peer_status")))) {
+               return NULL;
+       }
+       ast_str_append(&peerstatus_event_string, 0, "PeerStatus: %s\r\n", value);
+
+       if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "cause")))) {
+               ast_str_append(&peerstatus_event_string, 0, "Cause: %s\r\n", value);
+       }
+       if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "address")))) {
+               ast_str_append(&peerstatus_event_string, 0, "Address: %s\r\n", value);
+       }
+       if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "port")))) {
+               ast_str_append(&peerstatus_event_string, 0, "Port: %s\r\n", value);
+       }
+       if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "time")))) {
+               ast_str_append(&peerstatus_event_string, 0, "Time: %s\r\n", value);
+       }
+
+       return ast_manager_event_blob_create(EVENT_FLAG_SYSTEM, "PeerStatus",
+               "ChannelType: %s\r\n"
+               "Peer: %s/%s\r\n"
+               "%s",
+               obj->snapshot->tech,
+               obj->snapshot->tech,
+               obj->snapshot->resource,
+               ast_str_buffer(peerstatus_event_string));
+}
+
+static void endpoint_blob_dtor(void *obj)
+{
+       struct ast_endpoint_blob *event = obj;
+       ao2_cleanup(event->snapshot);
+       ast_json_unref(event->blob);
+}
+
+struct stasis_message *ast_endpoint_blob_create(struct ast_endpoint *endpoint,
+       struct stasis_message_type *type, struct ast_json *blob)
+{
+       RAII_VAR(struct ast_endpoint_blob *, obj, NULL, ao2_cleanup);
+       RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+       if (!blob) {
+               blob = ast_json_null();
+       }
+
+       if (!(obj = ao2_alloc(sizeof(*obj), endpoint_blob_dtor))) {
+               return NULL;
+       }
+
+       if (endpoint) {
+               if (!(obj->snapshot = ast_endpoint_snapshot_create(endpoint))) {
+                       return NULL;
+               }
+       }
+
+       obj->blob = ast_json_ref(blob);
+
+       if (!(msg = stasis_message_create(type, obj))) {
+               return NULL;
+       }
+
+       ao2_ref(msg, +1);
+       return msg;
+}
+
+void ast_endpoint_blob_publish(struct ast_endpoint *endpoint, struct stasis_message_type *type,
+       struct ast_json *blob)
+{
+       RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+       if (blob) {
+               message = ast_endpoint_blob_create(endpoint, type, blob);
+       }
+       if (message) {
+               stasis_publish(ast_endpoint_topic(endpoint), message);
+       }
+}
+
 struct stasis_topic *ast_endpoint_topic_all(void)
 {
        return endpoint_topic_all;
@@ -175,5 +302,9 @@ int ast_endpoint_stasis_init(void)
                return -1;
        }
 
+       if (STASIS_MESSAGE_TYPE_INIT(ast_endpoint_state_type) != 0) {
+               return -1;
+       }
+
        return 0;
 }
index 7f6b061..1af3fb7 100644 (file)
                                <configOption name="use_ptime" default="no">
                                        <synopsis>Use Endpoint's requested packetisation interval</synopsis>
                                </configOption>
+                               <configOption name="devicestate_busy_at" default="0">
+                                       <synopsis>The number of in-use channels which will cause busy to be returned as device state</synopsis>
+                                       <description><para>
+                                               When the number of in-use channels for the endpoint matches the devicestate_busy_at setting the
+                                               Gulp channel driver will return busy as the device state instead of in use.
+                                       </para></description>
+                               </configOption>
                        </configObject>
                        <configObject name="auth">
                                <synopsis>Authentication type</synopsis>
index 489ba6a..3488d52 100644 (file)
 #include "asterisk/utils.h"
 #include "asterisk/sorcery.h"
 #include "asterisk/callerid.h"
+#include "asterisk/stasis_endpoints.h"
+
+/*! \brief Number of buckets for persistent endpoint information */
+#define PERSISTENT_BUCKETS 53
+
+/*! \brief Persistent endpoint information */
+struct sip_persistent_endpoint {
+       /*! \brief Asterisk endpoint itself */
+       struct ast_endpoint *endpoint;
+       /*! \brief AORs that we should react to */
+       char *aors;
+};
+
+/*! \brief Container for persistent endpoint information */
+static struct ao2_container *persistent_endpoints;
 
 static struct ast_sorcery *sip_sorcery;
 
+/*! \brief Hashing function for persistent endpoint information */
+static int persistent_endpoint_hash(const void *obj, const int flags)
+{
+       const struct sip_persistent_endpoint *persistent = obj;
+       const char *id = (flags & OBJ_KEY ? obj : ast_endpoint_get_resource(persistent->endpoint));
+
+       return ast_str_hash(id);
+}
+
+/*! \brief Comparison function for persistent endpoint information */
+static int persistent_endpoint_cmp(void *obj, void *arg, int flags)
+{
+       const struct sip_persistent_endpoint *persistent1 = obj;
+       const struct sip_persistent_endpoint *persistent2 = arg;
+       const char *id = (flags & OBJ_KEY ? arg : ast_endpoint_get_resource(persistent2->endpoint));
+
+       return !strcmp(ast_endpoint_get_resource(persistent1->endpoint), id) ? CMP_MATCH | CMP_STOP : 0;
+}
+
+/*! \brief Callback function for changing the state of an endpoint */
+static int persistent_endpoint_update_state(void *obj, void *arg, int flags)
+{
+       struct sip_persistent_endpoint *persistent = obj;
+       char *aor = arg;
+       RAII_VAR(struct ast_sip_contact *, contact, NULL, ao2_cleanup);
+       RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+
+       if (!ast_strlen_zero(aor) && !strstr(persistent->aors, aor)) {
+               return 0;
+       }
+
+       if ((contact = ast_sip_location_retrieve_contact_from_aor_list(persistent->aors))) {
+               ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_ONLINE);
+               blob = ast_json_pack("{s: s}", "peer_status", "Reachable");
+       } else {
+               ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_OFFLINE);
+               blob = ast_json_pack("{s: s}", "peer_status", "Unreachable");
+       }
+
+       ast_endpoint_blob_publish(persistent->endpoint, ast_endpoint_state_type(), blob);
+
+       ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "Gulp/%s", ast_endpoint_get_resource(persistent->endpoint));
+
+       return 0;
+}
+
+/*! \brief Function called when stuff relating to a contact happens (created/deleted) */
+static void persistent_endpoint_contact_observer(const void *object)
+{
+       char *id = ast_strdupa(ast_sorcery_object_get_id(object)), *aor = NULL;
+
+       aor = strsep(&id, ";@");
+
+       ao2_callback(persistent_endpoints, OBJ_NODATA, persistent_endpoint_update_state, aor);
+}
+
+/*! \brief Observer for contacts so state can be updated on respective endpoints */
+static struct ast_sorcery_observer state_contact_observer = {
+       .created = persistent_endpoint_contact_observer,
+       .deleted = persistent_endpoint_contact_observer,
+};
+
 static char *handle_cli_show_endpoints(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
 {
        RAII_VAR(struct ao2_container *, endpoints, NULL, ao2_cleanup);
@@ -281,12 +358,67 @@ static void *sip_nat_hook_alloc(const char *name)
        return ao2_alloc(sizeof(struct ast_sip_nat_hook), NULL);
 }
 
+/*! \brief Destructor function for persistent endpoint information */
+static void persistent_endpoint_destroy(void *obj)
+{
+       struct sip_persistent_endpoint *persistent = obj;
+
+       ast_endpoint_shutdown(persistent->endpoint);
+       ast_free(persistent->aors);
+}
+
+/*! \brief Internal function which finds (or creates) persistent endpoint information */
+static struct ast_endpoint *persistent_endpoint_find_or_create(const struct ast_sip_endpoint *endpoint)
+{
+       RAII_VAR(struct sip_persistent_endpoint *, persistent, NULL, ao2_cleanup);
+       SCOPED_AO2LOCK(lock, persistent_endpoints);
+
+       if (!(persistent = ao2_find(persistent_endpoints, ast_sorcery_object_get_id(endpoint), OBJ_KEY | OBJ_NOLOCK))) {
+               if (!(persistent = ao2_alloc(sizeof(*persistent), persistent_endpoint_destroy))) {
+                       return NULL;
+               }
+
+               if (!(persistent->endpoint = ast_endpoint_create("Gulp", ast_sorcery_object_get_id(endpoint)))) {
+                       return NULL;
+               }
+
+               persistent->aors = ast_strdup(endpoint->aors);
+
+               if (ast_strlen_zero(persistent->aors)) {
+                       ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_UNKNOWN);
+               } else {
+                       persistent_endpoint_update_state(persistent, NULL, 0);
+               }
+
+               ao2_link_flags(persistent_endpoints, persistent, OBJ_NOLOCK);
+       }
+
+       ao2_ref(persistent->endpoint, +1);
+       return persistent->endpoint;
+}
+
+/*! \brief Callback function for when an object is finalized */
+static int sip_endpoint_apply_handler(const struct ast_sorcery *sorcery, void *obj)
+{
+       struct ast_sip_endpoint *endpoint = obj;
+
+       if (!(endpoint->persistent = persistent_endpoint_find_or_create(endpoint))) {
+               return -1;
+       }
+
+       return 0;
+}
+
 int ast_res_sip_initialize_configuration(void)
 {
        if (ast_cli_register_multiple(cli_commands, ARRAY_LEN(cli_commands))) {
                return -1;
        }
 
+       if (!(persistent_endpoints = ao2_container_alloc(PERSISTENT_BUCKETS, persistent_endpoint_hash, persistent_endpoint_cmp))) {
+               return -1;
+       }
+
        if (!(sip_sorcery = ast_sorcery_open())) {
                ast_log(LOG_ERROR, "Failed to open SIP sorcery failed to open\n");
                return -1;
@@ -305,7 +437,7 @@ int ast_res_sip_initialize_configuration(void)
 
        ast_sorcery_apply_default(sip_sorcery, "nat_hook", "memory", NULL);
 
-       if (ast_sorcery_object_register(sip_sorcery, "endpoint", ast_sip_endpoint_alloc, NULL, NULL)) {
+       if (ast_sorcery_object_register(sip_sorcery, "endpoint", ast_sip_endpoint_alloc, NULL, sip_endpoint_apply_handler)) {
                ast_log(LOG_ERROR, "Failed to register SIP endpoint object with sorcery\n");
                ast_sorcery_unref(sip_sorcery);
                sip_sorcery = NULL;
@@ -351,6 +483,7 @@ int ast_res_sip_initialize_configuration(void)
        ast_sorcery_object_field_register(sip_sorcery, "endpoint", "send_rpid", "no", OPT_BOOL_T, 1, FLDSET(struct ast_sip_endpoint, send_rpid));
        ast_sorcery_object_field_register(sip_sorcery, "endpoint", "mailboxes", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_endpoint, mailboxes));
        ast_sorcery_object_field_register(sip_sorcery, "endpoint", "aggregate_mwi", "yes", OPT_BOOL_T, 1, FLDSET(struct ast_sip_endpoint, aggregate_mwi));
+       ast_sorcery_object_field_register(sip_sorcery, "endpoint", "devicestate_busy_at", "0", OPT_UINT_T, 0, FLDSET(struct ast_sip_endpoint, devicestate_busy_at));
 
        if (ast_sip_initialize_sorcery_transport(sip_sorcery)) {
                ast_log(LOG_ERROR, "Failed to register SIP transport support with sorcery\n");
@@ -366,6 +499,8 @@ int ast_res_sip_initialize_configuration(void)
                return -1;
        }
 
+       ast_sorcery_observer_add(sip_sorcery, "contact", &state_contact_observer);
+
        if (ast_sip_initialize_sorcery_domain_alias(sip_sorcery)) {
                ast_log(LOG_ERROR, "Failed to register SIP domain aliases support with sorcery\n");
                ast_sorcery_unref(sip_sorcery);
@@ -404,6 +539,7 @@ static void endpoint_destructor(void* obj)
        destroy_auths(endpoint->sip_inbound_auths, endpoint->num_inbound_auths);
        destroy_auths(endpoint->sip_outbound_auths, endpoint->num_outbound_auths);
        ast_party_id_free(&endpoint->id);
+       ao2_cleanup(endpoint->persistent);
 }
 
 void *ast_sip_endpoint_alloc(const char *name)