Part 1 – Event-Driven Architecture – Implementation with Call Back Functions

Concepts:

Event-Driven architecture is the design pattern that focuses on transfering data from place to place and have different entities respond accordingly when the data arrive.

• Events: changes that requires notification to various system's components. Events could be input data what a component wait for, a task (function or callback function call) that should be invoked.

The components of an event-driven architecture includes:

• Manager: manages a list of components, workers, that listens to the events.

  • Manager should be a singleton
  • Manager registers workers. WorkerClass could be initialized by itself. In C++, this could be implemented via delete constructor.
  • Manager publishes/send that task to certain workers. The workers could be retrieved from the list that it manages or the worker could be given a task from manager directly.

• Workers: listen to events, perform certain tasks and returns results via callback. Those workers could be on separate threads or processes. Workers could be multiple instances of a module, or intances of many modules.

  • Implement onStart function: run a logic when it receives a notification
  • Implement start function: call another logic to run a notification.

The events are sent via "Messages". If the workers are on different threads, then Inter Process Communication mechanism should be used (pipe, memshare, android binder messages etc).

Implementation

Using observer pattern

The following program implements 2 classes: Manager and Worker. Manager can call instances of Worker to perform job, and returns a random number. Manager registers each instance of Worker. Manager invokes Worker::onJobReceived function to send job. Manager passes a callback to Worker to receive result.

Note that this implementation only allows Manager to send job to worker, that is it only allows communication between the Manager and Workers

Each time a worker performs a job, it is popped off from the list of available workers.

Note: in this implementation, I also add a callback as argument to Worker::onJobReceived function. The callback is defined as a lamda function to print out result once the callback is invoked inside Worker::onJobReceived.

// Version 1.3 with updated random number generation

#include <iostream>
#include <vector>
#include <functional>
#include <cstdlib>
#include <ctime>

class Manager;

class Worker {
 public:
  Worker() {}
  ~Worker() {}
  void onJobReceived(std::function<void(int)> callback) {
    std::cout << "Start a job" << std::endl;
    srand(time(nullptr));
    int result = rand() % 100 + 1;
    callback(result);
  }
};

class Manager {
 public:
  Manager() {}
  ~Manager() {}

  void registerWorker(Worker* worker) {
    workers_.push_back(worker);
  }

  void callWorker() {
    if (!workers_.empty()) {
      workers_.back()->onJobReceived([](int result) {
        std::cout << "Result: " << result << std::endl;
      });
      workers_.pop_back();
    } else {
      std::cout << "No workers available" << std::endl;
    }
  }

 private:
  std::vector<Worker*> workers_;
};

int main() {
  Manager manager;
  Worker worker1, worker2;
  manager.registerWorker(&worker1);
  manager.registerWorker(&worker2);
  manager.callWorker();
  manager.callWorker();
  return 0;
}

Implementing communication between all workers

We can implement a sendJob which accepts a Callback from a worker. sendJob function then invokes Manager::callWorker to call another worker to perform the job and returns result to the calling worker.

This can be done by adding Manager as private member of class Worker.

// Version 1.5 with sendJob function

#include <iostream>
#include <vector>
#include <functional>
#include <cstdlib>
#include <ctime>

class Manager;

class Worker {
 public:
  Worker(Manager* manager) : manager_(manager) {
    srand(time(nullptr) + id_);
    id_ = rand() % 100 + 1;
  }
  ~Worker() {}
  void onJobReceived(std::function<void(int)> callback) {
    std::cout << "Start a job" << std::endl;
    srand(time(nullptr));
    int result = rand() % 100 + 1;
    callback(result);
  }
  int getID() const { return id_; }
  void sendJob() {
    manager_->callWorker();
  }
 private:
  int id_;
  Manager* manager_;
};

class Manager {
 public:
  Manager() {}
  ~Manager() {}

  void registerWorker(Worker* worker) {
    workers_.push_back(worker);
  }

  void callWorker() {
    if (!workers_.empty()) {
      workers_.back()->onJobReceived([](int result) {
        std::cout << "Result: " << result << std::endl;
      });
      std::cout << "Assigned to worker with ID " << workers_.back()->getID() << std::endl;
      workers_.pop_back();
    } else {
      std::cout << "No workers available" << std::endl;
    }
  }

 private:
  std::vector<Worker*> workers_;
};

int main() {
  Manager manager;
  Worker worker1(&manager), worker2(&manager);
  manager.registerWorker(&worker1);
  manager.registerWorker(&worker2);
  worker1.sendJob();
  worker2.sendJob();
  return 0;
}

Note:

The above functions has forward declaration of Manager. It will not work and return error "invalid use of incomplete type ‘class Manager’ ". This happens because the compiler only knows class Manager via the forward declaration, but not its member. Thus it won't know how to run manager_->callWorker(). The solution to this is to use proper C++ pattern of the class declaration in the header files, and function implementation in the cpp files. I don't show it in this blog post

In addition, it is worthwhile to visit the post "Different ways to implement callbacks – My sky (freewindcode.com)".

In the next post, I will show how to pass data between different workers that was initialized on different process. Thus Inter Process Communication is used together with slightly different design.