Support for ControlMessage as an output type for HttpServerSourceStage and HttpClientSourceStage

python/morpheus/morpheus/_lib/common/__init__.pyi

@@ -110,7 +110,7 @@ class FilterSource():
     __members__: dict # value = {'Auto': <FilterSource.Auto: 0>, 'TENSOR': <FilterSource.TENSOR: 1>, 'DATAFRAME': <FilterSource.DATAFRAME: 2>}
     pass
 class HttpEndpoint():
-    def __init__(self, py_parse_fn: function, url: str, method: str) -> None: ...
+    def __init__(self, py_parse_fn: function, url: str, method: str, include_headers: bool = False) -> None: ...
     pass
 class HttpServer():
     def __enter__(self) -> HttpServer: ...

python/morpheus/morpheus/_lib/common/module.cpp

@@ -153,7 +153,11 @@ PYBIND11_MODULE(common, _module)
         .value("DATAFRAME", FilterSource::DATAFRAME);
 
     py::class_<HttpEndpoint, std::shared_ptr<HttpEndpoint>>(_module, "HttpEndpoint")
-        .def(py::init<>(&HttpEndpointInterfaceProxy::init), py::arg("py_parse_fn"), py::arg("url"), py::arg("method"));
+        .def(py::init<>(&HttpEndpointInterfaceProxy::init),
+             py::arg("py_parse_fn"),
+             py::arg("url"),
+             py::arg("method"),
+             py::arg("include_headers") = false);
 
     py::class_<HttpServer, std::shared_ptr<HttpServer>>(_module, "HttpServer")
         .def(py::init<>(&HttpServerInterfaceProxy::init),

python/morpheus/morpheus/_lib/include/morpheus/messages/control.hpp

@@ -39,128 +39,6 @@ enum class MORPHEUS_EXPORT ControlMessageType
     TRAINING
 };
 
-// class PayloadManager
-// {
-//   public:
-//     /**
-//      * @brief Get the tensor object identified by `name`
-//      *
-//      * @param name
-//      * @return TensorObject&
-//      * @throws std::runtime_error If no tensor matching `name` exists
-//      */
-//     TensorObject& get_tensor(const std::string& name)
-//     {
-//         return m_tensors->get_tensor(name);
-//     }
-
-//     /**
-//      * @brief Get the tensor object identified by `name`
-//      *
-//      * @param name
-//      * @return const TensorObject&
-//      * @throws std::runtime_error If no tensor matching `name` exists
-//      */
-//     const TensorObject& get_tensor(const std::string& name) const
-//     {
-//         return m_tensors->get_tensor(name);
-//     }
-
-//     /**
-//      * @brief Set the tensor object identified by `name`
-//      *
-//      * @param name
-//      * @param tensor
-//      * @throws std::length_error If the number of rows in `tensor` does not match `count`.
-//      */
-//     void set_tensor(const std::string& name, TensorObject&& tensor)
-//     {
-//         m_tensors->set_tensor(name, std::move(tensor));
-//     }
-
-//     /**
-//      * @brief Get a reference to the internal tensors map
-//      *
-//      * @return const TensorMap&
-//      */
-//     const TensorMap& get_tensors() const
-//     {
-//         return m_tensors->get_tensors();
-//     }
-
-//     /**
-//      * @brief Set the tensors object
-//      *
-//      * @param tensors
-//      * @throws std::length_error If the number of rows in the `tensors` do not match `count`.
-//      */
-//     void set_tensors(TensorMap&& tensors)
-//     {
-//         m_tensors->set_tensors(std::move(tensors));
-//     }
-
-//         /**
-//      * @brief Get the tensor object identified by `name`
-//      *
-//      * @param name
-//      * @return TensorObject&
-//      * @throws std::runtime_error If no tensor matching `name` exists
-//      */
-//     TensorObject& get_column(const std::string& name)
-//     {
-//         return m_tensors->get_tensor(name);
-//     }
-
-//     /**
-//      * @brief Get the tensor object identified by `name`
-//      *
-//      * @param name
-//      * @return const TensorObject&
-//      * @throws std::runtime_error If no tensor matching `name` exists
-//      */
-//     const TensorObject& get_column(const std::string& name) const
-//     {
-//         return m_tensors->get_tensor(name);
-//     }
-
-//     /**
-//      * @brief Set the tensor object identified by `name`
-//      *
-//      * @param name
-//      * @param tensor
-//      * @throws std::length_error If the number of rows in `tensor` does not match `count`.
-//      */
-//     void set_column(const std::string& name, TensorObject&& tensor)
-//     {
-//         m_tensors->set_tensor(name, std::move(tensor));
-//     }
-
-//     /**
-//      * @brief Get a reference to the internal tensors map
-//      *
-//      * @return const TensorMap&
-//      */
-//     TableInfo get_columns() const
-//     {
-//         return m_df->get_info();
-//     }
-
-//     /**
-//      * @brief Set the tensors object
-//      *
-//      * @param tensors
-//      * @throws std::length_error If the number of rows in the `tensors` do not match `count`.
-//      */
-//     void set_columns(TableInfo&& tensors)
-//     {
-//         m_tensors->set_tensors(std::move(tensors));
-//     }
-
-//   private:
-//     std::shared_ptr<MessageMeta> m_df;
-//     std::shared_ptr<TensorMemory> m_tensors;
-// };
-
 class MORPHEUS_EXPORT TensorMemory;
 
 // System-clock for better compatibility with pybind11/chrono
