C++ Data based Pipeline

Question

I'm currently rewriting a legacy program for Data Processing. So in general i have an Input Source (could be for example the Filesystem or Data that comes in via HTTP) then the Data should be further processed (for example wrapped into a json, adding some metadata to it) and in the end the data could be for example stored on the Disk or Send to other systems.

I started to think about some kind of Pipe and Filters Mechanism and started to develop a kind of Pipeline Architecture with commands that could be sticked together and then be executed.

For archiving a loose coupling of the Components of my application i used the Poco::NotificationQueue to communicate between the Input Source and the Command Pipeline.

My Code looks like this:

Core Library

In the Core Library i define the Message Class and a Wrapper around the Poco::NotificationQueue.

Message.hpp/cpp

#pragma once
#include <Poco/Notification.h>

#include <string>
#include <memory>
#include <unordered_map>
#include <optional>

#include "export.hpp"

namespace core {

  class CORE_EXPORT Message : public Poco::Notification {
  public:

    using Metadata = std::unordered_map<std::string, std::string>;

    explicit Message(const std::string& id);
    Message(const std::string& id, const std::string& content);
    ~Message();

    std::string content() const {
      return content_;
    }

    std::string id() const {
      return id_;
    }

    void content(const std::string& content) {
      content_ = content;
    }

    void id(const std::string& id) {
      id_ = id;
    }

    size_t metadataSize() const {
      return metadata_.size();
    }

    std::optional<std::string> getMetadata(const std::string& key);

    Metadata getAllMetadata() const {
      return metadata_;
    }

    void addMetadata(const std::string& key, const std::string& value);

    friend CORE_EXPORT bool operator==(const Message& l_message, const Message& r_message);
    friend CORE_EXPORT bool operator!=(const Message& l_message, const Message& r_message);

  private:
    std::string id_;
    std::string content_;
    Metadata metadata_;
  };

  using MessagePtr = std::unique_ptr<Message>;
  using MessageRawPtr = Message*;
}  // namespace Core

# Message.cpp
#include "Core/Message.hpp"

namespace core {

  Message::Message(const std::string& id) : Message(id, {}) {
  }

  Message::Message(const std::string& id, const std::string& content) : id_(id), content_(content) {}

  Message::~Message() = default;

  std::optional<std::string> Message::getMetadata(const std::string& key) {
    auto tmp = metadata_.find(key);
    return tmp == metadata_.end() ? std::nullopt : std::optional<std::string>{ tmp->second };
  }

  void Message::addMetadata(const std::string& key, const std::string& value) {
    metadata_.emplace(key, value);
  }

  bool operator==(const Message& l_message, const Message& r_message) {
    return l_message.id_ == r_message.id_;
  }

  bool operator!=(const Message& l_message, const Message& r_message) {
    return !(l_message == r_message);
  }
}

MessageQueue.hpp/MessageQueueImpl.hpp/cpp

#pragma once
#include <memory>
#include "export.hpp"

namespace core {
  class Message;

  class CORE_EXPORT MessageQueue {

  public:
    virtual void enqueue(std::unique_ptr<Message> message) = 0;
    virtual std::unique_ptr<Message> dequeue() = 0;
    virtual int size() const = 0;
    virtual void close() = 0;
  };

  using MessageQueuePtr = std::unique_ptr<MessageQueue>;
  using MessageQueueRawPtr = MessageQueue*;
}

//MessageQueueImpl.hpp
#pragma once
#include <memory>
#include "export.hpp"
#include "MessageQueue.hpp"

namespace core {
  class Message;

  class CORE_EXPORT MessageQueueImpl : public MessageQueue {
    class MessageQueuePimpl;
  public:
    MessageQueueImpl();
    virtual ~MessageQueueImpl();
    void enqueue(std::unique_ptr<Message> message) override;
    std::unique_ptr<Message> dequeue() override;
    int size() const override;
    void close() override;

  private:
    std::unique_ptr<MessageQueuePimpl> pimpl_;
  };

}

//MessageQueueImpl.cpp
#include "Core/MessageQueueImpl.hpp"
#include <Poco/NotificationQueue.h>
#include <Core/Message.hpp>

namespace core {
  class MessageQueueImpl::MessageQueuePimpl {
  public:
    void enqueue(std::unique_ptr<Message> message);
    std::unique_ptr<Message> dequeue();
    int size() const;
    void close();
  private:
    Poco::NotificationQueue queue_;
  };


