Migrate a large number of AMI events over to Stasis-Core
[asterisk/asterisk.git] / main / stasis_channels.c
index c032ec9..d3c543a 100644 (file)
@@ -48,6 +48,14 @@ STASIS_MESSAGE_TYPE_DEFN(ast_channel_user_event_type);
 STASIS_MESSAGE_TYPE_DEFN(ast_channel_hangup_request_type);
 STASIS_MESSAGE_TYPE_DEFN(ast_channel_dtmf_begin_type);
 STASIS_MESSAGE_TYPE_DEFN(ast_channel_dtmf_end_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_chanspy_start_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_chanspy_stop_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_fax_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_hangup_handler_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_moh_start_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_moh_stop_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_monitor_start_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_monitor_stop_type);
 /*! @} */
 
 /*! \brief Topic for all channels */
@@ -134,6 +142,7 @@ struct ast_channel_snapshot *ast_channel_snapshot_create(struct ast_channel *cha
                S_COR(ast_channel_connected(chan)->id.name.valid, ast_channel_connected(chan)->id.name.str, ""));
        ast_string_field_set(snapshot, connected_number,
                S_COR(ast_channel_connected(chan)->id.number.valid, ast_channel_connected(chan)->id.number.str, ""));
+       ast_string_field_set(snapshot, language, ast_channel_language(chan));
 
        snapshot->creationtime = ast_channel_creationtime(chan);
        snapshot->state = ast_channel_state(chan);
@@ -174,8 +183,7 @@ void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *pe
        struct ast_channel_snapshot *peer_snapshot;
 
        ast_assert(peer != NULL);
-       blob = ast_json_pack("{s: s, s: s, s: s}",
-                            "type", "dial",
+       blob = ast_json_pack("{s: s, s: s}",
                             "dialstatus", S_OR(dialstatus, ""),
                             "dialstring", S_OR(dialstring, ""));
        if (!blob) {
@@ -208,11 +216,13 @@ void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *pe
        publish_message_for_channel_topics(msg, caller);
 }
 
-struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
-       struct stasis_message_type *type, struct ast_json *blob)
+static struct stasis_message *create_channel_blob_message(struct ast_channel_snapshot *snapshot,
+               struct stasis_message_type *type,
+               struct ast_json *blob)
+
 {
-       RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup);
        RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+       RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup);
 
        if (blob == NULL) {
                blob = ast_json_null();
@@ -223,13 +233,10 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
                return NULL;
        }
 
-       if (chan) {
-               obj->snapshot = ast_channel_snapshot_create(chan);
-               if (obj->snapshot == NULL) {
-                       return NULL;
-               }
+       if (snapshot) {
+               obj->snapshot = snapshot;
+               ao2_ref(obj->snapshot, +1);
        }
-
        obj->blob = ast_json_ref(blob);
 
        msg = stasis_message_create(type, obj);
@@ -241,6 +248,29 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
        return msg;
 }
 
+struct stasis_message *ast_channel_cached_blob_create(struct ast_channel *chan,
+                                              struct stasis_message_type *type,
+                                              struct ast_json *blob)
+{
+       RAII_VAR(struct ast_channel_snapshot *, snapshot,
+                       ast_channel_snapshot_get_latest(ast_channel_uniqueid(chan)),
+                       ao2_cleanup);
+
+       return create_channel_blob_message(snapshot, type, blob);
+}
+
+struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
+       struct stasis_message_type *type, struct ast_json *blob)
+{
+       RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
+
+       if (chan) {
+               snapshot = ast_channel_snapshot_create(chan);
+       }
+
+       return create_channel_blob_message(snapshot, type, blob);
+}
+
 /*! \brief A channel snapshot wrapper object used in \ref ast_multi_channel_blob objects */
 struct channel_role_snapshot {
        struct ast_channel_snapshot *snapshot;  /*!< A channel snapshot */
@@ -312,6 +342,28 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
        return obj;
 }
 
