Event System

Note

Applies to: Spatial Extensions, Kit 109.0.3+, CloudXR 6

The XR event system provides a message bus architecture for reactive programming and component communication.

Message Bus Overview

The XR message bus is a carb.events.IEventStream that enables pub-sub messaging between XR components and your code.

Accessing the Message Bus

import omni.kit.xr.core
import carb.events

xr_core = omni.kit.xr.core.XRCore.get_singleton()
message_bus = xr_core.get_message_bus()

Basic Event Subscription

def on_xr_event(event):
    """Callback when event fires."""
    print(f"XR event received: {event}")

# Subscribe to event by type
event_type = carb.events.type_from_string("xr.enable")
subscription = message_bus.create_subscription_to_pop_by_type(
    event_type,
    on_xr_event
)

# Subscription stays active until reference is deleted
# To unsubscribe:
subscription = None

XR Lifecycle Events

Events emitted during XR system lifecycle.

System Events

xr.enable / xr.disable

Fired when XR system starts/stops.

def on_xr_enabled(event):
    print("XR system enabled")

def on_xr_disabled(event):
    print("XR system disabled")

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr.enable"),
    on_xr_enabled
)

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr.disable"),
    on_xr_disabled
)

xr_display.enable / xr_display.disable

Fired when XR display (rendering) starts/stops.

def on_display_enabled(event):
    print("XR display active - rendering started")
    # Good time to initialize XR UI elements

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_display.enable"),
    on_display_enabled
)

xr_viewport.enable / xr_viewport.disable

Fired when XR viewport is enabled/disabled.

def on_viewport_enabled(event):
    print("XR viewport enabled")

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_viewport.enable"),
    on_viewport_enabled
)

Profile Events

xr_profile.<profile_name>.enable / xr_profile.<profile_name>.disable

Fired when specific profile is enabled/disabled.

# VR profile enabled
def on_vr_profile_enabled(event):
    print("VR profile active")
    # Configure VR-specific features

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_profile.vr.enable"),
    on_vr_profile_enabled
)

# AR profile enabled
message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_profile.ar.enable"),
    lambda e: print("AR profile active")
)

profile_changed

Fired when active profile changes.

def on_profile_changed(event):
    current_profile = xr_core.get_current_profile_name()
    print(f"Profile changed to: {current_profile}")

from omni.kit.xr.core import XRCoreEventType
message_bus.create_subscription_to_pop_by_type(
    XRCoreEventType.profile_changed,
    on_profile_changed
)

Input Device Events

xr_input.<device_name>.enable / xr_input.<device_name>.disable

Fired when input device connects/disconnects.

def on_controller_connected(event):
    print("Left controller connected")
    # Initialize controller-specific features

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_input./user/hand/left.enable"),
    on_controller_connected
)

Frame Timing Events

Events synchronized with the XR rendering pipeline.

pre_sync_update

Executes in sync with rendering - ideal for updating visual elements.

from omni.kit.xr.core import XRCoreEventType

def on_pre_sync(event):
    """Called before rendering synchronization.
    
    Use this for:
    - Updating controller models
    - Updating selection beams
    - Updating any visual XR elements
    """
    # Update visual elements here
    pass

message_bus.create_subscription_to_pop_by_type(
    XRCoreEventType.pre_sync_update,
    on_pre_sync
)

post_sync_update

Executes after rendering scheduled (1-frame delay) - ideal for business logic.

def on_post_sync(event):
    """Called after rendering synchronization.
    
    Use this for:
    - Handling button presses
    - Menu logic
    - State transitions
    - Non-visual updates
    """
    # Handle business logic here
    pass

message_bus.create_subscription_to_pop_by_type(
    XRCoreEventType.post_sync_update,
    on_post_sync
)

post_device_events_update

Fired after input device events are processed.

message_bus.create_subscription_to_pop_by_type(
    XRCoreEventType.post_device_events_update,
    lambda e: print("Device events processed")
)

post_layer_update

Fired after XRUsdLayer updates complete.

message_bus.create_subscription_to_pop_by_type(
    XRCoreEventType.post_layer_update,
    lambda e: print("USD layers updated")
)

Input Event Generators

Custom events generated from input device gestures.

Creating Custom Input Events