  void MessageQueueImpl::MessageQueuePimpl::enqueue(std::unique_ptr<Message> message) {
    auto raw_message = message.release();
    queue_.enqueueNotification(raw_message);
  }

  std::unique_ptr<Message> MessageQueueImpl::MessageQueuePimpl::dequeue() {
    auto notification = queue_.waitDequeueNotification();
    return std::unique_ptr<Message>(dynamic_cast<Message*>(notification));
  }

  int MessageQueueImpl::MessageQueuePimpl::size() const {
    return queue_.size();
  }

  void MessageQueueImpl::MessageQueuePimpl::close() {
    queue_.wakeUpAll();
  }


  MessageQueueImpl::MessageQueueImpl() : pimpl_(std::make_unique<MessageQueuePimpl>()) {
  }

  MessageQueueImpl::~MessageQueueImpl() = default;

  void MessageQueueImpl::enqueue(std::unique_ptr<Message> message) {
    pimpl_->enqueue(std::move(message));
  }

  std::unique_ptr<Message> MessageQueueImpl::dequeue() {
    return pimpl_->dequeue();
  }

  int MessageQueueImpl::size() const {
    return pimpl_->size();
  }

  void MessageQueueImpl::close() {
    pimpl_->close();
  }
}

Commands Library

The Commands Library provides the Interface for the Commands and also a CommandFactory to register/deregister Commands. The idea is to register a set of predefined Commands during the Start of the Program and later also be able to register commands that where added to the program as a kind of plugins. For that i used the Prototype Pattern in the Commands. I took parts of this from the Book Practical C++ Design: From Programming to Architecture from Adam Singer which i find is a great book.

Command.hpp/cpp

#pragma once
#include <chrono>
#include <memory>
#include <string>
#include <unordered_map>
#include "export.hpp"


namespace core {
  class Message;
}

namespace commands {

  using commandsConfig = std::unordered_map<std::string, std::string>;

  class COMMANDS_EXPORT Command {
  public:
    static std::string version();
    explicit Command(const std::string& name = "");
    virtual ~Command();
    std::unique_ptr<core::Message> execute(std::unique_ptr<core::Message> message);
    Command* clone(commandsConfig config);

    std::string name() const;
    size_t messageCount() const;
    double totalDuration() const;

  protected:
    void incMessageCount();

  private:
    virtual std::unique_ptr<core::Message> executeImpl(std::unique_ptr<core::Message>  message) = 0;
    virtual void checkPreconditionsImpl(core::Message* message) = 0;
    virtual Command* cloneImpl() = 0;
    virtual void configureImpl(commandsConfig config) = 0;

    std::string name_{};
    size_t message_count_{ 0 };
    std::chrono::duration<double> total_duration_{};
  };

  using CommandPtr = std::unique_ptr<Command>;
  using CommandRawPtr = Command*;

  template<typename T, typename... Args>
  auto MakeCommandPtr(Args&&... args) {
    return CommandPtr{ new T{std::forward<Args>(args)...} };
  }

  inline auto MakeCommandPtr(Command* p) {
    return CommandPtr{ p };
  }
}

// Command.cpp
#include <Commands/Command.hpp>
#include <Commands/VersionInfo_Commands.h>

#include <Core/Message.hpp>

namespace commands {

  std::string Command::version() {
    return std::string{PRODUCT_VERSION_FULL_STR};
  }

  Command::Command(const std::string& name) : name_(name) {
  }

  Command::~Command() = default;

  core::MessagePtr Command::execute(core::MessagePtr message) {
    auto command_execution_duration_begin = std::chrono::steady_clock::now();
    checkPreconditionsImpl(message.get());
    auto result = executeImpl(std::move(message));
    total_duration_ = std::chrono::steady_clock::now() - command_execution_duration_begin;
    return result;
  }

  Command* Command::clone(commandsConfig config) {
    std::unique_ptr<Command> new_cmd(cloneImpl());
    new_cmd->configureImpl(config);
    return new_cmd.release();
  }

  std::string Command::name() const {
    return name_;
  }

  size_t Command::messageCount() const {
    return message_count_;
  }

  double Command::totalDuration() const {
    return total_duration_.count();
  }

  void Command::incMessageCount() {
    message_count_++;
  }
}

CommandFactory.hpp/cpp

#pragma once
#include "Command.hpp"
#include "export.hpp"
#include <set>

