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.
-
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.
-
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.
-
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.
-
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.
- Each backend service call would be an event in an asynchronous stream - similar to
Top comments (0)