With the quest stage execution loop down, we now need a way to trigger the game event loop regularly to process any new stuff coming in. I'm reviving the "create new game" pubsub trigger function from last time, but updating it with some new stuff we've done recently. The code is in branch 5d28b82
New dependency injection for pubsub events
Just like we add a decorator to decode payloads from HTTP function, I'm adding a decorator to framework.py for PubSub functions:
def inject_pubsub_model(func):
""" Wrap method with pydantic dependency injection for PubSub functions """
def wrapper(event: dict, context: Context):
kwargs = {}
for arg_name, arg_type in get_type_hints(func).items():
parse_raw = getattr(arg_type, "parse_raw", None)
if callable(parse_raw):
try:
kwargs[arg_name] = parse_raw(
b64decode(event["data"]).decode("utf-8")
)
except ValidationError as err:
raise StatusReturn(error=f"Validation error {err}", http_code=400)
logger.info(
"Decoded model and injected",
model=arg_type.__name__,
func=func.__name__,
)
return func(**kwargs)
return wrapper
This is very similar to the HTTP one, with two differences
- we do base64 decoding of the event key first, before putting it to the model
- there's no return structure handling
- some differences in the function signature of pubsub trigger events
New Cloud Scheduler task
Over on Google Cloud, we add a task to the Cloud Scheduler service. This scheduler will publish a fixed message to a pubsub topic of our choice on a schedule of our choice.
With the cron of */5 * * * * this schedule will post our desired payload {"source": "Cloud Scheduler"} to the pub/sub topic tick every five minutes. This is our game tick. We can adjust this time later, we might even in the future get rid of it in favour of triggers and hooks, we'll see.
We also need to create a pubsub topic as well.
Pubsub trigger function
The "boilerplate" for our pubsub trigger functions now looks like this:
@inject_pubsub_model
def tick(tick_event: TickEvent):
""" Game tick """
logger.info("Tick", source=tick_event.source)
And our GitHub CI gains a new deployment to publish this function
name: Deploy Tick
uses: google-github-actions/deploy-cloud-functions@v0.1.2
with:
credentials: ${{ secrets.GCP_CREDENTIALS }}
service_account_email: ${{ secrets.GCP_FUNCTIONS_SERVICE_ACCOUNT }}
source_dir: '${{ env.workdir }}/app'
name: tick
event_trigger_type: providers/cloud.pubsub/eventTypes/topic.publish
event_trigger_resource: projects/${{ secrets.GCP_PROJECT_ID }}/topics/tick
runtime: python39
env_vars: 'APP_VERSION=${{ steps.short_sha.outputs.sha7 }}'
Requisite tests:
@pytest.fixture
def tick_payload():
""" Payload for tick """
data = TickEvent(source="test").dict()
return {
"context": {
"eventId": "some-eventId",
"timestamp": "some-timestamp",
"eventType": "some-eventType",
"resource": "some-resource",
},
"data": {"data": b64encode(json.dumps(data).encode()).decode()},
}
def test_bad_payload(tick_client, tick_payload):
""" Test a bad payload """
tick_payload["data"]["data"] = b64encode('{"a": 1}'.encode()).decode()
res = tick_client.post("/", json=tick_payload)
assert res.status_code != 200
def test_tick(tick_client, tick_payload):
""" Test tick """
res = tick_client.post("/", json=tick_payload)
assert res.status_code == 200
With this boilerplate in place, we're ready to implement the game's tick loop, which should be quite simple! Loop through all available quests, and execute their stages.
Top comments (0)