namespace commands {

class COMMANDS_EXPORT CommandFactory {

public:
  static CommandFactory& Instance();

  void registerCommand(const std::string& name, CommandPtr command);
  CommandPtr deregisterCommand(const std::string& name);
  CommandPtr allocateCommand(const std::string& name, commandsConfig config) const;

  size_t getNumberOfCommands() const;
  std::set<std::string> getAllCommandNames() const;
  bool hasKey(const std::string& name) const;
  void clearAllCommands();

private:
  CommandFactory() = default;
  ~CommandFactory() = default;

  CommandFactory(CommandFactory&) = delete;
  CommandFactory(CommandFactory&&) = delete;
  CommandFactory& operator=(CommandFactory&) = delete;
  CommandFactory& operator=(CommandFactory&&) = delete;

  using Factory = std::unordered_map<std::string, commands::CommandPtr>;
  Factory factory_;

};

}

// CommandFactory.cpp
#include "Commands/CommandFactory.hpp"
#include <sstream>

namespace commands {
  CommandFactory& CommandFactory::Instance() {
    static CommandFactory instance;
    return instance;
  }

  void CommandFactory::registerCommand(const std::string& name, CommandPtr command) {
    if(hasKey(name)) {
      std::ostringstream oss;
      oss << "Command " << name << " already registered";
      throw std::runtime_error{oss.str()};
    }
    factory_.emplace(name, std::move(command));
  }

  CommandPtr CommandFactory::deregisterCommand(const std::string& name) {
    if(hasKey(name)) {
      auto it = factory_.find(name);
      auto tmp = MakeCommandPtr(it->second.release());
      factory_.erase(it);
      return tmp;
    }
    return MakeCommandPtr(nullptr);
  }

  CommandPtr CommandFactory::allocateCommand(const std::string& name, commandsConfig config) const {
    if (hasKey(name)) {
      const auto& command = factory_.find(name)->second;
      return MakeCommandPtr(command->clone(config));
    }
    return MakeCommandPtr(nullptr);
  }

  size_t CommandFactory::getNumberOfCommands() const {
    return factory_.size();
  }

  std::set<std::string> CommandFactory::getAllCommandNames() const {
    std::set<std::string> tmp;
    for(auto& pair : factory_) {
      tmp.insert(pair.first);
    }
    return tmp;
  }

  bool CommandFactory::hasKey(const std::string& name) const {
    return factory_.find(name) != factory_.end();
  }

  void CommandFactory::clearAllCommands() {
    factory_.clear();
  }
}

Concrete Commands

The Concrete Commands are developed in seperate Libraries. Here is an example of a Concrete Command:

WrapCommand.hpp/cpp

#pragma once
#include "export.hpp"
#include <Commands/Command.hpp>

namespace commands {
  class WRAPCOMMAND_EXPORT WrapCommand : public Command {
  public:

    using TransformAction = void(WrapCommand::*)(core::Message*);
    using TransformActions = std::unordered_map<std::string, TransformAction>;

    static std::string version();
    // Config keys

    // specifies the format type of the wrapped message
    //   e.g. 'json' or 'xml'
    inline static const std::string CFG_CMD_WRAP_TYPE{"type"};
    // specifies the node name of the message content
    //   e.g. 'content' or 'data'
    inline static const std::string CFG_CMD_WRAP_ELEMENT{"element"};
    // specifies if metadata is added to the wrapped message
    //   'true' = metadata is added
    //   'false' = metadata is not added
    inline static const std::string CFG_CMD_WRAP_METADATA{"metadata"};

    explicit WrapCommand(const std::string& name = "WrapCommand");
    
  private:
    WrapCommand(WrapCommand&&) = delete;
    WrapCommand(const WrapCommand&) = default;
    WrapCommand& operator=(const WrapCommand&) = delete;
    WrapCommand& operator=(WrapCommand&&) = delete;
    void checkPreconditionsImpl(core::Message* message) override;
    WrapCommand* cloneImpl() override;
    void configureImpl(commandsConfig config) override;
    std::unique_ptr<core::Message> executeImpl(std::unique_ptr<core::Message> message) override;

    void wrapInJson(core::Message*);
    void checkConfigHasKey(const std::string& key);
    void validateConfig();

    commandsConfig config_{};
    TransformActions actions_{
      {"json", &WrapCommand::wrapInJson}
    };
    bool add_metadata_{ false };
  };

}  // namespace commands

