wip

morpheus/pipeline/single_output_source.py

@@ -86,6 +86,23 @@ def stop(self):
         """
 
         # Indicate we need to stop
-        self._stop_requested = True
+        self.request_stop()
 
         return super().stop()
+
+    def request_stop(self):
+        """
+        Request the source to stop processing data.
+        """
+        self._stop_requested = True
+
+    def is_stop_requested(self) -> bool:
+        """
+        Returns `True` if a stop has been requested.
+
+        Returns
+        -------
+        bool:
+            True if a stop has been requested, False otherwise.
+        """
+        return self._stop_requested

morpheus/stages/input/arxiv_source.py

@@ -96,7 +96,6 @@ def __init__(self,
         self._total_pdfs = 0
         self._total_pages = 0
         self._total_chunks = 0
-        self._stop_requested = False
         self._cache_dir = cache_dir
 
     @property
@@ -142,7 +141,7 @@ def _generate_frames(self):
         )
 
         for x in search_results.results():
-            if self._stop_requested:
+            if self.is_stop_requested():
                 break
 
             full_path = os.path.join(self._cache_dir, x._get_default_filename())
@@ -175,7 +174,7 @@ def _process_pages(self, pdf_path: str):
 
                 logger.debug("Processing %s/%s: %s", len(documents), self._total_pages, pdf_path)
                 if self._total_pages > self._max_pages:
-                    self._stop_requested = True
+                    self.request_stop()
 
                 return documents
             except PdfStreamError:

morpheus/stages/input/control_message_kafka_source_stage.py

@@ -147,7 +147,7 @@ def _process_msg(self, consumer, msg):
             consumer.commit(message=msg, asynchronous=self._async_commits)
 
         if self._stop_after > 0 and self._records_emitted >= self._stop_after:
-            self._stop_requested = True
+            self.request_stop()
 
         return control_messages
 
@@ -159,7 +159,7 @@ def _source_generator(self):
 
             do_sleep = False
 
-            while not self._stop_requested:
+            while not self.is_stop_requested():
 
                 msg = consumer.poll(timeout=1.0)
                 if msg is None:
@@ -177,7 +177,7 @@ def _source_generator(self):
                     else:
                         raise ck.KafkaException(msg_error)
 
-                if do_sleep and not self._stop_requested:
+                if do_sleep and not self.is_stop_requested():
                     time.sleep(self._poll_interval)
 
         finally:

morpheus/stages/input/http_client_source_stage.py

@@ -185,7 +185,7 @@ def _generate_frames(self) -> typing.Iterator[MessageMeta]:
 
         request_args.update(self._requst_kwargs)
 
-        while (not self._stop_requested and (self._stop_after == 0 or num_records_emitted < self._stop_after)):
+        while (not self.is_stop_requested() and (self._stop_after == 0 or num_records_emitted < self._stop_after)):
             if self._query_params_fn is not None:
                 request_args['params'] = self._query_params_fn()
 

morpheus/stages/input/http_server_source_stage.py

@@ -122,7 +122,6 @@ def __init__(self,
         self._request_timeout_secs = request_timeout_secs
         self._lines = lines
         self._stop_after = stop_after
-        self._stop_requested = False
         self._payload_to_df_fn = payload_to_df_fn
         self._http_server = None
 
@@ -253,7 +252,7 @@ def _generate_frames(self) -> typing.Iterator[MessageMeta]:
                     df = self._queue.get()
                     self._queue_size -= 1
                 except queue.Empty:
-                    if (not self._http_server.is_running() or self._stop_requested):
+                    if (not self._http_server.is_running() or self.is_stop_requested()):
                         self._processing = False
                     else:
                         logger.debug("Queue empty, sleeping ...")

morpheus/stages/input/kafka_source_stage.py

@@ -125,9 +125,6 @@ def __init__(self,
         self._async_commits = async_commits
         self._client = None
 
-        # Flag to indicate whether or not we should stop
-        self._stop_requested = False
-
         self._poll_interval = pd.Timedelta(poll_interval).total_seconds()
         self._started = False
 
@@ -173,7 +170,7 @@ def _process_batch(self, consumer, batch):
                 self._num_messages += 1
 
                 if self._stop_after > 0 and self._records_emitted >= self._stop_after:
-                    self._stop_requested = True
+                    self.request_stop()
 
             batch.clear()
 
@@ -187,7 +184,7 @@ def _source_generator(self):
 
             batch = []
 
-            while not self._stop_requested:
+            while self.is_stop_requested():
                 do_process_batch = False
                 do_sleep = False
 
@@ -214,7 +211,7 @@ def _source_generator(self):
                     if message_meta is not None:
                         yield message_meta
 
-                if do_sleep and not self._stop_requested:
+                if do_sleep and not self.is_stop_requested():
                     time.sleep(self._poll_interval)
 
             message_meta = self._process_batch(consumer, batch)

morpheus/stages/input/rss_source_stage.py

@@ -69,8 +69,6 @@ def __init__(self,
                  request_timeout: float = 2.0,
                  strip_markup: bool = False):
         super().__init__(c)
-        self._stop_requested = False
-
         if (batch_size is None):
             batch_size = c.pipeline_batch_size