DEV Community

Cover image for LGTM Devlog 28: Game event loop using Google Cloud Scheduler and PubSub

LGTM Devlog 28: Game event loop using Google Cloud Scheduler and PubSub

Yuan Gao
CTO in tech 👨‍💻 Python, Vue.js, Former Electrical Engineer 🤖 Occasional robot robot builder and gamedev 🏆 Forbes 30 Under 30 Enterprise tech
・2 min read

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 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):
                    kwargs[arg_name] = parse_raw(
                except ValidationError as err:
                    raise StatusReturn(error=f"Validation error {err}", http_code=400)

                    "Decoded model and injected",

        return func(**kwargs)

    return wrapper
Enter fullscreen mode Exit fullscreen mode

This is very similar to the HTTP one, with two differences

  1. we do base64 decoding of the event key first, before putting it to the model
  2. there's no return structure handling
  3. 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.

New Cloud Scheduler Topic

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.

New PubSub Topic

Pubsub trigger function

The "boilerplate" for our pubsub trigger functions now looks like this:

def tick(tick_event: TickEvent):
    """ Game tick """"Tick", source=tick_event.source)
Enter fullscreen mode Exit fullscreen mode

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
  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 }}'
Enter fullscreen mode Exit fullscreen mode

Requisite tests:

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 ="/", json=tick_payload)
    assert res.status_code != 200

def test_tick(tick_client, tick_payload):
    """ Test tick """
    res ="/", json=tick_payload)
    assert res.status_code == 200
Enter fullscreen mode Exit fullscreen mode

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.

Discussion (0)