def send_game_events()

in event-generator/basic/empty_game.py [0:0]

• 21 lines of code
• 1 McCabe index (conditional complexity)

def send_game_events(project_id=default_project_id, topic_id=default_topic_id, game_id="generated"):
    publisher = pubsub.PublisherClient()
    topic_path = publisher.topic_path(project_id, topic_id)

    # Start game
    data = json.dumps({"GameId": game_id}).encode("utf-8")

    future = publisher.publish(topic_path, data,
        PinballEventType="GameStarted",
        MachineId=machine_id,
        Simulated=str(simulated),
        Timestamp=datetime.datetime.now().isoformat(),
    )
    print(future.result())

    # Let the game last a *little* while, at least
    time.sleep(1)

    # End game
    data = json.dumps({"GameId": game_id, "TotalScore": 0, "GameLengthMillieconds": 1000}).encode("utf-8")

    future = publisher.publish(topic_path, data,
        PinballEventType="GameEnded",
        MachineId=machine_id,
        Simulated=str(simulated),
        Timestamp=datetime.datetime.now().isoformat(),
    )
    print(future.result())

    print(f"Published messages to {topic_path}.")