replace RetryFilter with ErrorStrategy and improve the retry logic (#4)

Author: eli-darkly

contract-tests/stream_entity.py

@@ -44,9 +44,9 @@ def run(self):
                 request,
                 initial_retry_delay=millis_to_seconds(self.options.get("initialDelayMs")),
                 last_event_id=self.options.get("lastEventId"),
-                retry_filter=lambda _: RetryFilterResult(not self.closed),
-                logger=self.log,
-                defer_connect=True
+                error_strategy=ErrorStrategy.from_lambda(lambda _:
+                    (ErrorStrategy.FAIL if self.closed else ErrorStrategy.CONTINUE, None)),
+                logger=self.log
             )
             self.sse = sse
             for item in sse.all:

ld_eventsource/__init__.py

@@ -1,6 +1,6 @@
+from ld_eventsource.actions import *
 from ld_eventsource.errors import *
-from ld_eventsource.event import *
+from ld_eventsource.error_strategy import *
 from ld_eventsource.request_params import *
 from ld_eventsource.retry_delay_strategy import *
-from ld_eventsource.retry_filter import *
 from ld_eventsource.sse_client import *

ld_eventsource/event.py renamed to ld_eventsource/actions.py

@@ -2,7 +2,11 @@
 from typing import Optional
 
 
-class Event:
+class Action:
+    pass
+
+
+class Event(Action):
     """
     An event received by :class:`ld_eventsource.SSEClient`.
 
@@ -63,7 +67,7 @@ def __repr__(self):
         )
 
 
-class Comment:
+class Comment(Action):
     """
     A comment received by :class:`ld_eventsource.SSEClient`.
     
@@ -89,7 +93,7 @@ def __repr__(self):
         return ":" + self._comment
 
 
-class Start:
+class Start(Action):
     """
     Indicates that :class:`ld_eventsource.SSEClient` has successfully connected to a stream.
 
@@ -101,20 +105,20 @@ class Start:
     pass
 
 
-class Fault:
+class Fault(Action):
     """
     Indicates that :class:`ld_eventsource.SSEClient` encountered an error or end of stream.
 
     Instances of this class are only available from :prop:`ld_eventsource.SSEClient.all`.
-    They indicate either 1. a problem that happened after an initial successful connection
-    was made, or 2. a problem with the initial connection, if you passed `True` for
-    the `defer_connect` parameter to the :class:`ld_eventsource.SSEClient` constructor.
+
+    If you receive a Fault, the SSEClient is now in an inactive state since either a
+    connection attempt has failed or an existing connection has been closed. The SSEClient
+    will attempt to reconnect if you either call :func:`start()` or simply continue reading
+    events after this point.
     """
 
-    def __init__(self, error: Optional[Exception], will_retry: bool, retry_delay: float):
+    def __init__(self, error: Optional[Exception]):
         self.__error = error
-        self.__will_retry = will_retry
-        self.__retry_delay = retry_delay
 
     @property
     def error(self) -> Optional[Exception]:
@@ -124,19 +128,3 @@ def error(self) -> Optional[Exception]:
         in an orderly way after sending an EOF chunk as defined by chunked transfer encoding.
         """
         return self.__error
-    
-    @property
-    def will_retry(self) -> bool:
-        """
-        True if the :class:`ld_eventsource.SSEClient` will try to reconnect. This depends on
-        the nature of the error and whether a custom `retry_filter` was specified.
-        """
-        return self.__will_retry
-    
-    @property
-    def retry_delay(self) -> float:
-        """
-        The time, in seconds, that the :class:`ld_eventsource.SSEClient` will wait before
-        trying to reconnect. If :prop:`will_retry` was False, then this value is undefined.
-        """
-        return self.__retry_delay