# Bind event generator to device
device = xr_core.get_input_device("/user/hand/left")

event_generator = device.bind_event_generator(
    "trigger",              # Input name
    "weapon_fire",          # Event name prefix
    ["press", "release"],   # Event types
    {                       # Tooltips (optional)
        "press": "Fire Weapon",
        "release": "Stop Firing"
    }
)

# Subscribe to generated events
def on_weapon_fire_press(event):
    print("Weapon fired!")

def on_weapon_fire_release(event):
    print("Weapon stopped")

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("weapon_fire.press"),
    on_weapon_fire_press
)

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("weapon_fire.release"),
    on_weapon_fire_release
)

Event Generator Types

• press - Button transitions from unpressed to pressed

• release - Button transitions from pressed to unpressed

• update - Continuous updates while pressed (includes state data)

• touch - Button touched (capacitive)

• lift - Button no longer touched

• suspend - Focus taken by another event generator

• resume - Focus returned

• state - Complete state every frame (all gestures)

Global Event Generators

Create event generators not tied to specific devices:

# Bind event generator globally
# Useful for actions that can come from multiple sources
event_generator = xr_core.bind_input_event_generator(
    "menu_toggle",           # Event name
    [                        # List of input paths
        "/user/hand/left/menu.press",
        "/user/hand/right/menu.press",
        "keyboard/m.press"
    ],
    {                        # Tooltips
        "press": "Toggle Menu"
    }
)

# Subscribe once for all sources
message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("menu_toggle.press"),
    lambda e: print("Menu toggled from any source")
)

# Unbind when done
xr_core.unbind_input_event_generator("menu_toggle")

Tool and GUI Events

Events related to XR tools and GUI layers.

Tool Events

xr_tool..enable / xr_tool..disable

Fired when tool is activated/deactivated.

def on_teleport_tool_enabled(event):
    print("Teleport tool enabled")

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_tool.teleport.enable"),
    on_teleport_tool_enabled
)

GUI Layer Events

xr_gui..enable / xr_gui..disable

Fired when GUI layer is shown/hidden.

def on_menu_shown(event):
    print("XR menu GUI layer shown")

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("xr_gui.menu.enable"),
    on_menu_shown
)

Custom Events

Dispatch custom events through the XR message bus.

Dispatching Custom Events

# Simple dispatch
event_type = carb.events.type_from_string("my_custom_event")
message_bus.push(event_type, payload={"data": "value"})

# Dispatch with consume check
event_dict = {"key": "value", "number": 42}
consumed = xr_core.dispatch_message_bus_and_check_consume(
    carb.events.type_from_string("my_event"),
    event_dict
)

if consumed:
    print("Event was consumed by a listener")

Subscribing to Custom Events

def on_custom_event(event):
    # Access event payload
    payload = event.payload
    if "data" in payload:
        print(f"Received: {payload['data']}")

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("my_custom_event"),
    on_custom_event
)

Receiving Messages from CloudXR Clients

When a CloudXR client (Apple Vision Pro, Meta Quest, or a browser) sends a message over the opaque data channel, it arrives on the Kit message bus as a carb event. Understanding how the wire format maps to carb events is essential when writing custom Kit extensions that respond to client messages.

Wire Format to Carb Event Translation

Messages sent from clients use this JSON structure:

{
  "type": "myCustomAction",
  "payload": {
    "target": "/World/Geometry/Cube",
    "visible": true
  }
}

The omni.kit.xr.system.openxr plugin intercepts each incoming message and performs two translations:

1. It reads the "type" field and calls carb.events.type_from_string("myCustomAction") to produce the carb event type token. This token is what you subscribe to on the message bus.

2. It reads the "payload" object and flattens each child field directly into event.payload — so your handler receives event.payload["target"] and event.payload["visible"] as Python values, not a raw JSON string.

Important

The "type" field in the wire JSON and the carb event name are two different things. "type" is a CloudXR wire format field; carb has no concept of it. The OpenXR plugin is the bridge that converts one to the other. If you subscribe with carb.events.type_from_string("myCustomAction"), your handler fires whenever a client sends {"type": "myCustomAction", ...}. Nothing in the carb event system itself exposes or uses the string "type".

Writing a Custom Message Handler

import omni.ext
import carb.events
import omni.kit.app

