Use a modification safe observers list

1 files changed, 28 insertions, 10 deletions

diff --git a/src/monitor.cc b/src/monitor.cc
index e5a231d..6194268 100644
--- a/src/monitor.cc
+++ b/src/monitor.cc
@@ -3,10 +3,10 @@
 #include <icecc/comm.h>
 #include <iostream>
 #include <unordered_map>
-#include <vector>
 
 #include "looper.hh"
 #include "monitor.hh"
+#include "observers.hh"
 
 namespace {
 
@@ -75,7 +75,11 @@ public:
   }
 
   void add_observer(Observer* observer) override {
-    observers_.push_back(observer);
+    observers_.add(observer);
+  }
+
+  void remove_observer(Observer* observer) override {
+    observers_.remove(observer);
   }
 
 private:
@@ -190,8 +194,10 @@ private:
       return;
     }
 
-    for (auto* observer : observers_) {
+    auto observer = observers_.first();
+    while (observer) {
       observer->state(this, CONNECTED);
+      observer = observers_.next();
     }
   }
 
@@ -201,8 +207,10 @@ private:
     looper_->remove(channel_->fd);
     channel_.reset();
 
-    for (auto* observer : observers_) {
+    auto observer = observers_.first();
+    while (observer) {
       observer->state(this, SEARCHING);
+      observer = observers_.next();
     }
   }
 
@@ -286,8 +294,10 @@ private:
     auto& job = active_jobs_[job_id];
     job.source = source;
     job.target = target;
-    for (auto* observer : observers_) {
+    auto observer = observers_.first();
+    while (observer) {
       observer->added_job(this, job.source, job.target);
+      observer = observers_.next();
     }
   }
 
@@ -300,8 +310,10 @@ private:
     auto source = it->second.source;
     auto target = it->second.target;
     active_jobs_.erase(it);
-    for (auto* observer : observers_) {
+    auto observer = observers_.first();
+    while (observer) {
       observer->removed_job(this, source, target);
+      observer = observers_.next();
     }
   }
 
@@ -314,18 +326,24 @@ private:
       if (machine.name.empty()) {
         machines_.erase(msg->hostid);
         if (known) {
-          for (auto* observer : observers_) {
+          auto observer = observers_.first();
+          while (observer) {
             observer->removed_machine(this, msg->hostid);
+            observer = observers_.next();
           }
         }
       } else {
         if (known) {
-          for (auto* observer : observers_) {
+          auto observer = observers_.first();
+          while (observer) {
             observer->updated_machine(this, msg->hostid);
+            observer = observers_.next();
           }
         } else {
-          for (auto* observer : observers_) {
+          auto observer = observers_.first();
+          while (observer) {
             observer->added_machine(this, msg->hostid);
+            observer = observers_.next();
           }
         }
       }
@@ -379,7 +397,7 @@ private:
   }
 
   std::shared_ptr<Looper> looper_;
-  std::vector<Observer*> observers_;
+  Observers<Observer> observers_;
   std::unique_ptr<MsgChannel> channel_;
   std::unique_ptr<DiscoverSched> discover_;
   uint32_t discover_timer_;