[filter] Allow handleXXX() and replaceXXX() filter to block multiple …

…times for a series of events

src/buffer.hpp

@@ -154,6 +154,21 @@ class EventBuffer {
     }
   }
 
+  void flush_until(const std::function<bool(Event*)> &out) {
+    while (auto e = m_events.head()) {
+      m_events.remove(e);
+      e->m_in_buffer = false;
+      if (m_stats) {
+        if (auto data = e->as<Data>()) {
+          m_stats->size -= data->size();
+        }
+      }
+      auto ret = out(e);
+      e->release();
+      if (ret) break;
+    }
+  }
+
   void clear() {
     List<Event> events(std::move(m_events));
     while (auto e = events.head()) {

src/filters/handle.cpp

@@ -90,8 +90,9 @@ bool Handle::on_callback_return(const pjs::Value &result) {
     Filter::output(m_deferred_event);
     m_deferred_event = nullptr;
   }
-  m_event_buffer.flush([this](Event *evt) {
+  m_event_buffer.flush_until([this](Event *evt) {
     handle(evt);
+    return m_waiting;
   });
   return true;
 }

src/filters/on-body.cpp

@@ -74,10 +74,10 @@ void OnBody::handle(Event *evt) {
   } else if (evt->is<MessageEnd>() || evt->is<StreamEnd>()) {
     if (m_started) {
       auto body = m_body_buffer.flush();
+      m_started = false;
       if (Handle::callback(body)) {
         Handle::defer(evt);
       }
-      m_started = false;
       return;
     }
   }

src/filters/on-message.cpp

@@ -83,10 +83,10 @@ void OnMessage::handle(Event *evt) {
       }
       auto body = m_body_buffer.flush();
       pjs::Ref<Message> msg(Message::make(m_start->head(), body, tail, payload)), result;
+      m_start = nullptr;
       if (Handle::callback(msg)) {
         Handle::defer(evt);
       }
-      m_start = nullptr;
       return;
     }
   }

src/filters/replace-body.cpp

@@ -76,8 +76,8 @@ void ReplaceBody::handle(Event *evt) {
 
     } else if (evt->is<MessageEnd>() || evt->is<StreamEnd>()) {
       pjs::Ref<Data> body = m_body_buffer.flush();
-      if (!Replace::callback(body)) return;
       m_started = false;
+      if (!Replace::callback(body)) return;
       Replace::pass(evt);
     }
   }

src/filters/replace-message.cpp

@@ -84,8 +84,8 @@ void ReplaceMessage::handle(Event *evt) {
         payload = end->payload();
       }
       pjs::Ref<Message> msg(Message::make(m_start->head(), m_body_buffer.flush(), tail, payload));
-      if (!Replace::callback(msg)) return;
       m_start = nullptr;
+      if (!Replace::callback(msg)) return;
     }
   }
 }