+struct ast_channel_snapshot *ast_channel_snapshot_get_latest(const char *uniqueid)
+{
+       RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+       struct ast_channel_snapshot *snapshot;
+
+       ast_assert(!ast_strlen_zero(uniqueid));
+
+       message = stasis_cache_get(ast_channel_topic_all_cached(),
+                       ast_channel_snapshot_type(),
+                       uniqueid);
+       if (!message) {
+               return NULL;
+       }
+
+       snapshot = stasis_message_data(message);
+       if (!snapshot) {
+               return NULL;
+       }
+       ao2_ref(snapshot, +1);
+       return snapshot;
+}
+
 static void channel_role_snapshot_dtor(void *obj)
 {
        struct channel_role_snapshot *role_snapshot = obj;
@@ -390,6 +442,25 @@ struct ast_json *ast_multi_channel_blob_get_json(struct ast_multi_channel_blob *
        return obj->blob;
 }
 
+void ast_channel_publish_snapshot(struct ast_channel *chan)
+{
+       RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
+       RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+
+       snapshot = ast_channel_snapshot_create(chan);
+       if (!snapshot) {
+               return;
+       }
+
+       message = stasis_message_create(ast_channel_snapshot_type(), snapshot);
+       if (!message) {
+               return;
+       }
+
+       ast_assert(ast_channel_topic(chan) != NULL);
+       stasis_publish(ast_channel_topic(chan), message);
+}
+
 void ast_channel_publish_varset(struct ast_channel *chan, const char *name, const char *value)
 {
        RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
@@ -398,8 +469,7 @@ void ast_channel_publish_varset(struct ast_channel *chan, const char *name, cons
        ast_assert(name != NULL);
        ast_assert(value != NULL);
 
-       blob = ast_json_pack("{s: s, s: s, s: s}",
-                            "type", "varset",
+       blob = ast_json_pack("{s: s, s: s}",
                             "variable", name,
                             "value", value);
        if (!blob) {
@@ -417,6 +487,30 @@ void ast_channel_publish_varset(struct ast_channel *chan, const char *name, cons
        publish_message_for_channel_topics(msg, chan);
 }
 
+void ast_publish_channel_state(struct ast_channel *chan)
+{
+       RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
+       RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+
+       ast_assert(chan != NULL);
+       if (!chan) {
+               return;
+       }
+
+       snapshot = ast_channel_snapshot_create(chan);
+       if (!snapshot) {
+               return;
+       }
+
+       message = stasis_message_create(ast_channel_snapshot_type(), snapshot);
+       if (!message) {
+               return;
+       }
+
+       ast_assert(ast_channel_topic(chan) != NULL);
+       stasis_publish(ast_channel_topic(chan), message);
+}
+
 struct ast_json *ast_channel_snapshot_to_json(const struct ast_channel_snapshot *snapshot)
 {
        RAII_VAR(struct ast_json *, json_chan, NULL, ast_json_unref);
@@ -448,8 +542,42 @@ struct ast_json *ast_channel_snapshot_to_json(const struct ast_channel_snapshot
        return ast_json_ref(json_chan);
 }
 
+int ast_channel_snapshot_cep_equal(
+       const struct ast_channel_snapshot *old_snapshot,
+       const struct ast_channel_snapshot *new_snapshot)
+{
+       ast_assert(old_snapshot != NULL);
+       ast_assert(new_snapshot != NULL);
+
+       /* We actually get some snapshots with CEP set, but before the
+        * application is set. Since empty application is invalid, we treat
+        * setting the application from nothing as a CEP change.
+        */
+       if (ast_strlen_zero(old_snapshot->appl) &&
+           !ast_strlen_zero(new_snapshot->appl)) {
+               return 0;
+       }
+
+       return old_snapshot->priority == new_snapshot->priority &&
+               strcmp(old_snapshot->context, new_snapshot->context) == 0 &&
+               strcmp(old_snapshot->exten, new_snapshot->exten) == 0;
+}
+
+int ast_channel_snapshot_caller_id_equal(
+       const struct ast_channel_snapshot *old_snapshot,
+       const struct ast_channel_snapshot *new_snapshot)
+{
+       ast_assert(old_snapshot != NULL);
+       ast_assert(new_snapshot != NULL);
+       return strcmp(old_snapshot->caller_number, new_snapshot->caller_number) == 0 &&
+               strcmp(old_snapshot->caller_name, new_snapshot->caller_name) == 0;
+}
+
 void ast_stasis_channels_shutdown(void)
 {
+       channel_topic_all_cached = stasis_caching_unsubscribe_and_join(channel_topic_all_cached);
+       ao2_cleanup(channel_topic_all);
+       channel_topic_all = NULL;
        STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_snapshot_type);
        STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dial_type);
        STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_varset_type);
@@ -457,9 +585,14 @@ void ast_stasis_channels_shutdown(void)
        STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_hangup_request_type);
        STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dtmf_begin_type);
        STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dtmf_end_type);
-       ao2_cleanup(channel_topic_all);
-       channel_topic_all = NULL;
-       channel_topic_all_cached = stasis_caching_unsubscribe(channel_topic_all_cached);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_chanspy_start_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_chanspy_stop_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_fax_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_hangup_handler_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_moh_start_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_moh_stop_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_monitor_start_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_monitor_stop_type);
 }
 
 void ast_stasis_channels_init(void)
@@ -471,7 +604,14 @@ void ast_stasis_channels_init(void)
        STASIS_MESSAGE_TYPE_INIT(ast_channel_hangup_request_type);
        STASIS_MESSAGE_TYPE_INIT(ast_channel_dtmf_begin_type);
        STASIS_MESSAGE_TYPE_INIT(ast_channel_dtmf_end_type);
-
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_chanspy_start_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_chanspy_stop_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_fax_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_hangup_handler_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_moh_start_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_moh_stop_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_monitor_start_type);
+       STASIS_MESSAGE_TYPE_INIT(ast_channel_monitor_stop_type);
        channel_topic_all = stasis_topic_create("ast_channel_topic_all");
        channel_topic_all_cached = stasis_caching_topic_create(channel_topic_all, channel_snapshot_get_id);
 }