update docs and tutorials based on connect callback addition to source and sink (#775)

Author: tim-quix

docs/connectors/sinks/README.md

@@ -123,3 +123,13 @@ sdf.sink(influx_sink)
 if __name__ == '__main__':
     app.run()
 ```
+
+## Connection Callbacks
+
+Assuming a given connector is set up to utilize them, there are two callbacks that can 
+be set for when a client connects/authenticates successfully or not, named 
+`on_client_connect_success` and `on_client_connect_failure`, respectfully.
+
+Though having a `setup` method is required, it is not guaranteed that it is implemented or
+utilized fully with `community` sinks; you can inspect a given Sink's `setup` method 
+to confirm whether it tests the client connection there (and that the callbacks are then applicable).

docs/connectors/sinks/custom-sinks.md

@@ -11,61 +11,6 @@ The `StreamingDataFrame.sink()` accepts implementations of this class.
 Check out [InfluxDB3Sink](../../api-reference/sinks.md#influxdb3sink) and [CSVSink](../../api-reference/sinks.md#csvsink) for example implementations of the batching sinks.
 
 
-Here is the code for `BaseSink` class for the reference:
-
-```python
-import abc
-
-class BaseSink(abc.ABC):
-    """
-    This is a base class for all sinks.
-
-    Subclass and implement its methods to create your own sink.
-
-    Note that sinks are currently in beta, and their design may change over time.
-    """
-
-    @abc.abstractmethod
-    def flush(self, topic: str, partition: int):
-        """
-        This method is triggered by the Checkpoint class when it commits.
-
-        You can use `flush()` to write the batched data to the destination (in case of
-        a batching sink), or confirm the delivery of the previously sent messages
-        (in case of a streaming sink).
-
-        If flush() fails, the checkpoint will be aborted.
-        """
-
-    @abc.abstractmethod
-    def add(
-        self,
-        value: Any,
-        key: Any,
-        timestamp: int,
-        headers: List[Tuple[str, HeaderValue]],
-        topic: str,
-        partition: int,
-        offset: int,
-    ):
-        """
-        This method is triggered on every new record sent to this sink.
-
-        You can use it to accumulate batches of data before sending them outside, or
-        to send results right away in a streaming manner and confirm a delivery later
-        on flush().
-        """
-
-    def on_paused(self, topic: str, partition: int):
-        """
-        This method is triggered when the sink is paused due to backpressure, when
-        the `SinkBackpressureError` is raised.
-
-        Here you can react to backpressure events.
-        """
-```
-
-
 ## Sinks Workflow
 
 During processing, sinks do the following operations:
@@ -109,14 +54,23 @@ class MyDatabaseSink(BatchingSink):
     """
     Some sink writing data to a database
     """
+    def __init__(self, on_client_connect_success=None, on_client_connect_failure=None):
+        super().__init__(
+           on_client_connect_success=on_client_connect_success,
+           on_client_connect_failure=on_client_connect_failure
+        )
+        self._db_connection = None
+
+    def setup(self):
+        self._db_connection = my_database.connect('<connection credentials>')
+        self._db_connection.test()
 
     def write(self, batch: SinkBatch):
         # Simulate some DB connection here
-        db_connection = my_database.connect('<connection credentials>')
         data = [{'value': item.value} for item in batch]
         try:
             # Try to write data to the db
-            db_connection.write(data)
+            self._db_connection.write(data)
         except TimeoutError:
             # In case of timeout, tell the app to wait for 30s 
             # and retry the writing later

docs/connectors/sources/README.md

@@ -133,3 +133,13 @@ def main():
 if __name__ == "__main__":
     main()
 ```
+
+## Connection Callbacks
+
+Assuming a given connector is set up to utilize them, there are two callbacks that can 
+be set for when a client connects/authenticates successfully or not, named 
+`on_client_connect_success` and `on_client_connect_failure`, respectfully.
+
+Though having a `setup` method is required, it is not guaranteed that it is implemented or
+utilized fully with `community` sources; you can inspect a given Source's `setup` method 
+to confirm whether it tests the client connection there (and that the callbacks are then applicable).

docs/connectors/sources/custom-sources.md

@@ -10,14 +10,23 @@ Quix Streams also provides a set of classes to help users implement custom sourc
 
 The recommended parent class to create a new source. It handles configuring, starting and stopping the source, as well as implementing a series of helpers.
 
-To get started, implement the [`run`](../../api-reference/sources.md#sourcerun) method and return when `self.running` is `False`.
+To get started, implement the [`run`](../../api-reference/sources.md#sourcerun) method 
+which loops while `self.running` is `True` (or until it's done).
+
+When you also have a client pattern, it is recommended to use the required `setup` 
+method to establish the initial connection/authentication so that the built-in callbacks 
+of `on_client_connect_success` and `on_client_connect_failure` can be utilized. 
+Otherwise, just set it to return.
 
 Example subclass:
 
 ```python
 from quixstreams.sources.base import Source
 
 class MySource(Source):
+    def setup(self):
+        return
+    
     def run(self):
         with open("file.txt", "r") as f:
             while self.running:

docs/tutorials/anomaly-detection/tutorial_app.py

@@ -50,6 +50,9 @@ def __init__(self):
         }
         super().__init__(name="temperature-event-generator")
 
+    def setup(self):
+        return
+
     def update_machine_temp(self, machine_id):
         """
         Updates the temperature for a machine by -1, 0, or 1 based on its current temp.

docs/tutorials/purchase-filtering/tutorial_app.py

@@ -66,6 +66,9 @@ class PurchaseGenerator(Source):
     def __init__(self):
         super().__init__(name="customer-purchases")
 
+    def setup(self):
+        return
+
     def run(self):
         for cid, purchase_info in enumerate(self._purchases_data):
             event = self.serialize(key=f"CUSTOMER_ID{cid}", value=purchase_info)

docs/tutorials/websocket-source/tutorial_app.py

@@ -36,6 +36,9 @@ def __init__(
         self._url = url
         self._product_ids = product_ids
 
+    def setup(self):
+        return
+
     def run(self) -> None:
         """
         The main method of the source with the main logic.

docs/tutorials/word-count/tutorial_app.py

@@ -27,6 +27,9 @@ class ReviewGenerator(Source):
     def __init__(self):
         super().__init__(name="customer-reviews")
 
+    def setup(self):
+        return
+
     def run(self):
         for review in self._review_list:
             event = self.serialize(key=choice(self._product_list), value=review)