⚠️ Beta State

PyBevy is in an early and experimental stage. The API is incomplete, subject to breaking changes without notice, and you should expect bugs. Many features are still under development.

Messages Between Systems

Use double-buffered messages for frame-delayed communication between systems.

Introduction

While Events and Observers (tutorial 02) provide immediate, trigger-based communication, Messages provide frame-delayed, double-buffered communication.

Messages written in frame N are readable in frame N+1. This is useful for:

  • Input events that multiple systems need to process.
  • Decoupled system communication without observers.
  • Patterns where you want systems to react in the next frame.

Key types:

  • MessageWriter[T]: Write messages from a system.
  • MessageReader[T]: Read messages in another system (one frame later).
from dataclasses import dataclass
from pybevy.prelude import *
from pybevy.ecs import Message, MessageWriter, MessageReader

Defining a Message

Messages inherit from Message and carry data. They must be registered with app.add_message().

@dataclass
class DamageEvent(Message):
    entity_name: str
    amount: int
 
@dataclass
class HealEvent(Message):
    entity_name: str
    amount: int

Writing Messages

Any system can write messages using MessageWriter[T].

@resource
class GameLog(Resource):
    def __init__(self):
        self.entries: list[str] = []
 
frame_count = 0
 
def simulate_combat(
    damage_writer: MessageWriter[DamageEvent],
    heal_writer: MessageWriter[HealEvent],
):
    global frame_count
    frame_count += 1
 
    # Simulate events on specific frames
    if frame_count == 1:
        damage_writer.write(DamageEvent("Goblin", 25))
        damage_writer.write(DamageEvent("Dragon", 50))
    elif frame_count == 2:
        heal_writer.write(HealEvent("Player", 30))

Reading Messages

Messages arrive one frame after being written. MessageReader is iterable and will be empty if no messages were written last frame.

def process_damage(
    damage_reader: MessageReader[DamageEvent],
    log: ResMut[GameLog],
):
    for event in damage_reader:
        msg = f"{event.entity_name} dealt {event.amount} damage!"
        print(msg)
        log.entries.append(msg)
 
def process_healing(
    heal_reader: MessageReader[HealEvent],
    log: ResMut[GameLog],
):
    for event in heal_reader:
        msg = f"{event.entity_name} healed for {event.amount}!"
        print(msg)
        log.entries.append(msg)

Running the App

Register message types with add_message(), then add writer and reader systems.

from pybevy.app import ScheduleRunnerPlugin
 
def setup(commands: Commands):
    commands.insert_resource(GameLog())
 
@entrypoint
def main(app: App) -> App:
    return (
        app
        .add_plugins(ScheduleRunnerPlugin.run_once())
        .add_message(DamageEvent)
        .add_message(HealEvent)
        .add_systems(Startup, setup)
        .add_systems(Update, (simulate_combat, process_damage, process_healing))
    )
 
if __name__ == "__main__":
    main().run()

Messages are best for fire-and-forget communication where multiple systems need to react to the same event, and exact timing (same frame) isn't critical. For immediate reactions, use Events and Observers instead.