-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
115 additions
and
49 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
49 changes: 49 additions & 0 deletions
49
python/patterns-use-cases/src/statefulactors/machine_operator.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| import restate | ||
| from restate import VirtualObject, ObjectContext | ||
| from utils import bring_up_machine, Status, tear_down_machine | ||
|
|
||
| # This is a State Machine implemented with a Virtual Object | ||
| # | ||
| # - The object holds the state of the state machine and defines the methods | ||
| # to transition between the states. | ||
| # - The object's unique id identifies the state machine. Many parallel state | ||
| # machines exist, but only state machine (object) exists per id. | ||
| # - The "single-writer-per-key" characteristic of virtual objects ensures | ||
| # that one state transition per state machine is in progress at a time. | ||
| # Additional transitions are enqueued for that object, while a transition | ||
| # for a machine is still in progress. | ||
| machine_operator = VirtualObject("machine-operator") | ||
|
|
||
|
|
||
| @machine_operator.handler("setUp") | ||
| async def set_up(ctx: ObjectContext): | ||
| machine_id = ctx.key() | ||
|
|
||
| # Ignore duplicate calls to 'setUp' | ||
| status = await ctx.get("status") | ||
| if status == Status.UP: | ||
| return f"{machine_id} is already up, so nothing to do" | ||
|
|
||
| # Bringing up a machine is a slow process that frequently crashes | ||
| await bring_up_machine(ctx, machine_id) | ||
| ctx.set("status", Status.UP) | ||
|
|
||
| return f"{machine_id} is now up" | ||
|
|
||
|
|
||
| @machine_operator.handler("tearDown") | ||
| async def tear_down(ctx: ObjectContext): | ||
| machine_id = ctx.key() | ||
|
|
||
| status = await ctx.get("status") | ||
| if status != Status.UP: | ||
| return f"{machine_id} is not up, cannot tear down" | ||
|
|
||
| # Tearing down a machine is a slow process that frequently crashes | ||
| await tear_down_machine(ctx, machine_id) | ||
| ctx.set("status", Status.DOWN) | ||
|
|
||
| return f"{machine_id} is now down" | ||
|
|
||
|
|
||
| app = restate.app([machine_operator]) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| from datetime import timedelta | ||
|
|
||
| import restate | ||
| import logging | ||
| import os | ||
| import random | ||
|
|
||
| logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(process)d] [%(levelname)s] - %(message)s') | ||
|
|
||
|
|
||
| class Status: | ||
| UP = "UP" | ||
| DOWN = "DOWN" | ||
|
|
||
|
|
||
| async def bring_up_machine(ctx: restate.Context, machine_id: str): | ||
| logging.info(f"Beginning transition of {machine_id} to up") | ||
| maybe_crash(0.4) | ||
| await ctx.sleep(timedelta(seconds=5)) | ||
| logging.info(f"Done transitioning {machine_id} to up") | ||
|
|
||
|
|
||
| async def tear_down_machine(ctx: restate.Context, machine_id: str): | ||
| logging.info(f"Beginning transition of {machine_id} to down") | ||
| maybe_crash(0.4) | ||
| await ctx.sleep(timedelta(seconds=5)) | ||
| logging.info(f"Done transitioning {machine_id} to down") | ||
|
|
||
|
|
||
| def maybe_crash(probability: float = 0.5) -> None: | ||
| if random.random() < probability: | ||
| logging.error("A failure happened!") | ||
| raise Exception("A failure happened!") |