messaging.c 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548
  1. /*
  2. * Asterisk -- An open source telephony toolkit.
  3. *
  4. * Copyright (C) 2014, Digium, Inc.
  5. *
  6. * Matt Jordan <mjordan@digium.com>
  7. *
  8. * See http://www.asterisk.org for more information about
  9. * the Asterisk project. Please do not directly contact
  10. * any of the maintainers of this project for assistance;
  11. * the project provides a web site, mailing lists and IRC
  12. * channels for your use.
  13. *
  14. * This program is free software, distributed under the terms of
  15. * the GNU General Public License Version 2. See the LICENSE file
  16. * at the top of the source tree.
  17. */
  18. /*!
  19. * \file
  20. *
  21. * \brief Stasis out-of-call text message support
  22. *
  23. * \author Matt Jordan <mjordan@digium.com>
  24. */
  25. #include "asterisk.h"
  26. ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
  27. #include "asterisk/message.h"
  28. #include "asterisk/endpoints.h"
  29. #include "asterisk/astobj2.h"
  30. #include "asterisk/vector.h"
  31. #include "asterisk/lock.h"
  32. #include "asterisk/utils.h"
  33. #include "asterisk/test.h"
  34. #include "messaging.h"
  35. /*!
  36. * \brief Subscription to all technologies
  37. */
  38. #define TECH_WILDCARD "__AST_ALL_TECH"
  39. /*!
  40. * \brief Number of buckets for the \ref endpoint_subscriptions container
  41. */
  42. #define ENDPOINTS_NUM_BUCKETS 127
  43. /*! \brief Storage object for an application */
  44. struct application_tuple {
  45. /*! ao2 ref counted private object to pass to the callback */
  46. void *pvt;
  47. /*! The callback to call when this application has a message */
  48. message_received_cb callback;
  49. /*! The name (key) of the application */
  50. char app_name[];
  51. };
  52. /*! \brief A subscription to some endpoint or technology */
  53. struct message_subscription {
  54. /*! The applications that have subscribed to this endpoint or tech */
  55. AST_VECTOR(, struct application_tuple *) applications;
  56. /*! The name of this endpoint or tech */
  57. char token[];
  58. };
  59. /*! \brief The subscriptions to endpoints */
  60. static struct ao2_container *endpoint_subscriptions;
  61. /*!
  62. * \brief The subscriptions to technologies
  63. *
  64. * \note These are stored separately from standard endpoints, given how
  65. * relatively few of them there are.
  66. */
  67. static AST_VECTOR(,struct message_subscription *) tech_subscriptions;
  68. /*! \brief RWLock for \c tech_subscriptions */
  69. static ast_rwlock_t tech_subscriptions_lock;
  70. /*! \internal \brief Destructor for \c application_tuple */
  71. static void application_tuple_dtor(void *obj)
  72. {
  73. struct application_tuple *tuple = obj;
  74. ao2_cleanup(tuple->pvt);
  75. }
  76. /*! \internal \brief Constructor for \c application_tuple */
  77. static struct application_tuple *application_tuple_alloc(const char *app_name, message_received_cb callback, void *pvt)
  78. {
  79. struct application_tuple *tuple;
  80. size_t size = sizeof(*tuple) + strlen(app_name) + 1;
  81. ast_assert(callback != NULL);
  82. tuple = ao2_alloc_options(size, application_tuple_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
  83. if (!tuple) {
  84. return NULL;
  85. }
  86. strcpy(tuple->app_name, app_name); /* Safe */
  87. tuple->pvt = ao2_bump(pvt);
  88. tuple->callback = callback;
  89. return tuple;
  90. }
  91. /*! \internal \brief Destructor for \ref message_subscription */
  92. static void message_subscription_dtor(void *obj)
  93. {
  94. struct message_subscription *sub = obj;
  95. int i;
  96. for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
  97. struct application_tuple *tuple = AST_VECTOR_GET(&sub->applications, i);
  98. ao2_cleanup(tuple);
  99. }
  100. AST_VECTOR_FREE(&sub->applications);
  101. }
  102. /*! \internal \brief Constructor for \ref message_subscription */
  103. static struct message_subscription *message_subscription_alloc(const char *token)
  104. {
  105. struct message_subscription *sub;
  106. size_t size = sizeof(*sub) + strlen(token) + 1;
  107. sub = ao2_alloc_options(size, message_subscription_dtor, AO2_ALLOC_OPT_LOCK_RWLOCK);
  108. if (!sub) {
  109. return NULL;
  110. }
  111. strcpy(sub->token, token); /* Safe */
  112. return sub;
  113. }
  114. /*! AO2 hash function for \ref message_subscription */
  115. static int message_subscription_hash_cb(const void *obj, const int flags)
  116. {
  117. const struct message_subscription *sub;
  118. const char *key;
  119. switch (flags & OBJ_SEARCH_MASK) {
  120. case OBJ_SEARCH_KEY:
  121. key = obj;
  122. break;
  123. case OBJ_SEARCH_OBJECT:
  124. sub = obj;
  125. key = sub->token;
  126. break;
  127. default:
  128. /* Hash can only work on something with a full key. */
  129. ast_assert(0);
  130. return 0;
  131. }
  132. return ast_str_hash(key);
  133. }
  134. /*! AO2 comparison function for \ref message_subscription */
  135. static int message_subscription_compare_cb(void *obj, void *arg, int flags)
  136. {
  137. const struct message_subscription *object_left = obj;
  138. const struct message_subscription *object_right = arg;
  139. const char *right_key = arg;
  140. int cmp;
  141. switch (flags & OBJ_SEARCH_MASK) {
  142. case OBJ_SEARCH_OBJECT:
  143. right_key = object_right->token;
  144. /* Fall through */
  145. case OBJ_SEARCH_KEY:
  146. cmp = strcmp(object_left->token, right_key);
  147. break;
  148. case OBJ_SEARCH_PARTIAL_KEY:
  149. /*
  150. * We could also use a partial key struct containing a length
  151. * so strlen() does not get called for every comparison instead.
  152. */
  153. cmp = strncmp(object_left->token, right_key, strlen(right_key));
  154. break;
  155. default:
  156. /*
  157. * What arg points to is specific to this traversal callback
  158. * and has no special meaning to astobj2.
  159. */
  160. cmp = 0;
  161. break;
  162. }
  163. if (cmp) {
  164. return 0;
  165. }
  166. /*
  167. * At this point the traversal callback is identical to a sorted
  168. * container.
  169. */
  170. return CMP_MATCH;
  171. }
  172. /*! \internal \brief Convert a \c ast_msg To/From URI to a Stasis endpoint name */
  173. static void msg_to_endpoint(const struct ast_msg *msg, char *buf, size_t len)
  174. {
  175. const char *endpoint = ast_msg_get_endpoint(msg);
  176. snprintf(buf, len, "%s%s%s", ast_msg_get_tech(msg),
  177. ast_strlen_zero(endpoint) ? "" : "/",
  178. S_OR(endpoint, ""));
  179. }
  180. /*! \internal
  181. * \brief Callback from the \c message API that determines if we can handle
  182. * this message
  183. */
  184. static int has_destination_cb(const struct ast_msg *msg)
  185. {
  186. struct message_subscription *sub;
  187. int i;
  188. char buf[256];
  189. msg_to_endpoint(msg, buf, sizeof(buf));
  190. ast_rwlock_rdlock(&tech_subscriptions_lock);
  191. for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
  192. sub = AST_VECTOR_GET(&tech_subscriptions, i);
  193. if (!sub) {
  194. continue;
  195. }
  196. if (!strcmp(sub->token, TECH_WILDCARD)
  197. || !strncasecmp(sub->token, buf, strlen(sub->token))
  198. || !strncasecmp(sub->token, buf, strlen(sub->token))) {
  199. ast_rwlock_unlock(&tech_subscriptions_lock);
  200. goto match;
  201. }
  202. }
  203. ast_rwlock_unlock(&tech_subscriptions_lock);
  204. sub = ao2_find(endpoint_subscriptions, buf, OBJ_SEARCH_KEY);
  205. if (sub) {
  206. ao2_ref(sub, -1);
  207. goto match;
  208. }
  209. ast_debug(1, "No subscription found for %s\n", buf);
  210. return 0;
  211. match:
  212. return 1;
  213. }
  214. static struct ast_json *msg_to_json(struct ast_msg *msg)
  215. {
  216. struct ast_json *json_obj;
  217. struct ast_json *json_vars;
  218. struct ast_msg_var_iterator *it_vars;
  219. const char *name;
  220. const char *value;
  221. it_vars = ast_msg_var_iterator_init(msg);
  222. if (!it_vars) {
  223. return NULL;
  224. }
  225. json_vars = ast_json_array_create();
  226. if (!json_vars) {
  227. ast_msg_var_iterator_destroy(it_vars);
  228. return NULL;
  229. }
  230. while (ast_msg_var_iterator_next(msg, it_vars, &name, &value)) {
  231. struct ast_json *json_tuple;
  232. json_tuple = ast_json_pack("{s: s}", name, value);
  233. if (!json_tuple) {
  234. ast_json_unref(json_vars);
  235. ast_msg_var_iterator_destroy(it_vars);
  236. return NULL;
  237. }
  238. ast_json_array_append(json_vars, json_tuple);
  239. ast_msg_var_unref_current(it_vars);
  240. }
  241. ast_msg_var_iterator_destroy(it_vars);
  242. json_obj = ast_json_pack("{s: s, s: s, s: s, s: o}",
  243. "from", ast_msg_get_from(msg),
  244. "to", ast_msg_get_to(msg),
  245. "body", ast_msg_get_body(msg),
  246. "variables", json_vars);
  247. return json_obj;
  248. }
  249. static int handle_msg_cb(struct ast_msg *msg)
  250. {
  251. struct message_subscription *sub;
  252. int i;
  253. char buf[256];
  254. const char *endpoint_name;
  255. struct ast_json *json_msg;
  256. msg_to_endpoint(msg, buf, sizeof(buf));
  257. ast_rwlock_rdlock(&tech_subscriptions_lock);
  258. for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
  259. sub = AST_VECTOR_GET(&tech_subscriptions, i);
  260. if (!sub) {
  261. continue;
  262. }
  263. if (!strcmp(sub->token, TECH_WILDCARD)
  264. || !strncasecmp(sub->token, buf, strlen(sub->token))) {
  265. ast_rwlock_unlock(&tech_subscriptions_lock);
  266. ao2_bump(sub);
  267. endpoint_name = buf;
  268. goto match;
  269. }
  270. }
  271. ast_rwlock_unlock(&tech_subscriptions_lock);
  272. sub = ao2_find(endpoint_subscriptions, buf, OBJ_SEARCH_KEY);
  273. if (sub) {
  274. endpoint_name = buf;
  275. goto match;
  276. }
  277. return -1;
  278. match:
  279. ast_debug(3, "Dispatching message for %s\n", endpoint_name);
  280. json_msg = msg_to_json(msg);
  281. if (!json_msg) {
  282. ao2_ref(sub, -1);
  283. return -1;
  284. }
  285. for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
  286. struct application_tuple *tuple = AST_VECTOR_GET(&sub->applications, i);
  287. tuple->callback(endpoint_name, json_msg, tuple->pvt);
  288. }
  289. ast_json_unref(json_msg);
  290. ao2_ref(sub, -1);
  291. return 0;
  292. }
  293. struct ast_msg_handler ari_msg_handler = {
  294. .name = "ari",
  295. .handle_msg = handle_msg_cb,
  296. .has_destination = has_destination_cb,
  297. };
  298. static int messaging_subscription_cmp(struct message_subscription *sub, const char *key)
  299. {
  300. return !strcmp(sub->token, key) ? 1 : 0;
  301. }
  302. static int application_tuple_cmp(struct application_tuple *item, const char *key)
  303. {
  304. return !strcmp(item->app_name, key) ? 1 : 0;
  305. }
  306. static int is_app_subscribed(struct message_subscription *sub, const char *app_name)
  307. {
  308. int i;
  309. for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
  310. struct application_tuple *tuple;
  311. tuple = AST_VECTOR_GET(&sub->applications, i);
  312. if (tuple && !strcmp(tuple->app_name, app_name)) {
  313. return 1;
  314. }
  315. }
  316. return 0;
  317. }
  318. static struct message_subscription *get_subscription(struct ast_endpoint *endpoint)
  319. {
  320. struct message_subscription *sub = NULL;
  321. if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
  322. sub = ao2_find(endpoint_subscriptions, endpoint, OBJ_SEARCH_KEY);
  323. } else {
  324. int i;
  325. ast_rwlock_rdlock(&tech_subscriptions_lock);
  326. for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
  327. sub = AST_VECTOR_GET(&tech_subscriptions, i);
  328. if (sub && !strcmp(sub->token, endpoint ? ast_endpoint_get_tech(endpoint) : TECH_WILDCARD)) {
  329. ao2_bump(sub);
  330. break;
  331. }
  332. }
  333. ast_rwlock_unlock(&tech_subscriptions_lock);
  334. }
  335. return sub;
  336. }
  337. void messaging_app_unsubscribe_endpoint(const char *app_name, const char *endpoint_id)
  338. {
  339. RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
  340. RAII_VAR(struct ast_endpoint *, endpoint, NULL, ao2_cleanup);
  341. endpoint = ast_endpoint_find_by_id(endpoint_id);
  342. sub = get_subscription(endpoint);
  343. if (!sub) {
  344. return;
  345. }
  346. ao2_lock(sub);
  347. if (!is_app_subscribed(sub, app_name)) {
  348. ao2_unlock(sub);
  349. return;
  350. }
  351. AST_VECTOR_REMOVE_CMP_UNORDERED(&sub->applications, app_name, application_tuple_cmp, ao2_cleanup);
  352. if (AST_VECTOR_SIZE(&sub->applications) == 0) {
  353. if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
  354. ao2_unlink(endpoint_subscriptions, sub);
  355. } else {
  356. ast_rwlock_wrlock(&tech_subscriptions_lock);
  357. AST_VECTOR_REMOVE_CMP_UNORDERED(&tech_subscriptions, endpoint ? ast_endpoint_get_id(endpoint) : TECH_WILDCARD,
  358. messaging_subscription_cmp, AST_VECTOR_ELEM_CLEANUP_NOOP);
  359. ast_rwlock_unlock(&tech_subscriptions_lock);
  360. }
  361. }
  362. ao2_unlock(sub);
  363. ao2_ref(sub, -1);
  364. ast_debug(3, "App '%s' unsubscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
  365. ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Unsubscribed\r\nAppName: %s\r\nToken: %s\r\n",
  366. app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
  367. }
  368. static struct message_subscription *get_or_create_subscription(struct ast_endpoint *endpoint)
  369. {
  370. struct message_subscription *sub = get_subscription(endpoint);
  371. if (sub) {
  372. return sub;
  373. }
  374. sub = message_subscription_alloc(endpoint ? ast_endpoint_get_id(endpoint) : TECH_WILDCARD);
  375. if (!sub) {
  376. return NULL;
  377. }
  378. if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
  379. ao2_link(endpoint_subscriptions, sub);
  380. } else {
  381. ast_rwlock_wrlock(&tech_subscriptions_lock);
  382. ao2_ref(sub, +1);
  383. if (AST_VECTOR_APPEND(&tech_subscriptions, sub)) {
  384. /* Release the refs that were for the vector and the allocation. */
  385. ao2_ref(sub, -2);
  386. sub = NULL;
  387. }
  388. ast_rwlock_unlock(&tech_subscriptions_lock);
  389. }
  390. return sub;
  391. }
  392. int messaging_app_subscribe_endpoint(const char *app_name, struct ast_endpoint *endpoint, message_received_cb callback, void *pvt)
  393. {
  394. RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
  395. struct application_tuple *tuple;
  396. sub = get_or_create_subscription(endpoint);
  397. if (!sub) {
  398. return -1;
  399. }
  400. ao2_lock(sub);
  401. if (is_app_subscribed(sub, app_name)) {
  402. ao2_unlock(sub);
  403. return 0;
  404. }
  405. tuple = application_tuple_alloc(app_name, callback, pvt);
  406. if (!tuple) {
  407. ao2_unlock(sub);
  408. return -1;
  409. }
  410. if (AST_VECTOR_APPEND(&sub->applications, tuple)) {
  411. ao2_ref(tuple, -1);
  412. ao2_unlock(sub);
  413. return -1;
  414. }
  415. ao2_unlock(sub);
  416. ast_debug(3, "App '%s' subscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
  417. ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Subscribed\r\nAppName: %s\r\nToken: %s\r\n",
  418. app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
  419. return 0;
  420. }
  421. int messaging_cleanup(void)
  422. {
  423. ast_msg_handler_unregister(&ari_msg_handler);
  424. ao2_ref(endpoint_subscriptions, -1);
  425. AST_VECTOR_FREE(&tech_subscriptions);
  426. ast_rwlock_destroy(&tech_subscriptions_lock);\
  427. return 0;
  428. }
  429. int messaging_init(void)
  430. {
  431. endpoint_subscriptions = ao2_t_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
  432. ENDPOINTS_NUM_BUCKETS, message_subscription_hash_cb, NULL,
  433. message_subscription_compare_cb, "Endpoint messaging subscription container creation");
  434. if (!endpoint_subscriptions) {
  435. return -1;
  436. }
  437. if (AST_VECTOR_INIT(&tech_subscriptions, 4)) {
  438. ao2_ref(endpoint_subscriptions, -1);
  439. return -1;
  440. }
  441. if (ast_rwlock_init(&tech_subscriptions_lock)) {
  442. ao2_ref(endpoint_subscriptions, -1);
  443. AST_VECTOR_FREE(&tech_subscriptions);
  444. return -1;
  445. }
  446. if (ast_msg_handler_register(&ari_msg_handler)) {
  447. ao2_ref(endpoint_subscriptions, -1);
  448. AST_VECTOR_FREE(&tech_subscriptions);
  449. ast_rwlock_destroy(&tech_subscriptions_lock);
  450. return -1;
  451. }
  452. return 0;
  453. }