base

The reconnect loop every broker's websocket shares.

A broker's quotes socket and order updates socket live in stock_brokers/websockets/<broker>.py and subclass BrokerWebsocket. The subclass writes what is particular to its broker: how to open the connection and block until it closes (_connect), how to log in again after a refusal (_log_in_again), and how to read the broker's frames. This class holds the loop around them, which is the same for every broker:

1. Connect, and block until the connection closes.
2. A connection that opened and was not refused resets the failure count and the backoff.
3. A refused handshake, or six failed connects in a row, logs in again once before connecting again.
4. Still failing after that login, the socket gives up: gave_up is set and run_forever returns, so the script can exit 1 and let systemd restart it.
5. Otherwise wait, doubling the wait from 1 second up to 60, and connect again.

The loop never touches Redis. A socket hands what it decodes to a function its script gives it, and the script does the writing.

Classes:

Name	Description
BrokerWebsocket	One websocket to a broker that reconnects until it is closed or cannot recover.

BrokerWebsocket

BrokerWebsocket(name, logger)

One websocket to a broker that reconnects until it is closed or cannot recover.

Attributes:

Name	Type	Description
name	str	The socket's name, which starts every line it logs.
gave_up	bool	Whether the socket stopped because it could not recover.

Sets up a socket that has not connected yet.

Parameters:

Name	Type	Description	Default
name	str	The socket's name, which starts every line it logs.	required
logger	Logger	Where the socket reports.	required

Returns:

Type	Description
None	This method returns nothing.

Example programs

Reconnecting after a dropGiving up after a second refusal

examples/stock_brokers/websockets/base/BrokerWebsocket/example_1_reconnecting_after_a_drop.py

"""Writes a tiny socket on top of `BrokerWebsocket` and watches its reconnect loop.

`BrokerWebsocket` is the loop every broker's socket shares: connect and block until the connection closes, wait with a doubling backoff, and connect again, until `close` is called. A broker's subclass supplies `_connect`, which opens the connection, sets `_opened` when it opens and blocks until it closes, and `_log_in_again`, which the loop calls when the broker refuses the credentials.

This program's `ScriptedFeedSocket` is such a subclass. Its `_connect` opens nothing real: it takes the next line of a script, prints what the pretend connection did, and returns. The first connection opens and then drops, the second fails before opening, and the third opens and the program closes the socket from inside it, the way a script's signal handler would. The backoff is set to zero seconds so the program does not wait.

Notice that the loop logs a reconnect after each ended connection, that a connection that opened resets the backoff, and that `close` ends `run_forever` with `gave_up` still False.

Run it from the project root:

    python examples/stock_brokers/websockets/base/BrokerWebsocket/example_1_reconnecting_after_a_drop.py
"""

import logging
import sys

from stock_brokers.websockets.base import (
    BrokerWebsocket,
)


class ScriptedFeedSocket(BrokerWebsocket):
    """A socket whose connections follow a fixed script instead of reaching a broker.

    Attributes:
        script (list): What each connection does: `drop`, `fail` or `close`.
    """

    def __init__(self, script, logger):
        """Sets up the socket with its script.

        Args:
            script (list): What each connection does: `drop`, `fail` or `close`.
            logger (logging.Logger): Where the socket reports.

        Returns:
            None: This method returns nothing.
        """
        super().__init__('Scripted feed', logger)
        self.script = script

    def _connect(self):
        """Plays the next step of the script as one connection.

        Returns:
            None: This method returns nothing.

        Raises:
            ConnectionError: When the step is `fail`.
        """
        step = self.script.pop(0)
        if step == 'fail':
            raise ConnectionError('network unreachable')
        self._opened = True
        print('Connection opened.')
        if step == 'close':
            self.close()
        print('Connection ended.')

    def _log_in_again(self):
        """Reports a login, which this script never asks for.

        Returns:
            None: This method returns nothing.
        """
        print('Logging in again.')


class ReconnectingAfterADropExample:
    """Runs the scripted socket through a drop, a failure and a close.

    Attributes:
        socket (ScriptedFeedSocket): The socket being shown.
    """

    def __init__(self):
        """Builds the socket with a zero second backoff.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        script = [
            'drop',
            'fail',
            'close',
        ]
        self.socket = ScriptedFeedSocket(script, logging.getLogger('feed'))
        self.socket.MIN_BACKOFF_SECONDS = 0

    def run(self):
        """Runs the reconnect loop until the script closes the socket.

        Returns:
            None: This method returns nothing.
        """
        self.socket.run_forever()
        print(f'Gave up: {self.socket.gave_up}')


if __name__ == '__main__':
    ReconnectingAfterADropExample().run()

It prints:

Connection opened.
Connection ended.
WARNING Scripted feed disconnected. Reconnecting in 0 seconds.
ERROR Scripted feed connection failed: ConnectionError: network unreachable
WARNING Scripted feed disconnected. Reconnecting in 0 seconds.
Connection opened.
Connection ended.
INFO Scripted feed stopped.
Gave up: False

examples/stock_brokers/websockets/base/BrokerWebsocket/example_2_giving_up_after_a_second_refusal.py

"""Shows `BrokerWebsocket` logging in again once after a refusal and giving up when refused again.