// Wrapcommand.cpp
#include "WrapCommand/WrapCommand.hpp"
#include "WrapCommand/VersionInfo_WrapCommand.h"
#include <nlohmann/json.hpp>
#include <Core/Message.hpp>

using json = nlohmann::ordered_json;

namespace commands {
  std::string WrapCommand::version() {
    return std::string{ PRODUCT_VERSION_FULL_STR };
  }

  WrapCommand::WrapCommand(const std::string& name) : Command(name) {
  }

  void WrapCommand::checkPreconditionsImpl(core::MessageRawPtr message) {
    validateConfig();

  }

  WrapCommand* WrapCommand::cloneImpl() {
    return new WrapCommand{ *this };
  }

  void WrapCommand::configureImpl(commandsConfig config) {
    config_ = config;
  }

  core::MessagePtr WrapCommand::executeImpl(core::MessagePtr message) {
    (this->*actions_[config_[CFG_CMD_WRAP_TYPE]])(message.get());
    incMessageCount();
    return message;
  }

  void WrapCommand::wrapInJson(core::MessageRawPtr message) {
    json new_content;

    if (add_metadata_) {
      for (auto& [key, value] : message->getAllMetadata()) {
        new_content[key] = value;
      }
    }

    std::string content = message->content();
    new_content[config_[CFG_CMD_WRAP_ELEMENT]] = content;
    
    message->content(to_string(new_content));
  }

  void WrapCommand::checkConfigHasKey(const std::string& key) {
    if (config_.find(key) == config_.end()) {
      throw std::invalid_argument("Argument '" + key + "' is missing");
    }
  }

  void WrapCommand::validateConfig() {
    checkConfigHasKey(CFG_CMD_WRAP_TYPE);
    if (actions_.find(config_[CFG_CMD_WRAP_TYPE]) == actions_.end()) {
      throw std::invalid_argument("Parameter '" + CFG_CMD_WRAP_TYPE + "' = " + config_[CFG_CMD_WRAP_TYPE] + " is not supported");
    }

    checkConfigHasKey(CFG_CMD_WRAP_ELEMENT);

    checkConfigHasKey(CFG_CMD_WRAP_METADATA);
    if (config_[CFG_CMD_WRAP_METADATA] == "true") {
      add_metadata_ = true;
    } else if (config_[CFG_CMD_WRAP_METADATA] == "false") {
      add_metadata_ = false;
    } else {
      throw std::invalid_argument("Parameter '" + CFG_CMD_WRAP_METADATA + "' = " + config_[CFG_CMD_WRAP_METADATA] + " is invalid. Value must be 'true' or 'false'");
    }
  }
}  // namespace commands

Pipeline Library

The Pipeline itself should have a list of commands that it needs to execute for every message that was added to the Queue. For each message it executes all the commands.

Pipeline.hpp/cpp

#pragma once
#include <memory>
#include <string>
#include <unordered_map>
#include <variant>
#include <vector>

#include "export.hpp"
#include "Commands/Command.hpp"
#include "Core/MessageQueue.hpp"

namespace pipeline {

  std::string PIPELINE_EXPORT version();

  class PIPELINE_EXPORT Pipeline {
    class PipelineImpl;
  public:
    Pipeline(const std::string& name, std::vector<commands::CommandPtr> commands, commands::CommandPtr error_command);
    virtual ~Pipeline();
    void queue(core::MessageQueuePtr queue);
    void start();
    void stop();
    [[nodiscard]] std::string name() const;
    [[nodiscard]] size_t workerSize() const;
    [[nodiscard]] size_t availableWorkers() const;
    [[nodiscard]] size_t usedWorkers() const;

  private:
    std::unique_ptr<PipelineImpl> pimpl_;
  };


}

//Pipeline.cpp

#include <Pipeline/Pipeline.hpp>

#include <Poco/ThreadPool.h>
#include <Poco/RunnableAdapter.h>

#include <FilesystemInput/FilesystemInput.hpp>
#include <Commands/Command.hpp>
#include <PipelineWorker.hpp>
#include <Core/Message.hpp>
#include "Core/MessageQueueImpl.hpp"
#include "Pipeline/VersionInfo_Pipeline.h"

namespace pipeline {

  std::string version() {
    return std::string{ PRODUCT_VERSION_FULL_STR };
  }

class Pipeline::PipelineImpl {
public:
  PipelineImpl(const std::string& name, std::vector<commands::CommandPtr> commands, commands::CommandPtr error_command);

