DEV Community

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

Posted on

1

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

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
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:

@inject_pubsub_model
def tick(tick_event: TickEvent):
    """ Game tick """
    logger.info("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
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 }}'
Enter fullscreen mode Exit fullscreen mode

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
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.

Image of Timescale

🚀 pgai Vectorizer: SQLAlchemy and LiteLLM Make Vector Search Simple

We built pgai Vectorizer to simplify embedding management for AI applications—without needing a separate database or complex infrastructure. Since launch, developers have created over 3,000 vectorizers on Timescale Cloud, with many more self-hosted.

Read full post →

Top comments (0)

Billboard image

The Next Generation Developer Platform

Coherence is the first Platform-as-a-Service you can control. Unlike "black-box" platforms that are opinionated about the infra you can deploy, Coherence is powered by CNC, the open-source IaC framework, which offers limitless customization.

Learn more