When a broker refuses the credentials, the subclass sets `_authentication_rejected` and returns from `_connect`. The shared loop then calls `_log_in_again` once and connects again. If that next connection is refused too, the loop decides the login did not help: it sets `gave_up`, logs that it is giving up, and returns from `run_forever`, so the broker's script can exit 1 and let systemd restart it. Separately, `close` can be called before the loop even starts, and then `run_forever` returns without connecting.

This program's `RefusingFeedSocket` subclasses `BrokerWebsocket` and refuses every connection, so no broker is reached. The backoff is set to zero seconds so the program does not wait. A second socket is closed before it runs to show the early return.

Notice the single login between the two refusals, and that the closed socket never prints a connection attempt.

Run it from the project root:

    python examples/stock_brokers/websockets/base/BrokerWebsocket/example_2_giving_up_after_a_second_refusal.py
"""

import logging
import sys

from stock_brokers.websockets.base import (
    BrokerWebsocket,
)


class RefusingFeedSocket(BrokerWebsocket):
    """A socket whose every connection is refused by the pretend broker.

    Attributes:
        attempts (int): How many connections have been tried.
        logins (int): How many times the socket logged in again.
    """

    def __init__(self, name, logger):
        """Sets up the socket.

        Args:
            name (str): The socket's name.
            logger (logging.Logger): Where the socket reports.

        Returns:
            None: This method returns nothing.
        """
        super().__init__(name, logger)
        self.attempts = 0
        self.logins = 0

    def _connect(self):
        """Tries one connection, which the pretend broker refuses.

        Returns:
            None: This method returns nothing.
        """
        self.attempts = self.attempts + 1
        print(f'{self.name}: connection {self.attempts} refused with HTTP 401.')
        self._authentication_rejected = True

    def _log_in_again(self):
        """Pretends to log in again.

        Returns:
            None: This method returns nothing.
        """
        self.logins = self.logins + 1
        print(f'{self.name}: logged in again.')


class GivingUpAfterASecondRefusalExample:
    """Runs one socket until it gives up and another that was closed before it started.

    Attributes:
        refused_socket (RefusingFeedSocket): The socket that is refused twice.
        closed_socket (RefusingFeedSocket): The socket closed before it runs.
    """

    def __init__(self):
        """Builds both sockets with a zero second backoff.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        logger = logging.getLogger('feed')
        self.refused_socket = RefusingFeedSocket('Refused feed', logger)
        self.refused_socket.MIN_BACKOFF_SECONDS = 0
        self.closed_socket = RefusingFeedSocket('Closed feed', logger)

    def run(self):
        """Runs both sockets and prints how each one ended.

        Returns:
            None: This method returns nothing.
        """
        self.refused_socket.run_forever()
        print(f'Refused feed gave up: {self.refused_socket.gave_up} after {self.refused_socket.attempts} connections and {self.refused_socket.logins} login')
        self.closed_socket.close()
        self.closed_socket.run_forever()
        print(f'Closed feed gave up: {self.closed_socket.gave_up} after {self.closed_socket.attempts} connections')


if __name__ == '__main__':
    GivingUpAfterASecondRefusalExample().run()

It prints:

Refused feed: connection 1 refused with HTTP 401.
Refused feed: logged in again.
WARNING Refused feed disconnected. Reconnecting in 0 seconds.
Refused feed: connection 2 refused with HTTP 401.
ERROR Refused feed still cannot connect after logging in again. Giving up.
INFO Refused feed stopped.
Refused feed gave up: True after 2 connections and 1 login
INFO Closed feed stopped.
Closed feed gave up: False after 0 connections

Methods:

Name	Description
close	Closes the connection and stops reconnecting.
run_forever	Connects, and reconnects with backoff, until closed or unable to recover.

Source code in stock_brokers/websockets/base.py

def __init__(self, name, logger):
    """
    Sets up a socket that has not connected yet.

    Args:
        name (str): The socket's name, which starts every line it logs.
        logger (logging.Logger): Where the socket reports.

    Returns:
        None: This method returns nothing.
    """
    self.name = name
    self.gave_up = False
    self._logger = logger
    self._stop = threading.Event()
    self._websocket_application = None
    self._opened = False
    self._authentication_rejected = False

close

close()

Closes the connection and stops reconnecting.

Returns:

Type	Description
None	This method returns nothing.

Example usage

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.close()

Source code in stock_brokers/websockets/base.py