  void start();
  void stop();
  void queue(core::MessageQueuePtr queue);
  std::string name() const;
  size_t workerSize() const;
  size_t availableWorkers() const;
  size_t usedWorkers() const;

private:
  void internalCallback(const std::string& id, const std::string& content);

  std::vector<commands::CommandRawPtr> commandsRawPtr() const;

  std::string name_;
  std::unique_ptr<filesysteminput::FilesystemInput> fsi_;
  core::MessageQueuePtr queue_ = std::make_unique<core::MessageQueueImpl>();
  std::vector<commands::CommandPtr> commands_;
  commands::CommandPtr error_command_;
  std::vector<std::unique_ptr<PipelineWorker>> workers_{};
  std::vector<Poco::RunnableAdapter<PipelineWorker>> worker_adapters_{};
  Poco::ThreadPool worker_pool_{ 2, 16 };
};

Pipeline::PipelineImpl::PipelineImpl(const std::string& name, std::vector<commands::CommandPtr> commands, commands::CommandPtr error_command) :
name_(name),
fsi_(std::make_unique<filesysteminput::FilesystemInput>([this](const std::string& id, const std::string& content) {this->internalCallback(id, content); })),
commands_(std::move(commands)),
error_command_(std::move(error_command)) {
}

void Pipeline::PipelineImpl::start() {
  fsi_->start();

  for (int i = 0; i < worker_pool_.capacity(); ++i) {
    std::string name{ "PipelineWorker_" + std::to_string(i) };
    workers_.emplace_back(std::make_unique<PipelineWorker>(name, queue_.get(), commandsRawPtr(), error_command_.get()));
    worker_adapters_.emplace_back( *workers_[i], &PipelineWorker::run );
  }

  for (auto& worker : worker_adapters_) {
    worker_pool_.start(worker);
  }
}

void Pipeline::PipelineImpl::stop() {
  fsi_->stop();
  queue_->close();
  worker_pool_.joinAll();
}

void Pipeline::PipelineImpl::queue(core::MessageQueuePtr queue) {
  queue_ = std::move(queue);
}

std::string Pipeline::PipelineImpl::name() const {
  return name_;
}

size_t Pipeline::PipelineImpl::workerSize() const {
  return worker_pool_.capacity();
}

size_t Pipeline::PipelineImpl::availableWorkers() const {
  return worker_pool_.available();
}

size_t Pipeline::PipelineImpl::usedWorkers() const {
  return worker_pool_.used();
}

void Pipeline::PipelineImpl::internalCallback(const std::string& id, const std::string& content) {
  queue_->enqueue(std::make_unique<core::Message>(id, content));
}

std::vector<commands::CommandRawPtr> Pipeline::PipelineImpl::commandsRawPtr() const {
  std::vector<commands::CommandRawPtr> commands;
  for (const auto& command : commands_) {
    commands.emplace_back(command.get());
  }
  return commands;
}

Pipeline::Pipeline(const std::string& name, std::vector<commands::CommandPtr> commands, commands::CommandPtr error_command) :
  pimpl_(std::make_unique<PipelineImpl>(name, std::move(commands), std::move(error_command))) {
}

Pipeline::~Pipeline() = default;

void Pipeline::queue(core::MessageQueuePtr queue) {
  pimpl_->queue(std::move(queue));
}

void Pipeline::start() {
  pimpl_->start();
}

void Pipeline::stop() {
  pimpl_->stop();
}

std::string Pipeline::name() const {
  return pimpl_->name();
}

size_t Pipeline::workerSize() const {
  return pimpl_->workerSize();
}

size_t Pipeline::availableWorkers() const {
  return pimpl_->availableWorkers();
}

size_t Pipeline::usedWorkers() const {
  return pimpl_->usedWorkers();
}
}

The Pipeline uses a PipelineWorker which runs as a seperate Thread.

PipelineWorker.hpp/cpp

#pragma once
#include <memory>
#include <vector>

#include <Core/Message.hpp>
#include <Core/MessageQueue.hpp>
#include <Commands/Command.hpp>