class MyMessageHandlerExtension(omni.ext.IExt):
    def on_startup(self, ext_id):
        bus = omni.kit.app.get_app().get_message_bus_event_stream()

        # Subscribe using the same string the client sends in "type"
        self._sub = bus.create_subscription_to_pop_by_type(
            carb.events.type_from_string("myCustomAction"),
            self._on_custom_action
        )

    def _on_custom_action(self, event):
        # Payload fields are available directly — no JSON parsing needed
        target = event.payload.get("target", "")
        visible = event.payload.get("visible", True)
        print(f"Setting {target} visible={visible}")

    def on_shutdown(self):
        self._sub = None

From the client, trigger this handler by sending:

{
  "type": "myCustomAction",
  "payload": {
    "target": "/World/Geometry/Cube",
    "visible": false
  }
}

This works identically from any client — Apple Vision Pro, Meta Quest, or browser. The Kit extension cannot distinguish which client sent the message.

For the full message format reference, the "message" vs "payload" field distinction, USD Viewer built-in handlers, and client-side send code, see Scene Integration and Messaging.

Event Management Patterns

Subscription Lifecycle Management

class XREventManager:
    """Manage event subscriptions with automatic cleanup."""
    
    def __init__(self, xr_core):
        self.xr_core = xr_core
        self.message_bus = xr_core.get_message_bus()
        self._subscriptions = []
    
    def subscribe(self, event_name, callback):
        """Subscribe to event by name."""
        event_type = carb.events.type_from_string(event_name)
        sub = self.message_bus.create_subscription_to_pop_by_type(
            event_type,
            callback
        )
        self._subscriptions.append(sub)
        return sub
    
    def subscribe_once(self, event_name, callback):
        """Subscribe to event, auto-unsubscribe after first fire."""
        sub_holder = [None]  # Mutable container for closure
        
        def wrapper(event):
            callback(event)
            # Unsubscribe after first call
            if sub_holder[0] in self._subscriptions:
                self._subscriptions.remove(sub_holder[0])
            sub_holder[0] = None
        
        sub = self.subscribe(event_name, wrapper)
        sub_holder[0] = sub
        return sub
    
    def unsubscribe_all(self):
        """Unsubscribe from all events."""
        self._subscriptions.clear()
    
    def __del__(self):
        """Cleanup on destruction."""
        self.unsubscribe_all()

# Usage
event_manager = XREventManager(xr_core)

# Normal subscription
event_manager.subscribe("xr.enable", lambda e: print("XR enabled"))

# One-time subscription
event_manager.subscribe_once(
    "xr_display.enable",
    lambda e: print("Display enabled (fired once)")
)

# Clean up all
event_manager.unsubscribe_all()

Event Filtering Pattern

def create_filtered_subscription(message_bus, event_name, filter_func, callback):
    """Subscribe with filtering logic."""
    
    def filtered_callback(event):
        if filter_func(event):
            callback(event)
    
    event_type = carb.events.type_from_string(event_name)
    return message_bus.create_subscription_to_pop_by_type(
        event_type,
        filtered_callback
    )

# Usage: Only handle events with specific payload
def filter_important_events(event):
    return event.payload.get("priority") == "high"

sub = create_filtered_subscription(
    message_bus,
    "my_event",
    filter_important_events,
    lambda e: print("High priority event!")
)

Debounced Events Pattern

import asyncio

class DebouncedEventHandler:
    """Debounce rapid-fire events."""
    
    def __init__(self, callback, delay_seconds=0.5):
        self.callback = callback
        self.delay = delay_seconds
        self._task = None
    
    async def _debounced_call(self, event):
        await asyncio.sleep(self.delay)
        self.callback(event)
    
    def handle(self, event):
        # Cancel previous task
        if self._task:
            self._task.cancel()
        
        # Schedule new task
        self._task = asyncio.ensure_future(self._debounced_call(event))

# Usage: Debounce button spam
def on_button_action(event):
    print("Action executed (debounced)")

debounced_handler = DebouncedEventHandler(on_button_action, delay_seconds=0.3)

message_bus.create_subscription_to_pop_by_type(
    carb.events.type_from_string("button.press"),
    debounced_handler.handle
)

Complete Event System Example

