[C-Api] handle message callback

c/include/ml-api-service.h

@@ -53,6 +53,7 @@ typedef void *ml_service_h;
 typedef enum {
   ML_SERVICE_EVENT_UNKNOWN = 0,             /**< Unknown or invalid event type. */
   ML_SERVICE_EVENT_NEW_DATA = 1,            /**< New data is processed from machine learning service. */
+  ML_SERVICE_EVENT_MESSAGE = 2,             /**< New message from machine learning service (Since 10.1). */
 } ml_service_event_e;
 
 /**

c/src/ml-api-inference-pipeline-internal.h

@@ -98,6 +98,22 @@ typedef enum {
   ML_PIPELINE_ELEMENT_COMMON = 0xB,
 } ml_pipeline_element_e;
 
+/**
+ * @brief Callback for the message from pipeline.
+ * @param[in] type The type of pipeline message.
+ * @param[in] message The pipeline message details.
+ * @param[out] user_data User application's private data.
+ */
+typedef void (*ml_pipeline_message_cb) (const char *type, const char *message, void *user_data);
+
+/**
+ * @brief Internal data structure for the pipeline message callback, generated from nnstreamer elements.
+ */
+typedef struct {
+  ml_pipeline_message_cb cb; /**< Callback to send a message from pipeline */
+  void *user_data; /**< The user data passed when calling the callback */
+} pipeline_message_cb_s;
+
 /**
  * @brief Internal data structure for the pipeline state callback.
  */
@@ -114,6 +130,16 @@ typedef struct {
   gpointer handle; /**< pointer to resource handle */
 } pipeline_resource_s;
 
+/**
+ * @brief An information to create pipeline instance.
+ */
+typedef struct {
+  char *description;                /**< The pipeline description compatible with GStreamer gst_parse_launch(). */
+  pipeline_state_cb_s state_cb;     /**< Callback to notify the change of pipeline state. */
+  pipeline_message_cb_s message_cb; /**< Callback to send a message from pipeline. */
+  gboolean is_internal;             /**< True to ignore the permission in Tizen. */
+} ml_pipeline_preset;
+
 /**
  * @brief Internal private representation of pipeline handle.
  * @details This should not be exposed to applications
@@ -129,6 +155,7 @@ typedef struct _ml_pipeline {
   GHashTable *resources;          /**< hash table of resources to construct the pipeline */
   GHashTable *pipe_elm_type;      /**< hash table for type of pipeline element */
   pipeline_state_cb_s state_cb;   /**< Callback to notify the change of pipeline state */
+  pipeline_message_cb_s message_cb; /**< Callback to send a message from pipeline */
 } ml_pipeline;
 
 /**
@@ -192,6 +219,12 @@ int _ml_initialize_gstreamer (void);
  */
 int _ml_check_plugin_availability (const char *plugin_name, const char *element_name);
 
+/**
+ * @brief Construct the pipeline with custom options and returns the instance as a handle.
+ * This is internal function to handle various options.
+ */
+int _ml_pipeline_construct_custom (ml_pipeline_preset *preset, ml_pipeline_h *pipe);
+
 /**
  * @brief Gets the element of pipeline itself (GstElement).
  * @details With the returned reference, you can use GStreamer functions to handle the pipeline.

c/src/ml-api-inference-pipeline.c

@@ -488,6 +488,23 @@ cb_bus_sync_message (GstBus * bus, GstMessage * message, gpointer user_data)
         }
       }
       break;
+    case GST_MESSAGE_APPLICATION:
+    {
+      const GstStructure *s = gst_message_get_structure (message);
+
+      if (gst_structure_has_name (s, "nnstreamer-message")) {
+        const gchar *t = gst_structure_get_string (s, "type");
+        const gchar *m = gst_structure_get_string (s, "message");
+        const gchar *e = gst_structure_has_field (s, "element") ? gst_structure_get_string (s, "element") : "unknown";
+
+        _ml_logd (_ml_detail ("Message from '%s': %s (%s).", e, t, m));
+
+        if (pipe_h->message_cb.cb) {
+          pipe_h->message_cb.cb (t, m, pipe_h->message_cb.user_data);
+        }
+      }
+      break;
+    }
     default:
       break;
   }
@@ -974,9 +991,7 @@ create_internal_hash (ml_pipeline * pipe_h)
  * If is_internal is true, this will ignore the permission in Tizen.
  */
 static int
