[filter] Stop processing in repeat() filter when pipy is shut down

flomesh-io · May 6, 2024 · 08daf31 · 08daf31
1 parent 5502d69
commit 08daf31
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/src/filters/repeat.cpp b/src/filters/repeat.cpp
@@ -72,6 +72,7 @@ void Repeat::reset() {
   m_outputting = false;
   m_restarting = false;
   m_ended = false;
+  m_shutdown = false;
 }
 
 void Repeat::process(Event *evt) {
@@ -86,9 +87,17 @@ void Repeat::process(Event *evt) {
   m_pipeline->input()->input(evt);
 }
 
+void Repeat::shutdown() {
+  m_shutdown = true;
+}
+
 void Repeat::on_reply(Event *evt) {
   if (auto eos = evt->as<StreamEnd>()) {
     m_eos = eos;
+    if (m_shutdown) {
+      end();
+      return;
+    }
     pjs::Value arg(evt), ret;
     if (!Filter::callback(m_condition, 1, &arg, ret)) return;
     if (ret.is_promise()) {

diff --git a/src/filters/repeat.hpp b/src/filters/repeat.hpp
@@ -46,6 +46,7 @@ class Repeat : public Filter, public EventSource {
   virtual auto clone() -> Filter* override;
   virtual void reset() override;
   virtual void process(Event *evt) override;
+  virtual void shutdown() override;
   virtual void on_reply(Event *evt) override;
   virtual void dump(Dump &d) override;
 
@@ -58,6 +59,7 @@ class Repeat : public Filter, public EventSource {
   bool m_outputting = false;
   bool m_restarting = false;
   bool m_ended = false;
+  bool m_shutdown = false;
 
   void restart();
   void repeat();