dataclass representing a server sent event and the argument used to invoke event callbacks.
Attributes:
| Name |
Type |
Description |
event |
str
|
event name, defaults to "message"
|
data |
Any
|
event data, defaults to empty string
|
id |
Optional[str]
|
event id, defaults to None
|
retry |
Optional[int]
|
reconnection time in milliseconds, defaults to None, currently unused.
|
Source code in hololinked/hololinked/client/abstractions.py
| @dataclass
class SSE:
"""
dataclass representing a server sent event and the argument used to invoke event callbacks.
Attributes
----------
event: str
event name, defaults to "message"
data: Any
event data, defaults to empty string
id: Optional[str]
event id, defaults to None
retry: Optional[int]
reconnection time in milliseconds, defaults to None, currently unused.
"""
__slots__ = ("event", "data", "id", "retry")
def __init__(self):
self.clear()
def clear(self):
"""reset to default/empty values"""
self.event = "message" # type: str
self.data = "" # type: Any
self.id = None # type: str | None
self.retry = None # type: int | None
def flush(self) -> dict[str, Any] | None:
"""obtain the event payload as dictionary and reset to default values"""
if not self.data and self.id is None:
return None
payload = {
"event": self.event,
"data": self.data,
"id": self.id,
"retry": self.retry,
}
# reset to default for next event
self.clear()
return payload
|
Functions
clear
reset to default/empty values
Source code in hololinked/hololinked/client/abstractions.py
| def clear(self):
"""reset to default/empty values"""
self.event = "message" # type: str
self.data = "" # type: Any
self.id = None # type: str | None
self.retry = None # type: int | None
|
flush
flush() -> dict[str, Any] | None
obtain the event payload as dictionary and reset to default values
Source code in hololinked/hololinked/client/abstractions.py
| def flush(self) -> dict[str, Any] | None:
"""obtain the event payload as dictionary and reset to default values"""
if not self.data and self.id is None:
return None
payload = {
"event": self.event,
"data": self.data,
"id": self.id,
"retry": self.retry,
}
# reset to default for next event
self.clear()
return payload
|