import omni.kit.xr.core
import carb.events
from omni.kit.xr.core import XRCoreEventType, XRInputDeviceGeneratorEvent

class XRInteractionSystem:
    """Complete XR interaction system using events."""
    
    def __init__(self):
        self.xr_core = omni.kit.xr.core.XRCore.get_singleton()
        self.message_bus = self.xr_core.get_message_bus()
        self.subscriptions = []
        self.is_xr_active = False
        self.left_hand = None
        self.right_hand = None
    
    def initialize(self):
        """Set up all event subscriptions."""
        self._subscribe_lifecycle_events()
        self._subscribe_frame_events()
    
    def _subscribe_lifecycle_events(self):
        """Subscribe to XR lifecycle."""
        # XR enabled
        self.subscriptions.append(
            self.message_bus.create_subscription_to_pop_by_type(
                carb.events.type_from_string("xr.enable"),
                self._on_xr_enabled
            )
        )
        
        # XR disabled
        self.subscriptions.append(
            self.message_bus.create_subscription_to_pop_by_type(
                carb.events.type_from_string("xr.disable"),
                self._on_xr_disabled
            )
        )
        
        # Display enabled
        self.subscriptions.append(
            self.message_bus.create_subscription_to_pop_by_type(
                carb.events.type_from_string("xr_display.enable"),
                self._on_display_enabled
            )
        )
    
    def _subscribe_frame_events(self):
        """Subscribe to frame timing events."""
        # Pre-sync: Update visuals
        self.subscriptions.append(
            self.message_bus.create_subscription_to_pop_by_type(
                XRCoreEventType.pre_sync_update,
                self._on_pre_sync
            )
        )
        
        # Post-sync: Handle logic
        self.subscriptions.append(
            self.message_bus.create_subscription_to_pop_by_type(
                XRCoreEventType.post_sync_update,
                self._on_post_sync
            )
        )
    
    def _on_xr_enabled(self, event):
        """XR system enabled."""
        print("XR system enabled")
        self.is_xr_active = True
    
    def _on_xr_disabled(self, event):
        """XR system disabled."""
        print("XR system disabled")
        self.is_xr_active = False
        self._cleanup_input()
    
    def _on_display_enabled(self, event):
        """XR display active - set up input."""
        print("XR display enabled - setting up input")
        self._setup_input()
    
    def _setup_input(self):
        """Set up input event generators."""
        self.left_hand = self.xr_core.get_input_device("/user/hand/left")
        self.right_hand = self.xr_core.get_input_device("/user/hand/right")
        
        if self.left_hand:
            # Bind trigger for selection
            self.left_hand.bind_event_generator(
                "trigger", "left_trigger",
                ["press", "release"],
                {"press": "Select"}
            )
            
            # Subscribe to trigger events
            self.subscriptions.append(
                self.message_bus.create_subscription_to_pop_by_type(
                    carb.events.type_from_string("left_trigger.press"),
                    self._on_left_trigger_press
                )
            )
    
    def _cleanup_input(self):
        """Clean up input devices."""
        self.left_hand = None
        self.right_hand = None
    
    def _on_left_trigger_press(self, event):
        """Handle left trigger press."""
        print("Left trigger pressed - perform selection")
        # Implement selection logic
    
    def _on_pre_sync(self, event):
        """Update visuals before rendering."""
        if not self.is_xr_active:
            return
        
        # Update visual elements here
        # e.g., controller models, beams, etc.
        pass
    
    def _on_post_sync(self, event):
        """Handle business logic after rendering."""
        if not self.is_xr_active:
            return
        
        # Handle logic here
        # e.g., menu state, tool selection, etc.
        pass
    
    def shutdown(self):
        """Clean up all subscriptions."""
        self.subscriptions.clear()
        self.is_xr_active = False

# Usage
xr_system = XRInteractionSystem()
xr_system.initialize()

# When done
# xr_system.shutdown()

Next Steps

• Scene Integration – Work with USD in XR

• Custom Tools – Build event-driven tools

• Input & Interaction – Input event details

Key Takeaways

• Message bus provides pub-sub architecture

• Use pre_sync for visual updates, post_sync for logic

• Event generators create custom input events

• Proper subscription management prevents memory leaks

• Events enable reactive, decoupled architecture