-construct_pipeline_internal (const char *pipeline_description,
-    ml_pipeline_state_cb cb, void *user_data, ml_pipeline_h * pipe,
-    gboolean is_internal)
+construct_pipeline_internal (ml_pipeline_preset * preset, ml_pipeline_h * pipe)
 {
   GError *err = NULL;
   GstElement *pipeline;
@@ -991,7 +1006,11 @@ construct_pipeline_internal (const char *pipeline_description,
     _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
         "ml_pipeline_construct error: parameter pipe is NULL. It should be a valid ml_pipeline_h pointer. E.g., ml_pipeline_h pipe; ml_pipeline_construct (..., &pip);");
 
-  if (!pipeline_description)
+  if (!preset)
+    _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
+        "ml_pipeline_construct error: parameter preset is NULL. It should be a valid pointer for pipeline.");
+
+  if (!preset->description)
     _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
         "ml_pipeline_construct error: parameter pipeline_description is NULL. It should be a valid string of Gstreamer/NNStreamer pipeline description.");
 
@@ -1015,8 +1034,8 @@ construct_pipeline_internal (const char *pipeline_description,
   create_internal_hash (pipe_h);
 
   /* convert predefined element and launch the pipeline */
-  status = convert_description ((ml_pipeline_h) pipe_h, pipeline_description,
-      &description, is_internal);
+  status = convert_description ((ml_pipeline_h) pipe_h, preset->description,
+      &description, preset->is_internal);
   if (status != ML_ERROR_NONE) {
     _ml_error_report_continue
         ("ml_pipeline_construct error: failed while converting pipeline description for GStreamer w/ convert_description() function, which has returned %d",
@@ -1030,7 +1049,7 @@ construct_pipeline_internal (const char *pipeline_description,
   if (!GST_IS_PIPELINE (pipeline) || err) {
     _ml_error_report
         ("ml_pipeline_construct error: gst_parse_launch cannot parse and launch the given pipeline = [%s]. The error message from gst_parse_launch is '%s'.",
-        pipeline_description, (err) ? err->message : "unknown reason");
+        preset->description, (err) ? err->message : "unknown reason");
     g_clear_error (&err);
 
     if (pipeline)
@@ -1056,11 +1075,15 @@ construct_pipeline_internal (const char *pipeline_description,
       G_CALLBACK (cb_bus_sync_message), pipe_h);
 
   /* state change callback */
-  pipe_h->state_cb.cb = cb;
-  pipe_h->state_cb.user_data = user_data;
+  pipe_h->state_cb.cb = preset->state_cb.cb;
+  pipe_h->state_cb.user_data = preset->state_cb.user_data;
+
+  /* message callback */
+  pipe_h->message_cb.cb = preset->message_cb.cb;
+  pipe_h->message_cb.user_data = preset->message_cb.user_data;
 
   /* iterate elements and prepare element handle */
-  status = iterate_element (pipe_h, pipeline, is_internal);
+  status = iterate_element (pipe_h, pipeline, preset->is_internal);
   if (status != ML_ERROR_NONE) {
     _ml_error_report_continue ("ml_pipeline_construct error: ...");
     goto failed;
@@ -1102,9 +1125,15 @@ int
 ml_pipeline_construct (const char *pipeline_description,
     ml_pipeline_state_cb cb, void *user_data, ml_pipeline_h * pipe)
 {
+  ml_pipeline_preset preset = { 0, };
+
+  preset.description = (char *) pipeline_description;
+  preset.state_cb.cb = cb;
+  preset.state_cb.user_data = user_data;
+  preset.is_internal = FALSE;
+
   /* not an internal pipeline construction */
-  return construct_pipeline_internal (pipeline_description, cb, user_data, pipe,
-      FALSE);
+  return construct_pipeline_internal (&preset, pipe);
 }
 
 #if defined (__TIZEN__)
@@ -1115,12 +1144,28 @@ int
 ml_pipeline_construct_internal (const char *pipeline_description,
     ml_pipeline_state_cb cb, void *user_data, ml_pipeline_h * pipe)
 {
+  ml_pipeline_preset preset = { 0, };
+
+  preset.description = (char *) pipeline_description;
+  preset.state_cb.cb = cb;
+  preset.state_cb.user_data = user_data;
+  preset.is_internal = TRUE;
+
   /* Tizen internal pipeline construction */
-  return construct_pipeline_internal (pipeline_description, cb, user_data, pipe,
-      TRUE);
+  return construct_pipeline_internal (&preset, pipe);
 }
 #endif /* __TIZEN__ */
 
+/**
+ * @brief Construct the pipeline with custom options and returns the instance as a handle.
+ * This is internal function to handle various options.
+ */
+int
+_ml_pipeline_construct_custom (ml_pipeline_preset * preset, ml_pipeline_h * pipe)
+{
+  return construct_pipeline_internal (preset, pipe);
+}
+
 /**
  * @brief Destroy the pipeline (more info in nnstreamer.h)
  */
@@ -1142,6 +1187,7 @@ ml_pipeline_destroy (ml_pipeline_h pipe)
 
   /* Before changing the state, remove all callbacks. */
   p->state_cb.cb = NULL;
+  p->message_cb.cb = NULL;
 
   /* Destroy registered callback handles and resources */
   g_hash_table_destroy (p->namednodes);

c/src/ml-api-service-extension.c

@@ -67,6 +67,17 @@ typedef struct
   GHashTable *node_table;
 } ml_extension_s;
 
+/**
+ * @brief Internal function for message callback.
+ */
+static void
+_ml_extension_message_cb (const char *type, const char *message, void *user_data)
+{
+  ml_service_s *mls = (ml_service_s *) user_data;
+
+  _ml_service_invoke_event_message (mls, type, message);
+}
+
 /**
  * @brief Internal function to handle the asynchronous invoke.
  */
@@ -171,6 +182,7 @@ _ml_extension_msg_thread (gpointer data)
 {
   ml_service_s *mls = (ml_service_s *) data;
   ml_extension_s *ext = (ml_extension_s *) mls->priv;
+  gchar *errmsg;
   int status;
 
   g_mutex_lock (&mls->lock);
@@ -193,8 +205,11 @@ _ml_extension_msg_thread (gpointer data)
           if (status == ML_ERROR_NONE) {
             _ml_service_invoke_event_new_data (mls, NULL, msg->output);
           } else {
-            _ml_error_report
-                ("Failed to invoke the model in ml-service extension thread.");
+            errmsg = g_strdup ("Failed to invoke the model (single-shot) in ml-service extension thread.");
+
+            _ml_error_report ("%s", errmsg);
+            _ml_service_invoke_event_message (mls, "invoke-failure", errmsg);
+            g_free (errmsg);
           }
           break;
         }
@@ -211,13 +226,18 @@ _ml_extension_msg_thread (gpointer data)
             msg->input = NULL;
 
             if (status != ML_ERROR_NONE) {
-              _ml_error_report
-                  ("Failed to push input data into the pipeline in ml-service extension thread.");
+              errmsg = g_strdup_printf ("Failed to push input data into the input node '%s' in ml-service extension thread.", msg->name);
+
+              _ml_error_report ("%s", errmsg);
+              _ml_service_invoke_event_message (mls, "push-failure", errmsg);
+              g_free (errmsg);
             }
           } else {
-            _ml_error_report
-                ("Failed to push input data into the pipeline, cannot find input node '%s'.",
-                msg->name);
+            errmsg = g_strdup_printf ("Failed to push input data into the pipeline, cannot find input node '%s'.", msg->name);
+
+            _ml_error_report ("%s", errmsg);
+            _ml_service_invoke_event_message (mls, "push-failure", errmsg);
+            g_free (errmsg);
           }
           break;
         }
@@ -509,6 +529,7 @@ _ml_extension_conf_parse_pipeline_node (ml_service_s * mls, JsonNode * node,
 static int
 _ml_extension_conf_parse_pipeline (ml_service_s * mls, JsonObject * pipe)
 {
+  ml_pipeline_preset preset = { 0, };
   ml_extension_s *ext = (ml_extension_s *) mls->priv;
   g_autofree gchar *desc = NULL;
   int status;
@@ -535,7 +556,12 @@ _ml_extension_conf_parse_pipeline (ml_service_s * mls, JsonObject * pipe)
         "Failed to parse configuration file, cannot get the pipeline description.");
   }
 
-  status = ml_pipeline_construct (desc, NULL, NULL, &ext->pipeline);
+  preset.description = desc;
+  preset.message_cb.cb = _ml_extension_message_cb;
+  preset.message_cb.user_data = mls;
+  preset.is_internal = FALSE;
+
+  status = _ml_pipeline_construct_custom (&preset, &ext->pipeline);
   if (status != ML_ERROR_NONE) {
     _ml_error_report_return (status,
         "Failed to parse configuration file, cannot construct the pipeline.");

c/src/ml-api-service-private.h

@@ -19,6 +19,8 @@
 
 #include <ml-api-service.h>
 #include <ml-api-inference-internal.h>
+#include <ml-api-inference-pipeline-internal.h>
+#include <ml-api-inference-single-internal.h>
 #include <mlops-agent-interface.h>
 
 #ifdef __cplusplus
@@ -149,7 +151,12 @@ const gchar * _ml_service_get_json_string_member (JsonObject *object, const gcha
 /**
  * @brief Generating an ML service event and passing received data and event to a registered callback function.
  */
-int _ml_service_invoke_event_new_data (ml_service_s * mls, const char *name, const ml_tensors_data_h data);
+int _ml_service_invoke_event_new_data (ml_service_s *mls, const char *name, const ml_tensors_data_h data);
+
+/**
+ * @brief Generating an ML service event and passing a message to a registered callback function.
+ */
+int _ml_service_invoke_event_message (ml_service_s *mls, const char *type, const char *message);
 
 /**
  * @brief Callback for sink node in pipeline description.

c/src/ml-api-service.c

@@ -898,6 +898,53 @@ _ml_service_invoke_event_new_data (ml_service_s * mls, const char *name,
   return status;
 }
 
+/**
+ * @brief Generating an ML service event and passing a message to
+ * a registered callback function.
+ */
+int
+_ml_service_invoke_event_message (ml_service_s * mls, const char *type,
+    const char *message)
+{
+  ml_service_event_cb_info_s cb_info = { 0 };
+  ml_information_h ml_info = NULL;
+  int status = ML_ERROR_NONE;
+
+  if (!mls || !type || !message) {
+    _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
+        "Failed to create ml-service event data, invalid parameter.");
+  }
+
+  _ml_service_get_event_cb_info (mls, &cb_info);
+
+  if (cb_info.cb) {
+    /* Create information handle for ml-service event. */
+    status = _ml_information_create (&ml_info);
+    if (status != ML_ERROR_NONE)
+      goto done;
+
+    status = _ml_information_set (ml_info, "type", (void *) type, NULL);
+    if (status != ML_ERROR_NONE)
+      goto done;
+
+    status = _ml_information_set (ml_info, "message", (void *) message, NULL);
+    if (status != ML_ERROR_NONE)
+      goto done;
+
+    cb_info.cb (ML_SERVICE_EVENT_MESSAGE, ml_info, cb_info.pdata);
+  }
+
+done:
+  if (ml_info)
+    ml_information_destroy (ml_info);
+
+  if (status != ML_ERROR_NONE) {
+    _ml_error_report ("Failed to invoke 'message' event.");
+  }
+
+  return status;
+}
+
 /**
  * @brief Callback for sink node in pipeline description.
  * Processes incoming data from pipeline sink element and forwards it to

tests/capi/unittest_capi_service_extension.cc

@@ -274,6 +274,20 @@ _extension_test_imgclf_cb (ml_service_event_e event, ml_information_h event_data
       if (tdata)
         tdata->received++;
       break;
+    case ML_SERVICE_EVENT_MESSAGE:
+      {
+        gchar *type = NULL;
+        gchar *msg = NULL;
+
+        /* Error case if failed to invoke a model. */
+        status = ml_information_get (event_data, "type", (void **) (&type));
+        EXPECT_EQ (status, ML_ERROR_NONE);
+        status = ml_information_get (event_data, "message", (void **) (&msg));
+        EXPECT_EQ (status, ML_ERROR_NONE);
+
+        _ml_logd ("[DEBUG] Message received: %s (%s)", type, msg);
+        break;
+      }
     default:
       break;
   }