Implement ctf.lttng-live component
[babeltrace.git] / lib / graph / iterator.c
index 389eec528c022d6c1ff32b7e1b7ccde3a07e0a3f..1b1862bd1b87d4299c98ed60c2cb1c94da3db11a 100644 (file)
@@ -33,6 +33,7 @@
 #include <babeltrace/graph/component.h>
 #include <babeltrace/graph/component-source-internal.h>
 #include <babeltrace/graph/component-class-internal.h>
+#include <babeltrace/graph/notification.h>
 #include <babeltrace/graph/notification-iterator.h>
 #include <babeltrace/graph/notification-iterator-internal.h>
 #include <babeltrace/graph/notification-internal.h>
 #include <babeltrace/graph/notification-stream.h>
 #include <babeltrace/graph/notification-stream-internal.h>
 #include <babeltrace/graph/port.h>
+#include <babeltrace/types.h>
+#include <stdint.h>
 
 struct stream_state {
        struct bt_ctf_stream *stream; /* owned by this */
        struct bt_ctf_packet *cur_packet; /* owned by this */
-       bool is_ended;
+       bt_bool is_ended;
 };
 
 enum action_type {
@@ -139,7 +142,7 @@ void destroy_action(struct action *action)
        case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
                break;
        default:
-               assert(false);
+               assert(BT_FALSE);
        }
 }
 
@@ -217,7 +220,7 @@ void apply_actions(struct bt_notification_iterator *iterator)
                        bt_ctf_stream_add_destroy_listener(
                                action->payload.set_stream_state_is_ended.stream_state->stream,
                                stream_destroy_listener, iterator);
-                       action->payload.set_stream_state_is_ended.stream_state->is_ended = true;
+                       action->payload.set_stream_state_is_ended.stream_state->is_ended = BT_TRUE;
                        BT_PUT(action->payload.set_stream_state_is_ended.stream_state->stream);
                        break;
                case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
@@ -226,7 +229,7 @@ void apply_actions(struct bt_notification_iterator *iterator)
                                action->payload.set_stream_state_cur_packet.packet);
                        break;
                default:
-                       assert(false);
+                       assert(BT_FALSE);
                }
        }
 
@@ -341,16 +344,70 @@ void bt_notification_iterator_destroy(struct bt_object *obj)
        g_free(iterator);
 }
 
