TPP Topic 29: Juggling the Real World

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 19

In the FSM section we mentioned that you could move the generic state machine implementation into its own class. That class would probably be initialized by passing in a table of transitions and an initial state.

Try implementing the string extractor that way.

The generic part of the event/strings_fsm.rb example in the book is the logic to determine the appropriate state transition given an occurrence of some event . In this case the event is a character read from standard input; each new character determined the next state. Actually performing some action given a state transition is not generic and would be implemented by the caller of the generic state machine class.

# fsm.py

from typing import Dict, Hashable, Union, List, Tuple, Iterable

State = Hashable
Event = Hashable


class StateMachine:
    def __init__(
        self,
        transitions: Dict[State, Dict[Union[Event, State], List[State]]],
        default_state: State,
        initial_state: State,
    ):
        self.transitions = transitions
        self.default_state = default_state
        self.initial_state = initial_state

    def process(self, events: Iterable[Event]) -> Tuple[State, State, Event]:
        state = self.initial_state
        for event in events:
            state, action = self.transitions[state].get(
                event, self.transitions[state][self.default_state]
            )
            yield state, action, event

StateMachine.process takes the stream of events as input and returns a generator which, when iterated, yields the next state, action and the current event. The strings finite state machine could be implemented using this as follows:

# strings_fsm.py

from enum import Enum, auto

from fsm import StateMachine


class State(Enum):
    DEFAULT = auto()
    IGNORE = auto()
    LOOK_FOR_STRING = auto()
    START_NEW_STRING = auto()
    IN_STRING = auto()
    COPY_NEXT_CHAR = auto()
    ADD_CURRENT_TO_STRING = auto()
    FINISH_CURRENT_STRING = auto()


TRANSITIONS = {
    State.LOOK_FOR_STRING: {
        '"': [State.IN_STRING, State.START_NEW_STRING],
        State.DEFAULT: [State.LOOK_FOR_STRING, State.IGNORE],
    },
    State.IN_STRING: {
        '"': [State.LOOK_FOR_STRING, State.FINISH_CURRENT_STRING],
        "\\": [State.COPY_NEXT_CHAR, State.ADD_CURRENT_TO_STRING],
        State.DEFAULT: [State.IN_STRING, State.ADD_CURRENT_TO_STRING],
    },
    State.COPY_NEXT_CHAR: {
        State.DEFAULT: [State.IN_STRING, State.ADD_CURRENT_TO_STRING]
    },
}


def find_strings(char_stream):
    fsm = StateMachine(TRANSITIONS, State.DEFAULT, State.LOOK_FOR_STRING)

    for state, action, ch in fsm.process(char_stream):
        if action == State.IGNORE:
            continue
        elif action == State.START_NEW_STRING:
            result = []
        elif action == State.ADD_CURRENT_TO_STRING:
            result.append(ch)
        elif action == State.FINISH_CURRENT_STRING:
            yield "".join(result)

Usage:

>>> for s in find_strings('"foo" bar "baz bat"'):
...     print(s)
foo
baz bat

This version is further abstracted by separating the state machine input from it's processing. The original version read characters directly from standard input within the state machine loop whereas this implementation receives a char_stream parameter intended to be an iterable of characters (strings of length 1 in Python). The state transitions are the same as those given events/strings_fsm.rb in the book except defined using an Enum instead of strings.

The complete implementation including tests and reading from standard input can
be found on GitHub here.

Exercise 20

Which of these technologies (perhaps in combination) would be a good fit for the following situations:

• If you receive three network interface down events within five minutes, notify the operations staff.
• If it is after sunset, and there is motion detected at the bottom of the stairs followed by motion detected at the top of the stairs, turn on the upstairs lights.
• You want to notify various reporting systems that an order was completed.
• In order to determine whether a customer qualifies for a car loan, the application needs to send requests to three backend services and wait for the responses.

1. Event streams

   • The consumer of the network interface down events could group the stream of events into 3-tuples of consecutive events. The timestamp of the first and last events in each group would then be compared to determine if the operations staff should be notified.

2. Pubsub + state machine

   • Each motion detector would publish to a channel when it detects motion. A state machine can then subscribe to these channels to determine whether the downstairs motion event was produced prior to the upstairs motion event (with some appropriately sized delay between the two events). The lights would be subscribed to a control channel which the state machine can publish to in order to turn on the lights. The state machine will simply ignore the motion detector events if it is before sunset.

3. Pubsub

   • Each reporting system can subscribe to a completed orders channel. When the order is completed an event is published to the channel and the reporting systems can report to their heart's content.

4. Event streams

   • Each backend service call would be an event in an asynchronous stream - similar to event/rx1/index.js example in the book for fetching users from a REST API. This allows the requests to performed concurrently and therefore improve throughput.