def close(self):
    """
    Closes the connection and stops reconnecting.

    Returns:
        None: This method returns nothing.
    """
    self._stop.set()
    if self._websocket_application is not None:
        self._websocket_application.close()

run_forever

run_forever()

Connects, and reconnects with backoff, until closed or unable to recover.

Returns:

Type	Description
None	This method returns nothing. gave_up says whether the socket stopped because it could not recover.

Example usage

From example_1_reconnecting_after_a_drop.py

self.socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.refused_socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.run_forever()

Source code in stock_brokers/websockets/base.py

def run_forever(self):
    """
    Connects, and reconnects with backoff, until closed or unable to recover.

    Returns:
        None: This method returns nothing. `gave_up` says whether the socket stopped because it could not recover.
    """
    backoff = self.MIN_BACKOFF_SECONDS
    failed_connects = 0
    logged_in_again = False
    while not self._stop.is_set():
        self._opened = False
        self._authentication_rejected = False
        try:
            self._connect()
        except Exception as exception:
            self._logger.error(f"{self.name} connection failed: {type(exception).__name__}: {exception}")
        if self._stop.is_set():
            break

        if self._opened and not self._authentication_rejected:
            failed_connects = 0
            logged_in_again = False
            backoff = self.MIN_BACKOFF_SECONDS
        else:
            failed_connects = failed_connects + 1

        if self._authentication_rejected or failed_connects >= self.MAX_FAILED_CONNECTS:
            if logged_in_again and self._gives_up_after_logging_in_again(failed_connects):
                self.gave_up = True
                self._logger.error(f"{self.name} still cannot connect after logging in again. Giving up.")
                break
            try:
                self._log_in_again()
            except Exception as exception:
                self.gave_up = True
                self._logger.error(f"{self.name} could not log in again: {type(exception).__name__}: {exception}")
                break
            failed_connects = 0
            logged_in_again = True

        self._logger.warning(f"{self.name} disconnected. Reconnecting in {backoff} seconds.")
        self._stop.wait(backoff)
        backoff = min(backoff * 2, self.MAX_BACKOFF_SECONDS)
    self._logger.info(f"{self.name} stopped.")

_gives_up_after_logging_in_again

_gives_up_after_logging_in_again(failed_connects)

Whether a socket that has already logged in again gives up on this failure.

The loop asks only when the connection was refused or has failed six times in a row. By default the answer is yes, so a refusal straight after logging in again ends the socket. A broker whose refusals are not always about the login overrides this to give up only after six more failed connects.

Parameters:

Name	Type	Description	Default
failed_connects	int	How many connects in a row have failed since the last login.	required

Returns:

Type	Description
bool	True to give up now.

Source code in stock_brokers/websockets/base.py

def _gives_up_after_logging_in_again(self, failed_connects):
    """
    Whether a socket that has already logged in again gives up on this failure.

    The loop asks only when the connection was refused or has failed six times in a row.
    By default the answer is yes, so a refusal straight after logging in again ends the socket.
    A broker whose refusals are not always about the login overrides this to give up only after six more failed connects.

    Args:
        failed_connects (int): How many connects in a row have failed since the last login.

    Returns:
        bool: True to give up now.
    """
    return True

_connect

_connect()

Opens the connection with the credentials in force now and blocks until it closes.

A subclass sets _opened when the connection opens and _authentication_rejected when the broker refuses the credentials.

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
NotImplementedError	Always, until a subclass implements it.

Source code in stock_brokers/websockets/base.py

def _connect(self):
    """
    Opens the connection with the credentials in force now and blocks until it closes.

    A subclass sets `_opened` when the connection opens and `_authentication_rejected` when the broker refuses the credentials.

    Returns:
        None: This method returns nothing.

    Raises:
        NotImplementedError: Always, until a subclass implements it.
    """
    raise NotImplementedError

_log_in_again

_log_in_again()

Logs in again after the broker refused the credentials, unless another process already has.

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
NotImplementedError	Always, until a subclass implements it.

Source code in stock_brokers/websockets/base.py

def _log_in_again(self):
    """
    Logs in again after the broker refused the credentials, unless another process already has.

    Returns:
        None: This method returns nothing.

    Raises:
        NotImplementedError: Always, until a subclass implements it.
    """
    raise NotImplementedError

_on_close

_on_close(websocket_connection, status_code, message)

Reports that the connection closed.

Parameters:

Name	Type	Description	Default
websocket_connection	WebSocketApp	The connection that closed.	required
status_code	int | None	The close status the broker sent.	required
message	str | None	The close reason the broker sent.	required

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/base.py

def _on_close(self, websocket_connection, status_code, message):
    """
    Reports that the connection closed.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that closed.
        status_code (int | None): The close status the broker sent.
        message (str | None): The close reason the broker sent.

    Returns:
        None: This method returns nothing.
    """
    self._logger.info(f"{self.name} closed with status {status_code}: {message}")