ld_eventsource/error_strategy.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+import time
+from typing import Callable, Optional, Tuple
+
+
+class ErrorStrategy:
+    """Base class of strategies for determining how SSEClient should handle a stream error or the
+    end of a stream.
+    
+    The parameter that SSEClient passes to :func:`apply()` is either ``None`` if the server ended
+    the stream normally, or an exception. If it is an exception, it could be an I/O exception
+    (failure to connect, broken connection, etc.), or one of the error types defined in this
+    package such as :class:`ldeventsource.HTTPStatusError`.
+
+    The two options for the result are:
+
+    - :const:`FAIL`: This means that SSEClient should throw an exception to the caller-- or, in
+      the case of a stream ending without an error, it should simply stop iterating through events.
+    - :const:`CONTINUE`: This means that you intend to keep reading events, so SSEClient should
+      transparently retry the connection. If you are reading from :prop:`ld_eventsource.SSEClient.all`,
+      you will also receive a :class:`ld_eventsource.Fault` describing the error.
+
+    With either option, it is still always possible to explicitly reconnect the stream by calling
+    :func:`ld_eventsource.SSEClient.start()` again, or simply by trying to read from
+    :prop:`ld_eventsource.SSEClient.events` or :prop:`ld_eventsource.SSEClient.all` again.
+
+    Subclasses should be immutable. To implement strategies that behave differently on consecutive
+    retries, the strategy should return a new instance of its own class as the second return value
+    from ``apply``, rather than modifying the state of the existing instance. This makes it easy
+    for SSEClient to reset to the original error-handling state when appropriate by simply reusing
+    the original instance.
+    """
+
+    FAIL = True
+    CONTINUE = False
+
+    def apply(self, exception: Optional[Exception]) -> Tuple[bool, ErrorStrategy]:
+        """Applies the strategy to determine what to do after a failure.
+
+        :param exception: an I/O error, or one of the exception types defined by this package
+            (such as :class:`ldeventsource.HTTPStatusError`), or None if the stream simply ended
+        :return: a tuple where the first element is either :const:`FAIL` to raise an exception
+            or :const:`CONTINUE` to continue, and the second element is the strategy object to
+            use next time (which could be ``self``)
+        """
+        raise NotImplementedError("ErrorStrategy base class cannot be used by itself")
+
+    @staticmethod
+    def always_fail() -> ErrorStrategy:
+        """
+        Specifies that SSEClient should always treat an error as a stream failure. This is the
+        default behavior if you do not configure another.
+        """
+        return _LambdaErrorStrategy(lambda e: (ErrorStrategy.FAIL, None))
+
+    @staticmethod
+    def always_continue() -> ErrorStrategy:
+        """
+        Specifies that SSEClient should never raise an exception, but should transparently retry
+        or, if :prop:`ld_eventsource.SSEClient.all` is being used, return the error as a
+        :class:`ld_eventsource.Fault`.
+
+        Be aware that using this mode could cause connection attempts to block indefinitely if
+        the server is unavailable.
+        """
+        return _LambdaErrorStrategy(lambda e: (ErrorStrategy.CONTINUE, None))
+
+    @staticmethod
+    def continue_with_max_attempts(max_attempts: int) -> ErrorStrategy:
+        """
+        Specifies that SSEClient should automatically retry after an error for up to this
+        number of consecutive attempts, but should fail after that point.
+
+        :param max_attempts: the maximum number of consecutive retries
+        """
+        return _MaxAttemptsErrorStrategy(max_attempts, 0)
+    
+    @staticmethod
+    def continue_with_time_limit(max_time: float) -> ErrorStrategy:
+        """
+        Specifies that SSEClient should automatically retry after a failure and can retry
+        repeatedly until this amount of time has elapsed, but should fail after that point.
+
+        :param max_time: the time limit, in seconds
+        """
+        return _TimeLimitErrorStrategy(max_time, 0)
+
+    @staticmethod
+    def from_lambda(fn: Callable[[Optional[Exception]], Tuple[bool, Optional[ErrorStrategy]]]) -> ErrorStrategy:
+        """
+        Convenience method for creating an ErrorStrategy whose ``apply`` method is equivalent to
+        the given lambda.
+        
+        The one difference is that the second return value is an ``Optional[ErrorStrategy]`` which
+        can be None to mean "no change", since the lambda cannot reference the strategy's ``self``.
+        """
+        return _LambdaErrorStrategy(fn)
+
+
+class _LambdaErrorStrategy(ErrorStrategy):
+    def __init__(self, fn: Callable[[Optional[Exception]], Tuple[bool, Optional[ErrorStrategy]]]):
+        self.__fn = fn
+    
+    def apply(self, exception: Optional[Exception]) -> Tuple[bool, ErrorStrategy]:
+        should_raise, maybe_next = self.__fn(exception)
+        return (should_raise, maybe_next or self)
+
+class _MaxAttemptsErrorStrategy(ErrorStrategy):
+    def __init__(self, max_attempts: int, counter: int):
+        self.__max_attempts = max_attempts
+        self.__counter = counter
+    
+    def apply(self, exception: Optional[Exception]) -> Tuple[bool, ErrorStrategy]:
+        if self.__counter >= self.__max_attempts:
+            return (ErrorStrategy.FAIL, self)
+        return (ErrorStrategy.CONTINUE, _MaxAttemptsErrorStrategy(self.__max_attempts, self.__counter + 1))
+
+class _TimeLimitErrorStrategy(ErrorStrategy):
+    def __init__(self, max_time: float, start_time: float):
+        self.__max_time = max_time
+        self.__start_time = start_time
+    
+    def apply(self, exception: Optional[Exception]) -> Tuple[bool, ErrorStrategy]:
+        if self.__start_time == 0:
+            return (ErrorStrategy.CONTINUE, _TimeLimitErrorStrategy(self.__max_time, time.time()))
+        if (time.time() - self.__start_time) < self.__max_time:
+            return (ErrorStrategy.CONTINUE, self)
+        return (ErrorStrategy.FAIL, self)

ld_eventsource/reader.py

@@ -1,4 +1,4 @@
-from ld_eventsource.event import Comment, Event
+from ld_eventsource.actions import Comment, Event
 
 from typing import Callable, Iterable, Optional
 

ld_eventsource/request_params.py

@@ -7,9 +7,6 @@ class RequestParams:
 
     When calling the :class:`ld_eventsource.SSEClient` constructor, you can pass a
     `RequestParams` instance for the first parameter instead of a simple URL string.
-    Also, if you have specified a custom `retry_filter`, the filter can set
-    :prop:`ld_eventsource.retry.RetryFilterResult.request_params` to change the
-    parameters for the next request on a retry.
     """
     def __init__(
         self,