+static
+int create_subscription_mask_from_notification_types(
+               struct bt_notification_iterator *iterator,
+               const enum bt_notification_type *notif_types)
+{
+       const enum bt_notification_type *notif_type;
+       int ret = 0;
+
+       assert(notif_types);
+       iterator->subscription_mask = 0;
+
+       for (notif_type = notif_types;
+                       *notif_type != BT_NOTIFICATION_TYPE_SENTINEL;
+                       notif_type++) {
+               switch (*notif_type) {
+               case BT_NOTIFICATION_TYPE_ALL:
+                       iterator->subscription_mask |=
+                               BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT |
+                               BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY |
+                               BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN |
+                               BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END |
+                               BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN |
+                               BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
+                       break;
+               case BT_NOTIFICATION_TYPE_EVENT:
+                       iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
+                       break;
+               case BT_NOTIFICATION_TYPE_INACTIVITY:
+                       iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
+                       break;
+               case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
+                       iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
+                       break;
+               case BT_NOTIFICATION_TYPE_STREAM_END:
+                       iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
+                       break;
+               case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
+                       iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
+                       break;
+               case BT_NOTIFICATION_TYPE_PACKET_END:
+                       iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
+                       break;
+               default:
+                       ret = -1;
+                       goto end;
+               }
+       }
+
+end:
+       return ret;
+}
+
 BT_HIDDEN
 struct bt_notification_iterator *bt_notification_iterator_create(
                struct bt_component *upstream_comp,
-               struct bt_port *upstream_port)
+               struct bt_port *upstream_port,
+               const enum bt_notification_type *notification_types)
 {
        enum bt_component_class_type type;
        struct bt_notification_iterator *iterator = NULL;
 
        assert(upstream_comp);
        assert(upstream_port);
+       assert(notification_types);
        assert(bt_port_is_connected(upstream_port));
 
        type = bt_component_get_class_type(upstream_comp);
@@ -369,6 +426,11 @@ struct bt_notification_iterator *bt_notification_iterator_create(
 
        bt_object_init(iterator, bt_notification_iterator_destroy);
 
+       if (create_subscription_mask_from_notification_types(iterator,
+                       notification_types)) {
+               goto error;
+       }
+
        iterator->stream_states = g_hash_table_new_full(g_direct_hash,
                g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
        if (!iterator->stream_states) {
@@ -456,12 +518,45 @@ end:
 }
 
 static
-bool validate_notification(struct bt_notification_iterator *iterator,
+enum bt_notification_iterator_notif_type
+bt_notification_iterator_notif_type_from_notif_type(
+               enum bt_notification_type notif_type)
+{
+       enum bt_notification_iterator_notif_type iter_notif_type;
+
+       switch (notif_type) {
+       case BT_NOTIFICATION_TYPE_EVENT:
+               iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
+               break;
+       case BT_NOTIFICATION_TYPE_INACTIVITY:
+               iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
+               break;
+       case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
+               iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
+               break;
+       case BT_NOTIFICATION_TYPE_STREAM_END:
+               iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
+               break;
+       case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
+               iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
+               break;
+       case BT_NOTIFICATION_TYPE_PACKET_END:
+               iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
+               break;
+       default:
+               assert(BT_FALSE);
+       }
+
+       return iter_notif_type;
+}
+
+static
+bt_bool validate_notification(struct bt_notification_iterator *iterator,
                struct bt_notification *notif,
                struct bt_ctf_stream *notif_stream,
                struct bt_ctf_packet *notif_packet)
 {
-       bool is_valid = true;
+       bt_bool is_valid = BT_TRUE;
        struct stream_state *stream_state;
        struct bt_port *stream_comp_cur_port;
 
@@ -495,7 +590,7 @@ bool validate_notification(struct bt_notification_iterator *iterator,
                         * bad: the API guarantees that it can never
                         * happen.
                         */
-                       is_valid = false;
+                       is_valid = BT_FALSE;
                        goto end;
                }
 
@@ -513,7 +608,7 @@ bool validate_notification(struct bt_notification_iterator *iterator,
                         * bad: the API guarantees that it can never
                         * happen.
                         */
-                       is_valid = false;
+                       is_valid = BT_FALSE;
                        goto end;
                }
 
@@ -524,12 +619,12 @@ bool validate_notification(struct bt_notification_iterator *iterator,
                         * we already returned a "stream begin"
                         * notification: this is an invalid duplicate.
                         */
-                       is_valid = false;
+                       is_valid = BT_FALSE;
                        goto end;
                case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
                        if (notif_packet == stream_state->cur_packet) {
                                /* Duplicate "packet begin" notification */
-                               is_valid = false;
+                               is_valid = BT_FALSE;
                                goto end;
                        }
                        break;
@@ -542,18 +637,32 @@ end:
        return is_valid;
 }
 
+static
+bt_bool is_subscribed_to_notification_type(struct bt_notification_iterator *iterator,
+               enum bt_notification_type notif_type)
+{
+       uint32_t iter_notif_type =
+               (uint32_t) bt_notification_iterator_notif_type_from_notif_type(
+                       notif_type);
+
+       return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE;
+}
+
 static
 void add_action_push_notif(struct bt_notification_iterator *iterator,
                struct bt_notification *notif)
 {
        struct action action = {
                .type = ACTION_TYPE_PUSH_NOTIF,
-               .payload.push_notif = {
-                       .notif = bt_get(notif),
-               },
        };
 
        assert(notif);
+
+       if (!is_subscribed_to_notification_type(iterator, notif->type)) {
+               return;
+       }
+
+       action.payload.push_notif.notif = bt_get(notif);
        add_action(iterator, &action);
 }
 
@@ -565,6 +674,11 @@ int add_action_push_notif_stream_begin(
        int ret = 0;
        struct bt_notification *stream_begin_notif = NULL;
 
+       if (!is_subscribed_to_notification_type(iterator,
+                       BT_NOTIFICATION_TYPE_STREAM_BEGIN)) {
+               goto end;
+       }
+
        assert(stream);
        stream_begin_notif = bt_notification_stream_begin_create(stream);
        if (!stream_begin_notif) {
@@ -590,6 +704,11 @@ int add_action_push_notif_stream_end(
        int ret = 0;
        struct bt_notification *stream_end_notif = NULL;
 
+       if (!is_subscribed_to_notification_type(iterator,
+                       BT_NOTIFICATION_TYPE_STREAM_END)) {
+               goto end;
+       }
+
        assert(stream);
        stream_end_notif = bt_notification_stream_end_create(stream);
        if (!stream_end_notif) {
@@ -615,6 +734,11 @@ int add_action_push_notif_packet_begin(
        int ret = 0;
        struct bt_notification *packet_begin_notif = NULL;
 
+       if (!is_subscribed_to_notification_type(iterator,
+                       BT_NOTIFICATION_TYPE_PACKET_BEGIN)) {
+               goto end;
+       }
+
        assert(packet);
        packet_begin_notif = bt_notification_packet_begin_create(packet);
        if (!packet_begin_notif) {
@@ -640,6 +764,11 @@ int add_action_push_notif_packet_end(
        int ret = 0;
        struct bt_notification *packet_end_notif = NULL;
 
+       if (!is_subscribed_to_notification_type(iterator,
+                       BT_NOTIFICATION_TYPE_PACKET_END)) {
+               goto end;
+       }
+
        assert(packet);
        packet_end_notif = bt_notification_packet_end_create(packet);
        if (!packet_end_notif) {
@@ -967,6 +1096,9 @@ int enqueue_notification_and_automatic(
 
        assert(notif);
 
+       // TODO: Skip most of this if the iterator is only subscribed
+       //       to event/inactivity notifications.
+
        /* Get the stream and packet referred by the notification */
        switch (notif->type) {
        case BT_NOTIFICATION_TYPE_EVENT:
@@ -995,7 +1127,7 @@ int enqueue_notification_and_automatic(
                break;
        case BT_NOTIFICATION_TYPE_INACTIVITY:
                /* Always valid */
-               break;
+               goto handle_notif;
        default:
                /*
                 * Invalid type of notification. Only the notification
@@ -1023,6 +1155,7 @@ int enqueue_notification_and_automatic(
                goto error;
        }
 
+handle_notif:
        switch (notif->type) {
        case BT_NOTIFICATION_TYPE_EVENT:
                ret = handle_notif_event(iterator, notif, notif_stream,
@@ -1042,6 +1175,9 @@ int enqueue_notification_and_automatic(
                ret = handle_notif_packet_end(iterator, notif, notif_stream,
                        notif_packet);
                break;
+       case BT_NOTIFICATION_TYPE_INACTIVITY:
+               add_action_push_notif(iterator, notif);
+               break;
        default:
                break;
        }
@@ -1159,61 +1295,63 @@ enum bt_notification_iterator_status ensure_queue_has_notifications(
                break;
        }
        default:
-               assert(false);
+               assert(BT_FALSE);
                break;
        }
 
        /*
         * Call the user's "next" method to get the next notification
-        * and status, skipping the forwarded automatic notifications
-        * if any.
+        * and status.
         */
        assert(next_method);
-       next_return = next_method(priv_iterator);
-       if (next_return.status < 0) {
-               status = next_return.status;
-               goto end;
-       }
 
-       switch (next_return.status) {
-       case BT_NOTIFICATION_ITERATOR_STATUS_END:
-               ret = handle_end(iterator);
-               if (ret) {
-                       status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
+       while (iterator->queue->length == 0) {
+               next_return = next_method(priv_iterator);
+               if (next_return.status < 0) {
+                       status = next_return.status;
                        goto end;
                }
 
-               if (iterator->queue->length == 0) {
-                       status = BT_NOTIFICATION_ITERATOR_STATUS_END;
-               }
+               switch (next_return.status) {
+               case BT_NOTIFICATION_ITERATOR_STATUS_END:
+                       ret = handle_end(iterator);
+                       if (ret) {
+                               status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
+                               goto end;
+                       }
 
-               iterator->is_ended = true;
-               break;
-       case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
-               status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
-               break;
-       case BT_NOTIFICATION_ITERATOR_STATUS_OK:
-               if (!next_return.notification) {
-                       status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
-                       goto end;
-               }
+                       if (iterator->queue->length == 0) {
+                               status = BT_NOTIFICATION_ITERATOR_STATUS_END;
+                       }
 
-               /*
-                * We know the notification is valid. Before we push it
-                * to the head of the queue, push the appropriate
-                * automatic notifications if any.
-                */
-               ret = enqueue_notification_and_automatic(iterator,
-                       next_return.notification);
-               BT_PUT(next_return.notification);
-               if (ret) {
-                       status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
+                       iterator->is_ended = BT_TRUE;
+                       goto end;
+               case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
+                       status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
                        goto end;
+               case BT_NOTIFICATION_ITERATOR_STATUS_OK:
+                       if (!next_return.notification) {
+                               status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
+                               goto end;
+                       }
+
+                       /*
+                        * We know the notification is valid. Before we
+                        * push it to the head of the queue, push the
+                        * appropriate automatic notifications if any.
+                        */
+                       ret = enqueue_notification_and_automatic(iterator,
+                               next_return.notification);
+                       BT_PUT(next_return.notification);
+                       if (ret) {
+                               status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
+                               goto end;
+                       }
+                       break;
+               default:
+                       /* Unknown non-error status */
+                       assert(BT_FALSE);
                }
-               break;
-       default:
-               /* Unknown non-error status */
-               assert(false);
        }
 
 end:
This page took 0.028798 seconds and 4 git commands to generate.