@@ -178,6 +56,7 @@ class MORPHEUS_EXPORT ControlMessage
   public:
     ControlMessage();
     explicit ControlMessage(const morpheus::utilities::json_t& config);
+    explicit ControlMessage(const std::shared_ptr<MessageMeta>& payload, const morpheus::utilities::json_t& config);
 
     ControlMessage(const ControlMessage& other);  // Copies config and metadata, but not payload
 
@@ -387,6 +266,14 @@ struct MORPHEUS_EXPORT ControlMessageProxy
      */
     static std::shared_ptr<ControlMessage> create(pybind11::dict& config);
 
+    /**
+     * @brief Creates a new ControlMessage instance from a Python MessageMeta and a configuration dictionary.
+     * @param meta Python instance of MessageMeta
+     * @param config A pybind11::dict representing the configuration for the ControlMessage.
+     * @return A shared_ptr to a newly created ControlMessage instance.
+     */
+    static std::shared_ptr<ControlMessage> create(const pybind11::object& meta, pybind11::dict& config);
+
     /**
      * @brief Creates a new ControlMessage instance as a copy of an existing one.
      * @param other A shared_ptr to another ControlMessage instance to copy.

python/morpheus/morpheus/_lib/include/morpheus/stages/deserialize.hpp

@@ -22,6 +22,7 @@
 #include "morpheus/messages/meta.hpp"
 #include "morpheus/messages/multi.hpp"
 #include "morpheus/types.hpp"                  // for TensorIndex
+#include "morpheus/utilities/json_types.hpp"   // for control_message_task_t
 #include "morpheus/utilities/python_util.hpp"  // for show_warning_message
 #include "morpheus/utilities/string_util.hpp"  // for MORPHEUS_CONCAT_STR
 
@@ -50,18 +51,16 @@ namespace morpheus {
  * @file
  */
 
-using cm_task_t = std::pair<std::string, nlohmann::json>;
-
 void make_output_message(std::shared_ptr<MessageMeta>& incoming_message,
                          TensorIndex start,
                          TensorIndex stop,
-                         cm_task_t* task,
+                         control_message_task_t* task,
                          std::shared_ptr<MultiMessage>& windowed_message);
 
 void make_output_message(std::shared_ptr<MessageMeta>& incoming_message,
                          TensorIndex start,
                          TensorIndex stop,
-                         cm_task_t* task,
+                         control_message_task_t* task,
                          std::shared_ptr<ControlMessage>& windowed_message);
 
 /****** DeserializationStage********************************/
@@ -83,8 +82,8 @@ class MORPHEUS_EXPORT DeserializeStage
      * @param task Optional task to be added to all outgoing `ControlMessage`s, ignored when `OutputT` is `MultiMessage`
      */
     DeserializeStage(TensorIndex batch_size,
-                     bool ensure_sliceable_index     = true,
-                     std::unique_ptr<cm_task_t> task = nullptr) :
+                     bool ensure_sliceable_index                  = true,
+                     std::unique_ptr<control_message_task_t> task = nullptr) :
       base_t(base_t::op_factory_from_sub_fn(build_operator())),
       m_batch_size(batch_size),
       m_ensure_sliceable_index(ensure_sliceable_index),
@@ -95,7 +94,7 @@ class MORPHEUS_EXPORT DeserializeStage
 
     TensorIndex m_batch_size;
     bool m_ensure_sliceable_index{true};
-    std::unique_ptr<cm_task_t> m_task{nullptr};
+    std::unique_ptr<control_message_task_t> m_task{nullptr};
 };
 
 /****** DeserializationStageInterfaceProxy******************/