Neomega

外观

bus_events API ​

完整接口目录 · 插件运行时用法 · 工具包下载

模块说明 ​

text
Typed, staged event declarations; validation never calls business code.

SourceContext ​

python
@dataclass(frozen=True)
class SourceContext

字段（包含继承字段）：

SourceContext.publisher_id ​

python
publisher_id: str

数据类字段；声明于 bus_events.SourceContext。

SourceContext.publisher ​

python
publisher: Identity

数据类字段；声明于 bus_events.SourceContext。

SourceContext.member ​

python
member: str

数据类字段；声明于 bus_events.SourceContext。

SourceContext.major ​

python
major: int

数据类字段；声明于 bus_events.SourceContext。

SourceContext.schema_digest ​

python
schema_digest: str

数据类字段；声明于 bus_events.SourceContext。

SourceContext.cancelled ​

python
@property
def cancelled(self)

EventContext ​

python
@dataclass(frozen=True)
class EventContext

SDK 说明：

text
Host-attested online event metadata, never copied from business values.

字段（包含继承字段）：

EventContext.sequence ​

python
sequence: int

数据类字段；声明于 bus_events.EventContext。

EventContext.publisher_id ​

python
publisher_id: str | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.publisher ​

python
publisher: Identity | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.member ​

python
member: str | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.major ​

python
major: int | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.schema_digest ​

python
schema_digest: str | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.accepted_at ​

python
accepted_at: datetime | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.packet_id ​

python
packet_id: int | None = None

数据类字段；声明于 bus_events.EventContext。

EventContext.connection_id ​

python
connection_id: str | None = None

数据类字段；声明于 bus_events.EventContext。

StreamStatus ​

python
@dataclass(frozen=True)
class StreamStatus

SDK 说明：

text
One current-generation snapshot; only an explicit pause may resume.

字段（包含继承字段）：

StreamStatus.index ​

python
index: int

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.kind ​

python
kind: str

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.state ​

python
state: str

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.member ​

python
member: str | None = None

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.major ​

python
major: int | None = None

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.packet_id ​

python
packet_id: int | None = None

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.reason ​

python
reason: str | None = None

数据类字段；声明于 bus_events.StreamStatus。

StreamStatus.gap ​

python
gap: bool = False

数据类字段；声明于 bus_events.StreamStatus。

event_options ​

python
def event_options(name, major)

SourceRegistration ​

python
class SourceRegistration

SourceRegistration.__init__ ​

python
def __init__(self, registry, key, handler, *, name, major=1, unique=True)

SourceRegistration.descriptor ​

python
@property
def descriptor(self)

SourceRegistration.create ​

python
def create(self, context)

SourceRegistration.close ​

python
def close(self)

SourceRegistration.binding ​

python
self.binding

公开实例属性，在 __init__ 中设置。

SourceRegistration.handler ​

python
self.handler

公开实例属性，在 __init__ 中设置。

SourceRegistration.member ​

python
self.member

公开实例属性，在 __init__ 中设置。

SourceRegistration.major ​

python
self.major

公开实例属性，在 __init__ 中设置。

EventRegistration ​

python
class EventRegistration

EventRegistration.__init__ ​

python
def __init__(
    self,
    handler,
    *,
    asynchronous,
    member = None,
    major = 1,
    packet_type = None,
    max_events = 256,
    max_bytes = 512 << 10,
)

EventRegistration.arguments ​

python
def arguments(self, value, context)

EventRegistration.binding ​

python
self.binding

公开实例属性，在 __init__ 中设置。

EventRegistration.handler ​

python
self.handler

公开实例属性，在 __init__ 中设置。

EventRegistration.asynchronous ​

python
self.asynchronous

公开实例属性，在 __init__ 中设置。

EventRegistration.member ​

python
self.member

公开实例属性，在 __init__ 中设置。

EventRegistration.major ​

python
self.major

公开实例属性，在 __init__ 中设置。

EventRegistration.packet_type ​

python
self.packet_type

公开实例属性，在 __init__ 中设置。

EventRegistration.max_events ​

python
self.max_events

公开实例属性，在 __init__ 中设置。

EventRegistration.max_bytes ​

python
self.max_bytes

公开实例属性，在 __init__ 中设置。

bus_publish ​

python
def bus_publish(name, *, major=1, unique=True)

on_bus_data ​

python
def on_bus_data(name, **options)

on_bus_data_async ​

python
def on_bus_data_async(name, **options)

on_packet ​

python
def on_packet(packet_type, **options)

on_packet_async ​

python
def on_packet_async(packet_type, **options)

Neomega · 使用、开发与参考文档