Writing Event Handlers¶
Event handlers are functions that respond to events emitted by the game. This guide covers patterns and best practices for writing effective handlers.
Basic Handler Structure¶
Async Handlers¶
The standard handler pattern:
from maid_engine.core.events import Event
async def handle_damage(event: DamageEvent) -> None:
"""Handle damage dealt to an entity."""
entity = world.get_entity(event.target_id)
if entity is None:
return
health = entity.try_get(HealthComponent)
if health:
health.current = max(0, health.current - event.damage)
Sync Handlers¶
Synchronous handlers are automatically wrapped:
def handle_damage_sync(event: DamageEvent) -> None:
"""Synchronous handler - automatically wrapped."""
print(f"Damage dealt: {event.damage}")
Subscribing to Events¶
Simple Subscription¶
from maid_engine.core.events import EventBus
bus = EventBus()
# Subscribe and store handler ID
handler_id = bus.subscribe(DamageEvent, handle_damage)
# Later, unsubscribe
bus.unsubscribe(handler_id)
With Priority¶
from maid_engine.core.events import EventPriority
# Validation handler - runs first
bus.subscribe(
DamageEvent,
validate_damage,
priority=EventPriority.HIGHEST
)
# Main handler - runs at normal priority
bus.subscribe(
DamageEvent,
apply_damage,
priority=EventPriority.NORMAL
)
# Logging handler - runs last
bus.subscribe(
DamageEvent,
log_damage,
priority=EventPriority.LOWEST
)
One-Time Handlers¶
# Automatically unsubscribed after first call
bus.subscribe(
PlayerConnectedEvent,
send_motd,
once=True
)
With Content Pack Tracking¶
# Track source for hot reload support
bus.subscribe(
DamageEvent,
handle_damage,
source_pack="my-combat-pack"
)
# Unsubscribe all handlers from a pack
bus.unsubscribe_pack("my-combat-pack")
Handler Patterns¶
Validation Pattern¶
Use high priority to validate and potentially cancel events:
async def validate_attack(event: AttackEvent) -> None:
"""Validate attack is legal before processing."""
attacker = world.get_entity(event.attacker_id)
target = world.get_entity(event.target_id)
# Cancel if entities don't exist
if not attacker or not target:
event.cancel()
return
# Cancel if target is invulnerable
if target.has_tag("invulnerable"):
await send_message(attacker, "Target is invulnerable!")
event.cancel()
return
# Cancel if attacker is stunned
if attacker.try_get(StunnedComponent):
await send_message(attacker, "You are stunned!")
event.cancel()
return
bus.subscribe(AttackEvent, validate_attack, priority=EventPriority.HIGHEST)
Modification Pattern¶
Modify event data at high priority:
async def apply_armor_reduction(event: DamageEvent) -> None:
"""Reduce damage based on target's armor."""
target = world.get_entity(event.target_id)
if not target:
return
armor = target.try_get(ArmorComponent)
if armor:
# Reduce damage (modify event data)
reduction = min(event.damage, armor.defense)
event.damage -= reduction
bus.subscribe(DamageEvent, apply_armor_reduction, priority=EventPriority.HIGH)
Processing Pattern¶
Main logic at normal priority:
async def apply_damage(event: DamageEvent) -> None:
"""Apply damage to target's health."""
if event.cancelled:
return
target = world.get_entity(event.target_id)
if not target:
return
health = target.try_get(HealthComponent)
if health:
health.current = max(0, health.current - event.damage)
# Emit death event if health depleted
if health.current <= 0:
await bus.emit(EntityDiedEvent(
entity_id=event.target_id,
killer_id=event.source_id,
))
bus.subscribe(DamageEvent, apply_damage, priority=EventPriority.NORMAL)
Side Effect Pattern¶
Effects at low priority:
async def trigger_on_hit_effects(event: DamageEvent) -> None:
"""Trigger any on-hit effects."""
if event.cancelled:
return
source = world.get_entity(event.source_id)
if source:
effects = source.try_get(OnHitEffectsComponent)
if effects:
for effect in effects.effects:
await apply_effect(event.target_id, effect)
bus.subscribe(DamageEvent, trigger_on_hit_effects, priority=EventPriority.LOW)
Logging Pattern¶
Logging at lowest priority:
async def log_damage(event: DamageEvent) -> None:
"""Log all damage for analytics."""
logger.info(
f"Damage: {event.damage} to {event.target_id} "
f"from {event.source_id} (cancelled={event.cancelled})"
)
bus.subscribe(DamageEvent, log_damage, priority=EventPriority.LOWEST)
Handler in Systems¶
Systems often subscribe to events:
class DeathSystem(System):
"""Handles entity death."""
async def startup(self) -> None:
"""Subscribe to events on startup."""
self._death_handler = self.events.subscribe(
EntityDiedEvent,
self._on_death,
source_pack="my-pack",
)
async def shutdown(self) -> None:
"""Unsubscribe on shutdown."""
self.events.unsubscribe(self._death_handler)
async def _on_death(self, event: EntityDiedEvent) -> None:
"""Handle entity death."""
entity = self.entities.get(event.entity_id)
if not entity:
return
# Notify room
pos = entity.try_get(PositionComponent)
if pos:
await self._announce_death(pos.room_id, entity)
# Drop loot
await self._drop_loot(entity)
# Destroy entity
self.entities.destroy(event.entity_id)
async def update(self, delta: float) -> None:
"""May not need to do anything in update."""
pass
Handler Classes¶
For complex handlers, use a class:
class CombatEventHandler:
"""Handles all combat-related events."""
def __init__(self, world: World, bus: EventBus):
self._world = world
self._bus = bus
self._handlers: list[UUID] = []
def register(self) -> None:
"""Register all handlers."""
self._handlers.append(
self._bus.subscribe(
AttackEvent,
self._on_attack,
priority=EventPriority.NORMAL,
)
)
self._handlers.append(
self._bus.subscribe(
DamageEvent,
self._on_damage,
priority=EventPriority.NORMAL,
)
)
self._handlers.append(
self._bus.subscribe(
EntityDiedEvent,
self._on_death,
priority=EventPriority.NORMAL,
)
)
def unregister(self) -> None:
"""Unregister all handlers."""
for handler_id in self._handlers:
self._bus.unsubscribe(handler_id)
self._handlers.clear()
async def _on_attack(self, event: AttackEvent) -> None:
# Handle attack
pass
async def _on_damage(self, event: DamageEvent) -> None:
# Handle damage
pass
async def _on_death(self, event: EntityDiedEvent) -> None:
# Handle death
pass
Error Handling¶
Graceful Degradation¶
async def safe_handler(event: DamageEvent) -> None:
"""Handler that won't crash the game."""
try:
entity = world.get_entity(event.target_id)
if entity is None:
logger.warning(f"Entity {event.target_id} not found")
return
health = entity.try_get(HealthComponent)
if health is None:
logger.warning(f"Entity {event.target_id} has no health")
return
health.current -= event.damage
except Exception as e:
logger.exception(f"Error in damage handler: {e}")
# Don't re-raise - let other handlers run
Defensive Checks¶
async def defensive_handler(event: ItemPickedUpEvent) -> None:
"""Handler with defensive checks."""
# Check entity exists
entity = world.get_entity(event.entity_id)
if not entity:
return
# Check item exists
item = world.get_entity(event.item_id)
if not item:
return
# Check inventory exists
inventory = entity.try_get(InventoryComponent)
if not inventory:
return
# Check item component exists
item_data = item.try_get(ItemComponent)
if not item_data:
return
# Now safe to process
inventory.items.append(event.item_id)
Testing Handlers¶
Unit Testing¶
import pytest
from unittest.mock import AsyncMock, MagicMock
@pytest.mark.asyncio
async def test_damage_handler():
# Setup
world = MagicMock()
entity = MagicMock()
health = MagicMock()
health.current = 100
world.get_entity.return_value = entity
entity.try_get.return_value = health
# Create event
event = DamageEvent(
target_id=UUID("12345678-1234-1234-1234-123456789abc"),
damage=25,
)
# Call handler
await handle_damage(event)
# Assert
assert health.current == 75
Integration Testing¶
@pytest.mark.asyncio
async def test_damage_event_flow():
# Setup world and bus
world = World(settings)
bus = world.events
# Track events received
received_events = []
async def track_events(event):
received_events.append(event)
bus.subscribe(DamageEvent, track_events)
# Create entity with health
entity = world.create_entity()
entity.add(HealthComponent(current=100, maximum=100))
# Emit damage event
await bus.emit(DamageEvent(
target_id=entity.id,
damage=30,
))
# Assert
assert len(received_events) == 1
assert received_events[0].damage == 30
Best Practices¶
1. Check for Cancellation¶
async def handler(event: Event) -> None:
if event.cancelled:
return # Early exit if cancelled
# Process event...
2. Use try_get for Optional Components¶
async def handler(event: Event) -> None:
entity = world.get_entity(event.entity_id)
if not entity:
return
# Use try_get for potentially missing components
health = entity.try_get(HealthComponent)
if health:
# Process...
pass
3. Don't Block the Event Loop¶
# Bad - blocking operation
async def bad_handler(event: Event) -> None:
time.sleep(1) # Blocks!
# Good - async operation
async def good_handler(event: Event) -> None:
await asyncio.sleep(1) # Non-blocking
4. Keep Handlers Focused¶
# Good - focused handler
async def handle_damage(event: DamageEvent) -> None:
# Only damage application logic
pass
async def handle_death_check(event: DamageEvent) -> None:
# Only death checking logic
pass
# Bad - handler does too much
async def handle_everything(event: DamageEvent) -> None:
# Apply damage
# Check death
# Drop loot
# Update achievements
# Send notifications
# Too much!
5. Document Handler Behavior¶
async def handle_combat_start(event: CombatStartedEvent) -> None:
"""Initialize combat state when combat begins.
This handler:
- Creates combat context entity
- Adds InCombatComponent to participants
- Schedules combat tick timer
Priority: NORMAL
Cannot be cancelled (combat already initiated)
"""
pass
Next Steps¶
- Event Overview - Understanding the event system
- Creating Custom Events - Define your own events