namespace pipeline {

class PipelineWorker {
    class PipelineWorkerImpl;
public:
    PipelineWorker(const std::string& name, core::MessageQueueRawPtr queue, const std::vector<commands::CommandRawPtr>& commands, commands::CommandRawPtr error_command);
    virtual ~PipelineWorker();
    void run();
    std::string name() const;
    double pipelineProcessDuration() const;

private:
    std::unique_ptr<PipelineWorkerImpl> pimpl_;
};

}

// PipelineWorker.cpp

#include <PipelineWorker.hpp>

namespace pipeline {

class PipelineWorker::PipelineWorkerImpl {
public:
  PipelineWorkerImpl(const std::string& name, core::MessageQueueRawPtr queue, const std::vector<commands::CommandRawPtr>& commands, commands::CommandRawPtr error_command);
  void run();
  std::string name() const;
  double pipelineProcessDuration() const;

private:
  std::string name_;
  core::MessageQueueRawPtr queue_;
  std::vector<commands::CommandRawPtr> commands_;
  commands::CommandRawPtr error_command_;
  std::chrono::duration<double> pipeline_process_duration_{};
};

std::string PipelineWorker::PipelineWorkerImpl::name() const {
  return name_;
}

double PipelineWorker::PipelineWorkerImpl::pipelineProcessDuration() const {
  return pipeline_process_duration_.count();
}

PipelineWorker::PipelineWorkerImpl::PipelineWorkerImpl(const std::string& name, core::MessageQueueRawPtr queue, const std::vector<commands::CommandRawPtr>& commands, commands::CommandRawPtr error_command) :
name_(name),
queue_(queue),
commands_(commands),
error_command_(error_command) {
}

void PipelineWorker::PipelineWorkerImpl::run() {
  while (auto message = queue_->dequeue()) {
    auto pipeline_process_duration_begin = std::chrono::steady_clock::now();
    for (auto command : commands_) {
      try {
        message = command->execute(std::move(message));
      } catch (const std::exception& e) {
        error_command_->execute(std::make_unique<core::Message>(e.what(), std::string{ "Worker: " + name() + "\nCommand: " + command->name() }));
        break;
      }
    }
    pipeline_process_duration_ = std::chrono::steady_clock::now() - pipeline_process_duration_begin;
  }
}

PipelineWorker::PipelineWorker(const std::string& name, core::MessageQueueRawPtr queue, const std::vector<commands::CommandRawPtr>& commands, commands::CommandRawPtr error_command) :
  pimpl_( std::make_unique<PipelineWorkerImpl>(name, queue, commands, error_command) ) {
}

PipelineWorker::~PipelineWorker() = default;

void PipelineWorker::run() {
  pimpl_->run();
}

std::string PipelineWorker::name() const {
  return pimpl_->name();
}

double PipelineWorker::pipelineProcessDuration() const {
  return pimpl_->pipelineProcessDuration();
}

}

Main

In the Main class all the components needs to be sticked together for example:

#include <FilesystemInput/FilesystemInput.hpp>
#include <filesystem>
#include <iostream>
#include <thread>
#include <Commands/Command.hpp>
#include <Commands/CommandFactory.hpp>
#include <AddMetadataCommand/AddMetadataCommand.hpp>
#include <WrapCommand/WrapCommand.hpp>
#include <ErrorCommand/ErrorCommand.hpp>
#include <Pipeline/Pipeline.hpp>
#include <Core/Message.hpp>

class TestCommand : public commands::Command {
  void checkPreconditionsImpl(core::Message* msg) override {
    std::cout << "Check Precondition...\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    std::cout << "...done\n";
  }

  Command* cloneImpl() override {
    return new TestCommand{ *this };
  }

  void configureImpl(commands::commandsConfig config) override {
    std::cout << "Configure\n";
  }

  std::unique_ptr<core::Message> executeImpl(std::unique_ptr<core::Message> msg) override {
    std::cout << "Message received write through\n";
    std::cout << "Message ID: " << msg->id() << '\n';
    std::cout << "Message Content: " << msg->content() << '\n';
    return msg;
  }
};

