Allow setting multiple callbacks

modules/canbus/canbus.cc

@@ -130,7 +130,7 @@ Status Canbus::Start() {
   const double duration = 1.0 / FLAGS_chassis_freq;
   timer_ = AdapterManager::CreateTimer(ros::Duration(duration),
                                        &Canbus::OnTimer, this);
-  AdapterManager::SetControlCommandCallback(&Canbus::OnControlCommand, this);
+  AdapterManager::AddControlCommandCallback(&Canbus::OnControlCommand, this);
 
   // last step: publish monitor messages
   apollo::common::monitor::MonitorBuffer buffer(&monitor_);

modules/canbus/tools/teleop.cc

@@ -335,7 +335,7 @@ class Teleop {
       return -1;
     }
     is_running_ = true;
-    AdapterManager::SetChassisCallback(&Teleop::OnChassis, this);
+    AdapterManager::AddChassisCallback(&Teleop::OnChassis, this);
     keyboard_thread_.reset(
         new std::thread([this] { KeyboardLoopThreadFunc(); }));
     if (keyboard_thread_ == nullptr) {

modules/common/adapters/adapter.h

@@ -145,7 +145,7 @@ class Adapter {
    */
   void OnReceive(const D& message) {
     EnqueueData(message);
-    FireCallback(message);
+    FireCallbacks(message);
   }
 
   /**
@@ -228,8 +228,8 @@ class Adapter {
    * message hits the adapter.
    * @param callback the callback with signature void(const D &).
    */
-  void SetCallback(Callback callback) {
-    receive_callback_ = callback;
+  void AddCallback(Callback callback) {
+    receive_callbacks_.push_back(callback);
   }
 
   /**
@@ -346,12 +346,13 @@ class Adapter {
   }
 
   /**
-   * @brief proactively invokes the callback with the specified data.
+   * @brief proactively invokes the callbacks one by one registered with the
+   * specified data.
    * @param data the specified data.
    */
-  void FireCallback(const D& data) {
-    if (receive_callback_ != nullptr) {
-      receive_callback_(data);
+  void FireCallbacks(const D& data) {
+    for (const auto& callback : receive_callbacks_) {
+      callback(data);
     }
   }
 
@@ -391,7 +392,7 @@ class Adapter {
   std::list<std::shared_ptr<D>> observed_queue_;
 
   /// User defined function when receiving a message
-  Callback receive_callback_ = nullptr;
+  std::vector<Callback> receive_callbacks_;
 
   /// The mutex guarding data_queue_ and observed_queue_
   mutable std::mutex mutex_;

modules/common/adapters/adapter_manager.h

@@ -77,15 +77,15 @@ namespace adapter {
                   "Data type must be the same with adapter's type!");          \
     instance()->name##_->FillHeader(module_name, data);                        \
   }                                                                            \
-  static void Set##name##Callback(name##Adapter::Callback callback) {          \
+  static void Add##name##Callback(name##Adapter::Callback callback) {          \
     CHECK(instance()->name##_)                                                 \
         << "Initialize adapter before setting callback";                       \
-    instance()->name##_->SetCallback(callback);                                \
+    instance()->name##_->AddCallback(callback);                                \
   }                                                                            \
   template <class T>                                                           \
-  static void Set##name##Callback(                                             \
+  static void Add##name##Callback(                                             \
       void (T::*fp)(const name##Adapter::DataType &data), T *obj) {            \
-    Set##name##Callback(std::bind(fp, obj, std::placeholders::_1));            \
+    Add##name##Callback(std::bind(fp, obj, std::placeholders::_1));            \
   }                                                                            \
                                                                                \
  private:                                                                      \

modules/control/control.cc

@@ -86,8 +86,8 @@ Status Control::Init() {
   CHECK(AdapterManager::GetControlCommand())
       << "ControlCommand publisher is not initialized.";
 
-  AdapterManager::SetPadCallback(&Control::OnPad, this);
-  AdapterManager::SetMonitorCallback(&Control::OnMonitor, this);
+  AdapterManager::AddPadCallback(&Control::OnPad, this);
+  AdapterManager::AddMonitorCallback(&Control::OnMonitor, this);
 
   return Status::OK();
 }

modules/control/tools/terminal.cc

@@ -157,7 +157,7 @@ int main(int argc, char **argv) {
   }
 
   AdapterManager::Init(config);
-  AdapterManager::SetChassisCallback(on_chassis);
+  AdapterManager::AddChassisCallback(on_chassis);
 
   help();
   std::thread terminal_thread(terminal_thread_func);

modules/dreamview/backend/sim_control/sim_control.cc

@@ -80,8 +80,8 @@ void SimControl::SetStartPoint(const RoutingResponse& routing) {
 void SimControl::Start() {
   if (initial_start_) {
     // Setup planning and routing result data callback.
-    AdapterManager::SetPlanningCallback(&SimControl::OnPlanning, this);
-    AdapterManager::SetRoutingResponseCallback(&SimControl::SetStartPoint, this);
+    AdapterManager::AddPlanningCallback(&SimControl::OnPlanning, this);
+    AdapterManager::AddRoutingResponseCallback(&SimControl::SetStartPoint, this);
 
     // Start timer to publish localiztion and chassis messages.
     sim_control_timer_ = AdapterManager::CreateTimer(

modules/dreamview/backend/simulation_world/simulation_world_service.cc

@@ -601,7 +601,7 @@ void SimulationWorldService::UpdateSimulationWorld(
 
 void SimulationWorldService::RegisterMonitorCallback() {
   if (CheckAdapterInitialized("Monitor", AdapterManager::GetMonitor())) {
-    AdapterManager::SetMonitorCallback(
+    AdapterManager::AddMonitorCallback(
         &SimulationWorldService::UpdateSimulationWorld, this);
   }
 }

modules/perception/perception.cc

@@ -44,7 +44,7 @@ Status Perception::Init() {
   }
 
   CHECK(AdapterManager::GetPointCloud()) << "PointCloud is not initialized.";
-  AdapterManager::SetPointCloudCallback(&Perception::OnPointCloud, this);
+  AdapterManager::AddPointCloudCallback(&Perception::OnPointCloud, this);
   return Status::OK();
 }
 

modules/prediction/prediction.cc

@@ -73,7 +73,7 @@ Status Prediction::Init() {
   CHECK(AdapterManager::GetPerceptionObstacles()) << "Perception is not ready.";
 
   // Set perception obstacle callback function
-  AdapterManager::SetPerceptionObstaclesCallback(&Prediction::OnPerception,
+  AdapterManager::AddPerceptionObstaclesCallback(&Prediction::OnPerception,
                                                  this);
   return Status::OK();
 }

modules/routing/routing.cc

@@ -28,9 +28,9 @@ using apollo::common::ErrorCode;
 
 std::string Routing::Name() const { return FLAGS_node_name; }
 
-Routing::Routing() : 
+Routing::Routing() :
   monitor_(apollo::common::monitor::MonitorMessageItem::ROUTING) {
-  
+
   std::string graph_path = FLAGS_graph_dir + "/" + FLAGS_graph_file_name;
   AINFO << "Use routing topology graph path: " <<  graph_path.c_str();
   _navigator_ptr.reset(new Navigator(graph_path));
@@ -40,8 +40,8 @@ apollo::common::Status Routing::Init() {
   std::string graph_path = FLAGS_graph_dir + "/" + FLAGS_graph_file_name;
 
   AdapterManager::Init(FLAGS_adapter_config_path);
-  AdapterManager::SetMonitorCallback(&Routing::OnMonitor, this);
-  AdapterManager::SetRoutingRequestCallback(&Routing::OnRouting_Request, this);
+  AdapterManager::AddMonitorCallback(&Routing::OnMonitor, this);
+  AdapterManager::AddRoutingRequestCallback(&Routing::OnRouting_Request, this);
 
   return apollo::common::Status::OK();
 }