Asterisk - The Open Source Telephony Project  GIT-master-93d0901
stasis_endpoints.c
Go to the documentation of this file.
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * David M. Lee, II <dlee@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 endpoint API.
22  *
23  * \author David M. Lee, II <dlee@digium.com>
24  */
25 
26 /*** MODULEINFO
27  <support_level>core</support_level>
28  ***/
29 
30 #include "asterisk.h"
31 
32 #include "asterisk/astobj2.h"
33 #include "asterisk/stasis.h"
35 
36 /*** DOCUMENTATION
37  <managerEvent language="en_US" name="PeerStatus">
38  <managerEventInstance class="EVENT_FLAG_SYSTEM">
39  <synopsis>Raised when the state of a peer changes.</synopsis>
40  <syntax>
41  <parameter name="ChannelType">
42  <para>The channel technology of the peer.</para>
43  </parameter>
44  <parameter name="Peer">
45  <para>The name of the peer (including channel technology).</para>
46  </parameter>
47  <parameter name="PeerStatus">
48  <para>New status of the peer.</para>
49  <enumlist>
50  <enum name="Unknown"/>
51  <enum name="Registered"/>
52  <enum name="Unregistered"/>
53  <enum name="Rejected"/>
54  <enum name="Lagged"/>
55  </enumlist>
56  </parameter>
57  <parameter name="Cause">
58  <para>The reason the status has changed.</para>
59  </parameter>
60  <parameter name="Address">
61  <para>New address of the peer.</para>
62  </parameter>
63  <parameter name="Port">
64  <para>New port for the peer.</para>
65  </parameter>
66  <parameter name="Time">
67  <para>Time it takes to reach the peer and receive a response.</para>
68  </parameter>
69  </syntax>
70  </managerEventInstance>
71  </managerEvent>
72  <managerEvent language="en_US" name="ContactStatus">
73  <managerEventInstance class="EVENT_FLAG_SYSTEM">
74  <synopsis>Raised when the state of a contact changes.</synopsis>
75  <syntax>
76  <parameter name="URI">
77  <para>This contact's URI.</para>
78  </parameter>
79  <parameter name="ContactStatus">
80  <para>New status of the contact.</para>
81  <enumlist>
82  <enum name="Unknown"/>
83  <enum name="Unreachable"/>
84  <enum name="Reachable"/>
85  <enum name="Unqualified"/>
86  <enum name="Removed"/>
87  <enum name="Updated"/>
88  </enumlist>
89  </parameter>
90  <parameter name="AOR">
91  <para>The name of the associated aor.</para>
92  </parameter>
93  <parameter name="EndpointName">
94  <para>The name of the associated endpoint.</para>
95  </parameter>
96  <parameter name="RoundtripUsec">
97  <para>The RTT measured during the last qualify.</para>
98  </parameter>
99  </syntax>
100  </managerEventInstance>
101  </managerEvent>
102 ***/
103 
105 
107 {
108  return endpoint_cache_all;
109 }
110 
112 {
114 }
115 
117 {
119 }
120 
122 {
124 }
125 
127 
129 {
130  struct ast_endpoint_blob *obj = stasis_message_data(msg);
131  RAII_VAR(struct ast_str *, peerstatus_event_string, ast_str_create(64), ast_free);
132  const char *value;
133 
134  /* peer_status is the only *required* thing */
135  if (!(value = ast_json_string_get(ast_json_object_get(obj->blob, "peer_status")))) {
136  return NULL;
137  }
138  ast_str_append(&peerstatus_event_string, 0, "PeerStatus: %s\r\n", value);
139 
140  if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "cause")))) {
141  ast_str_append(&peerstatus_event_string, 0, "Cause: %s\r\n", value);
142  }
143  if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "address")))) {
144  ast_str_append(&peerstatus_event_string, 0, "Address: %s\r\n", value);
145  }
146  if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "port")))) {
147  ast_str_append(&peerstatus_event_string, 0, "Port: %s\r\n", value);
148  }
149  if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "time")))) {
150  ast_str_append(&peerstatus_event_string, 0, "Time: %s\r\n", value);
151  }
152 
154  "ChannelType: %s\r\n"
155  "Peer: %s/%s\r\n"
156  "%s",
157  obj->snapshot->tech,
158  obj->snapshot->tech,
159  obj->snapshot->resource,
160  ast_str_buffer(peerstatus_event_string));
161 }
162 
163 static struct ast_json *peerstatus_to_json(struct stasis_message *msg, const struct stasis_message_sanitizer *sanitize)
164 {
165  struct ast_endpoint_blob *obj = stasis_message_data(msg);
166  struct ast_json *json_endpoint;
167  struct ast_json *json_peer;
168  struct ast_json *json_final;
169  const struct timeval *tv = stasis_message_timestamp(msg);
170 
171  json_endpoint = ast_endpoint_snapshot_to_json(obj->snapshot, NULL);
172  if (!json_endpoint) {
173  return NULL;
174  }
175 
176  json_peer = ast_json_object_create();
177  if (!json_peer) {
178  ast_json_unref(json_endpoint);
179  return NULL;
180  }
181 
182  /* Copy all fields from the blob */
183  ast_json_object_update(json_peer, obj->blob);
184 
185  json_final = ast_json_pack("{s: s, s: o, s: o, s: o }",
186  "type", "PeerStatusChange",
187  "timestamp", ast_json_timeval(*tv, NULL),
188  "endpoint", json_endpoint,
189  "peer", json_peer);
190  if (!json_final) {
191  ast_json_unref(json_endpoint);
192  ast_json_unref(json_peer);
193  }
194 
195  return json_final;
196 }
197 
200  .to_json = peerstatus_to_json,
201 );
202 
204 {
205  struct ast_endpoint_blob *obj = stasis_message_data(msg);
206  RAII_VAR(struct ast_str *, contactstatus_event_string, ast_str_create(64), ast_free);
207  const char *value;
208 
209  if (!(value = ast_json_string_get(ast_json_object_get(obj->blob, "uri")))) {
210  return NULL;
211  }
212  ast_str_append(&contactstatus_event_string, 0, "URI: %s\r\n", value);
213 
214  if (!(value = ast_json_string_get(ast_json_object_get(obj->blob, "contact_status")))) {
215  return NULL;
216  }
217  ast_str_append(&contactstatus_event_string, 0, "ContactStatus: %s\r\n", value);
218 
219  if (!(value = ast_json_string_get(ast_json_object_get(obj->blob, "aor")))) {
220  return NULL;
221  }
222  ast_str_append(&contactstatus_event_string, 0, "AOR: %s\r\n", value);
223 
224  if (!(value = ast_json_string_get(ast_json_object_get(obj->blob, "endpoint_name")))) {
225  return NULL;
226  }
227  ast_str_append(&contactstatus_event_string, 0, "EndpointName: %s\r\n", value);
228 
229  if ((value = ast_json_string_get(ast_json_object_get(obj->blob, "roundtrip_usec")))) {
230  ast_str_append(&contactstatus_event_string, 0, "RoundtripUsec: %s\r\n", value);
231  }
232 
233  return ast_manager_event_blob_create(EVENT_FLAG_SYSTEM, "ContactStatus",
234  "%s", ast_str_buffer(contactstatus_event_string));
235 }
236 
237 static struct ast_json *contactstatus_to_json(struct stasis_message *msg, const struct stasis_message_sanitizer *sanitize)
238 {
239  struct ast_endpoint_blob *obj = stasis_message_data(msg);
240  struct ast_json *json_endpoint;
241  struct ast_json *json_final;
242  const char *rtt;
243  const struct timeval *tv = stasis_message_timestamp(msg);
244 
245  json_endpoint = ast_endpoint_snapshot_to_json(obj->snapshot, NULL);
246  if (!json_endpoint) {
247  return NULL;
248  }
249 
250  /* The roundtrip time is optional. */
251  rtt = ast_json_string_get(ast_json_object_get(obj->blob, "roundtrip_usec"));
252  if (!ast_strlen_zero(rtt)) {
253  json_final = ast_json_pack("{s: s, s: o, s: o, s: { s: s, s: s, s: s, s: s } } ",
254  "type", "ContactStatusChange",
255  "timestamp", ast_json_timeval(*tv, NULL),
256  "endpoint", json_endpoint,
257  "contact_info",
258  "uri", ast_json_string_get(ast_json_object_get(obj->blob, "uri")),
259  "contact_status", ast_json_string_get(ast_json_object_get(obj->blob,
260  "contact_status")),
261  "aor", ast_json_string_get(ast_json_object_get(obj->blob, "aor")),
262  "roundtrip_usec", rtt);
263  } else {
264  json_final = ast_json_pack("{s: s, s: o, s: o, s: { s: s, s: s, s: s } } ",
265  "type", "ContactStatusChange",
266  "timestamp", ast_json_timeval(*tv, NULL),
267  "endpoint", json_endpoint,
268  "contact_info",
269  "uri", ast_json_string_get(ast_json_object_get(obj->blob, "uri")),
270  "contact_status", ast_json_string_get(ast_json_object_get(obj->blob,
271  "contact_status")),
272  "aor", ast_json_string_get(ast_json_object_get(obj->blob, "aor")));
273  }
274  if (!json_final) {
275  ast_json_unref(json_endpoint);
276  }
277 
278  return json_final;
279 }
280 
283  .to_json = contactstatus_to_json
284 );
285 
286 static void endpoint_blob_dtor(void *obj)
287 {
288  struct ast_endpoint_blob *event = obj;
289  ao2_cleanup(event->snapshot);
290  ast_json_unref(event->blob);
291 }
292 
294  struct stasis_message_type *type, struct ast_json *blob)
295 {
296  struct ast_endpoint_blob *obj;
297  struct stasis_message *msg;
298 
299  if (!type) {
300  return NULL;
301  }
302  if (!blob) {
303  blob = ast_json_null();
304  }
305 
306  if (!(obj = ao2_alloc(sizeof(*obj), endpoint_blob_dtor))) {
307  return NULL;
308  }
309 
310  if (endpoint) {
311  if (!(obj->snapshot = ast_endpoint_snapshot_create(endpoint))) {
312  ao2_ref(obj, -1);
313 
314  return NULL;
315  }
316  }
317 
318  obj->blob = ast_json_ref(blob);
319  msg = stasis_message_create(type, obj);
320  ao2_ref(obj, -1);
321 
322  return msg;
323 }
324 
326  struct ast_json *blob)
327 {
328  struct stasis_message *message;
329 
330  if (!blob) {
331  return;
332  }
333 
334  message = ast_endpoint_blob_create(endpoint, type, blob);
335  if (message) {
337  ao2_ref(message, -1);
338  }
339 }
340 
342  const char *name)
343 {
344  char *id = NULL;
345  struct stasis_message *msg;
346  struct ast_endpoint_snapshot *snapshot;
347 
348  if (ast_strlen_zero(name)) {
349  ast_asprintf(&id, "%s", tech);
350  } else {
351  ast_asprintf(&id, "%s/%s", tech, name);
352  }
353  if (!id) {
354  return NULL;
355  }
356  ast_tech_to_upper(id);
357 
359  ast_free(id);
360  if (!msg) {
361  return NULL;
362  }
363 
364  snapshot = stasis_message_data(msg);
365  ast_assert(snapshot != NULL);
366 
367  ao2_ref(snapshot, +1);
368  ao2_ref(msg, -1);
369 
370  return snapshot;
371 }
372 
373 /*!
374  * \brief Callback extract a unique identity from a snapshot message.
375  *
376  * This identity is unique to the underlying object of the snapshot, such as the
377  * UniqueId field of a channel.
378  *
379  * \param message Message to extract id from.
380  * \return String representing the snapshot's id.
381  * \retval NULL if the message_type of the message isn't a handled snapshot.
382  * \since 12
383  */
385 {
386  struct ast_endpoint_snapshot *snapshot;
387 
389  return NULL;
390  }
391 
392  snapshot = stasis_message_data(message);
393 
394  return snapshot->id;
395 }
396 
397 
399  const struct ast_endpoint_snapshot *snapshot,
400  const struct stasis_message_sanitizer *sanitize)
401 {
402  struct ast_json *json;
403  struct ast_json *channel_array;
404  int i;
405 
406  json = ast_json_pack("{s: s, s: s, s: s, s: []}",
407  "technology", snapshot->tech,
408  "resource", snapshot->resource,
409  "state", ast_endpoint_state_to_string(snapshot->state),
410  "channel_ids");
411 
412  if (json == NULL) {
413  return NULL;
414  }
415 
416  if (snapshot->max_channels != -1) {
417  int res = ast_json_object_set(json, "max_channels",
419  if (res != 0) {
420  ast_json_unref(json);
421 
422  return NULL;
423  }
424  }
425 
426  channel_array = ast_json_object_get(json, "channel_ids");
427  ast_assert(channel_array != NULL);
428  for (i = 0; i < snapshot->num_channels; ++i) {
429  int res;
430 
431  if (sanitize && sanitize->channel_id
432  && sanitize->channel_id(snapshot->channel_ids[i])) {
433  continue;
434  }
435 
436  res = ast_json_array_append(channel_array,
437  ast_json_string_create(snapshot->channel_ids[i]));
438  if (res != 0) {
439  ast_json_unref(json);
440 
441  return NULL;
442  }
443  }
444 
445  return json;
446 }
447 
448 static void endpoints_stasis_cleanup(void)
449 {
453 
456 }
457 
459 {
460  int res = 0;
462 
463  endpoint_cache_all = stasis_cp_all_create("endpoint:all",
465  if (!endpoint_cache_all) {
466  return -1;
467  }
468 
472 
473  return res;
474 }
Asterisk main include file. File version handling, generic pbx functions.
int ast_register_cleanup(void(*func)(void))
Register a function to be executed before Asterisk gracefully exits.
Definition: clicompat.c:19
#define ast_free(a)
Definition: astmm.h:180
#define ast_asprintf(ret, fmt,...)
A wrapper for asprintf()
Definition: astmm.h:267
#define ao2_cleanup(obj)
Definition: astobj2.h:1934
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ao2_alloc(data_size, destructor_fn)
Definition: astobj2.h:409
static const char type[]
Definition: chan_ooh323.c:109
const char * ast_endpoint_state_to_string(enum ast_endpoint_state state)
Returns a string representation of the given endpoint state.
static const char name[]
Definition: format_mp3.c:68
void ast_endpoint_blob_publish(struct ast_endpoint *endpoint, struct stasis_message_type *type, struct ast_json *blob)
Creates and publishes a ast_endpoint_blob message.
struct stasis_topic * ast_endpoint_topic_all(void)
Topic for all endpoint related messages.
struct stasis_cp_all * ast_endpoint_cache_all(void)
struct stasis_message_type * ast_endpoint_contact_state_type(void)
Message type for endpoint contact state changes.
struct stasis_topic * ast_endpoint_topic(struct ast_endpoint *endpoint)
Returns the topic for a specific endpoint.
struct stasis_cache * ast_endpoint_cache(void)
Backend cache for ast_endpoint_topic_all_cached().
struct stasis_message_type * ast_endpoint_state_type(void)
Message type for endpoint state changes.
struct ast_endpoint_snapshot * ast_endpoint_latest_snapshot(const char *tech, const char *name)
Retrieve the most recent snapshot for the endpoint with the given name.
struct stasis_message * ast_endpoint_blob_create(struct ast_endpoint *endpoint, struct stasis_message_type *type, struct ast_json *blob)
Creates a ast_endpoint_blob message.
struct stasis_topic * ast_endpoint_topic_all_cached(void)
Cached topic for all endpoint related messages.
struct stasis_message_type * ast_endpoint_snapshot_type(void)
Message type for ast_endpoint_snapshot.
struct ast_endpoint_snapshot * ast_endpoint_snapshot_create(struct ast_endpoint *endpoint)
Create a snapshot of an endpoint.
struct ast_json * ast_json_ref(struct ast_json *value)
Increase refcount on value.
Definition: json.c:67
struct ast_json * ast_json_timeval(const struct timeval tv, const char *zone)
Construct a timeval as JSON.
Definition: json.c:649
void ast_json_unref(struct ast_json *value)
Decrease refcount on value. If refcount reaches zero, value is freed.
Definition: json.c:73
int ast_json_array_append(struct ast_json *array, struct ast_json *value)
Append to an array.
Definition: json.c:368
struct ast_json * ast_json_object_get(struct ast_json *object, const char *key)
Get a field from a JSON object.
Definition: json.c:397
struct ast_json * ast_json_object_create(void)
Create a new JSON object.
Definition: json.c:389
const char * ast_json_string_get(const struct ast_json *string)
Get the value of a JSON string.
Definition: json.c:273
struct ast_json * ast_json_null(void)
Get the JSON null value.
Definition: json.c:248
struct ast_json * ast_json_pack(char const *format,...)
Helper for creating complex JSON values.
Definition: json.c:591
int ast_json_object_set(struct ast_json *object, const char *key, struct ast_json *value)
Set a field in a JSON object.
Definition: json.c:404
struct ast_json * ast_json_integer_create(intmax_t value)
Create a JSON integer.
Definition: json.c:317
struct ast_json * ast_json_string_create(const char *value)
Construct a JSON string from value.
Definition: json.c:268
int ast_json_object_update(struct ast_json *object, struct ast_json *other)
Update object with all of the fields of other.
Definition: json.c:416
struct ast_manager_event_blob * ast_manager_event_blob_create(int event_flags, const char *manager_event, const char *extra_fields_fmt,...)
Construct a ast_manager_event_blob.
Definition: manager.c:9724
#define EVENT_FLAG_SYSTEM
Definition: manager.h:75
static void to_ami(struct ast_sip_subscription *sub, struct ast_str **buf)
#define NULL
Definition: resample.c:96
Stasis Message Bus API. See Stasis Message Bus API for detailed documentation.
const struct timeval * stasis_message_timestamp(const struct stasis_message *msg)
Get the time when a message was created.
#define STASIS_MESSAGE_TYPE_CLEANUP(name)
Boiler-plate messaging macro for cleaning up message types.
Definition: stasis.h:1515
#define STASIS_MESSAGE_TYPE_INIT(name)
Boiler-plate messaging macro for initializing message types.
Definition: stasis.h:1493
struct stasis_message * stasis_message_create(struct stasis_message_type *type, void *data)
Create a new message.
struct stasis_message_type * stasis_message_type(const struct stasis_message *msg)
Get the message type for a stasis_message.
struct stasis_message * stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
Retrieve an item from the cache for the ast_eid_default entity.
Definition: stasis_cache.c:686
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition: stasis.c:1513
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.
struct stasis_cp_all * stasis_cp_all_create(const char *name, snapshot_get_id id_fn)
Create an all instance of the cache pattern.
struct stasis_cache * stasis_cp_all_cache(struct stasis_cp_all *all)
Get the cache.
struct stasis_topic * stasis_cp_all_topic(struct stasis_cp_all *all)
Get the aggregate topic.
struct stasis_topic * stasis_cp_all_topic_cached(struct stasis_cp_all *all)
Get the caching topic.
STASIS_MESSAGE_TYPE_DEFN(ast_endpoint_snapshot_type)
struct ast_json * ast_endpoint_snapshot_to_json(const struct ast_endpoint_snapshot *snapshot, const struct stasis_message_sanitizer *sanitize)
Build a JSON object from a ast_endpoint_snapshot.
static struct ast_json * contactstatus_to_json(struct stasis_message *msg, const struct stasis_message_sanitizer *sanitize)
static void endpoint_blob_dtor(void *obj)
static void endpoints_stasis_cleanup(void)
static struct ast_manager_event_blob * contactstatus_to_ami(struct stasis_message *msg)
static struct stasis_cp_all * endpoint_cache_all
static const char * endpoint_snapshot_get_id(struct stasis_message *message)
Callback extract a unique identity from a snapshot message.
static struct ast_json * peerstatus_to_json(struct stasis_message *msg, const struct stasis_message_sanitizer *sanitize)
static struct ast_manager_event_blob * peerstatus_to_ami(struct stasis_message *msg)
int ast_endpoint_stasis_init(void)
Initialization function for endpoint stasis support.
Endpoint abstractions.
int ast_str_append(struct ast_str **buf, ssize_t max_len, const char *fmt,...)
Append to a thread local dynamic string.
Definition: strings.h:1115
char * ast_str_buffer(const struct ast_str *buf)
Returns the string buffer within the ast_str buf.
Definition: strings.h:737
char * ast_tech_to_upper(char *dev_str)
Convert the tech portion of a device string to upper case.
Definition: strings.h:1212
static force_inline int attribute_pure ast_strlen_zero(const char *s)
Definition: strings.h:65
#define ast_str_create(init_len)
Create a malloc'ed dynamic length string.
Definition: strings.h:638
Blob of data associated with an endpoint.
struct ast_endpoint_snapshot * snapshot
struct ast_json * blob
A snapshot of an endpoint's state.
const ast_string_field tech
enum ast_endpoint_state state
const ast_string_field id
const ast_string_field resource
Abstract JSON element (object, array, string, int, ...).
Struct containing info for an AMI event to send out.
Definition: manager.h:490
Support for dynamic strings.
Definition: strings.h:602
Definition: astman.c:222
Structure containing callbacks for Stasis message sanitization.
Definition: stasis.h:200
int(* channel_id)(const char *channel_id)
Callback which determines whether a channel should be sanitized from a message based on the channel's...
Definition: stasis.h:210
int value
Definition: syslog.c:37
#define RAII_VAR(vartype, varname, initval, dtor)
Declare a variable that will call a destructor function when it goes out of scope.
Definition: utils.h:936
#define ast_assert(a)
Definition: utils.h:734