int main() {

  auto& cmd_factory = commands::CommandFactory::Instance();


  cmd_factory.registerCommand("AddMetadataCommand", commands::MakeCommandPtr<commands::AddMetadataCommand>());
  cmd_factory.registerCommand("WrapCommand", commands::MakeCommandPtr<commands::WrapCommand>());
  cmd_factory.registerCommand("TestCommand", commands::MakeCommandPtr<TestCommand>());
  cmd_factory.registerCommand("ErrorCommand", commands::MakeCommandPtr<commands::ErrorCommand>());

  commands::commandsConfig addmetadata_config{ {"key_1", "value_1"}, { "key_2", "value_2" } };
  commands::commandsConfig wrap_config{
    {commands::WrapCommand::CFG_CMD_WRAP_TYPE, "json"},
    { commands::WrapCommand::CFG_CMD_WRAP_ELEMENT, "content" },
    { commands::WrapCommand::CFG_CMD_WRAP_METADATA, "true" }
  };

  std::vector<commands::CommandPtr> pipe_commands{};

  pipe_commands.push_back(cmd_factory.allocateCommand("AddMetadataCommand", addmetadata_config));
  pipe_commands.push_back(cmd_factory.allocateCommand("WrapCommand", wrap_config));
  pipe_commands.push_back(cmd_factory.allocateCommand("TestCommand", {}));

  auto pipeline = std::make_unique<pipeline::Pipeline>("Pipeline 1", std::move(pipe_commands), std::move(cmd_factory.allocateCommand("ErrorCommand", {})));

  std::cout << "Start Pipeline...\n";

  pipeline->start();

  std::cout << "Wait 2 Minutes...\n";

  std::this_thread::sleep_for(std::chrono::minutes(2));

  std::cout << "Stop Pipeline...\n";

  pipeline->stop();

}

Later the plan is to provide a ConfigurationLoader Class to create the Pipeline from a configuration file.

I'm open for any suggestions, improvements or changes. I also uploaded the Example with the other commands to Github: DataProcessor

Answer 1

Simplify things for the caller

Looking at your example main() function, I think that you have to write a lot of verbose code to use your pipeline. Let's just look at this line:

cmd_factory.registerCommand("AddMetadataCommand", commands::MakeCommandPtr<commands::AddMetadataCommand>());

The word "command", in various forms, is used 7 times on this line! A lot of it is redundant. I would rather be able to write this:

cmd_factory.register("AddMetadata", commands::AddMetadata);

You can do this by first removing "Command" for any member of a class that already has "Command" in its name. Then, let CommandFactory::register() take care of making a std::unique_ptr<Command> from the provided function.

Building the pipeline itself is also a lot of work. First you have to create a vector of commands, then you move that vector into the pipeline. Why not cut out the middle man, and make it so you can write:

auto pipeline = pipeline::Pipeline("Pipeline 1");
pipeline.add("AddMetadata", addmetadata_config);
pipeline.add("Wrap", wrap_config);
pipeline.add("Test", {});
pipeline.addErrorHandler("Error", {});
pipeline.start();

You can go even further by letting register(), add() and so on return a reference to *this, so the caller can then chain them together like so:

pipeline.add("AddMetadata", "addmetadata_config")
        .add("Wrap", wrap_config);
        …
        .start();

Pass config by reference

You are using (const) references or smart pointers to pass most things to functions, but commandConfigs are passed by value. That means expensive copies are being made.

Do you need all this?

You have written a lot of code that looks very flexible, but it actually does very little. Instead of registering commands, passing them configuration, and building a pipeline, wouldn't it be much easier if the caller could just write a single function like so:

void processMessage(core::Message message) {
    try {
        message.addMetadata("key_1", "value_1");
        message.addMetadata("key_2", "value_2");
        commands::wrapInJson(&mesage);
        std::cout << "Message received write through\n"
                     "Message ID: " << msg->id() << "\n"
                     "Message Content: " << msg->content() << '\n';
    } catch (const std::exception& e) {
        std::cerr << "Error occured: " << e.what() << '\n';
    }
}

Now you no longer need a command factory, command configuration, and your Pipeline now just needs to take a single function. It avoids a lot of memory allocations and pointer passing, so it will be more efficient. It would also allow reducing your main() to:

int main() {
    auto pipeline = pipeline::Pipeline("Pipeline 1", processMessage);
    pipeline.start();
    …
}

If you want to keep the pipeline stages separate and make it explicit how you combine them, then consider that C++ already natively allows you to combine functions. Consider:

// pipeline stages:
core::Message AddMetadata(core::Message message, …) {…}
core::Message Wrap(core::Message message, …) {…}
core::Message Test(core::Message message) {…}

void processMessage(core::Message message) {
    // pipeline composition:
    Test(Wrap(AddMetadata(message, …), …));
}

Your code does have some added functionality, like keeping track of the total duration a command ran for and how many messages it processed, but is that really used?