Session¶

diffusion.session ¶

Public session for end users.

ANONYMOUS_PRINCIPAL `module-attribute` ¶

ANONYMOUS_PRINCIPAL = ''

Anonymous principal username

Session ¶

The client session to a server or cluster of servers.

Notes

A session can be created by connecting to a server using SessionFactory.open(server_url: str), specifying the server URL. The session factory can be configured to control the behavior of the session.

The session provides a variety of operations to the application. These are grouped into feature interfaces, such as Topics and Messaging, exposed to the application through various properties.

Session lifecycle¶

Each session is managed by a server. The server assigns the session a unique identity session_id and manages the session's topic subscriptions, security details, and session properties.

A session can be terminated using close. A session may also be terminated by the server because of an error or a time out, or by other privileged sessions using the ClientControl feature.

The current state of the session can be retrieved with the state property. A listener can be registered with the state_changed event which will be notified when the session state changes.

Session Properties¶

For each session, the server stores a set of session properties that describe various attributes of the session.

There are two types of session property. Fixed properties are assigned by the server. User-defined properties are assigned by the application.

Many operations use session filter expressions that use session properties to select sessions.

Each property is identified by a key. Most properties have a single string values.

Fixed properties are identified by keys with a $ prefix. The available fixed session properties are:

Key	Description
`$ClientIP()`	The Internet address of the client in string format.
`$ClientType()`	The client type of the session. One of `ANDROID`, `C`, `DOTNET`, `IOS`, `JAVA`, `JAVASCRIPT_BROWSER`, `MQTT`, `PYTHON`, `REST`, or `OTHER`.
`$Environment()`	The environment in which the client is running. For possible values see the user manual.
`$Connector()`	The configuration name of the server connector that the client connected to.
`$Country()`	The country code for the country where the client's Internet address was allocated (for example, `NZ` for New Zealand). If the country code could not be determined, this will be a zero length string.
`$GatewayType()`	Gateway client type. Only set for gateway client sessions. If present it indicates the type of gateway client (e.g. Kafka).
`$GatewayId()`	The identity of a gateway client session. Only present if the `$GatewayType` session property is present.
`$Language()`	The language code for the official language of the country where the client's Internet address was allocated (for example, en for English). If the language could not be determined or is not applicable, this will be a zero length string.
`$Latitude()`	The client's latitude, if available. This will be the string representation of a floating point number and will be `NaN` if not available.
`$Longitude()`	The client's longitude, if available. This will be the string representation of a floating point number and will be `NaN` if not available.
`$MQTTClientId()`	The MQTT client identifier. Only set for MQTT sessions. If present, the value of the `$ClientType` session property will be `MQTT`.
`$Principal()`	The security principal associated with the client session.
`$Roles()`	Authorisation roles assigned to the session. This is a set of roles represented as quoted strings (for example, `"role1","role2"`). The utility method can be used to parse the string value into a set of roles.
`$ServerName()`	The name of the server to which the session is connected.
`$SessionId()`	TThe session identifier. Equivalent to `str(x.session_id)`, where `x` is a Session.
`$StartTime()`	The session's start time in milliseconds since the epoch.
`$Transport()`	The session transport type. One of `WEBSOCKET`, `HTTP_LONG_POLL`, `TCP`, or `OTHER`.

All listed property keys constants can be found in SessionProperty.

All user-defined property keys are non-empty strings and are case-sensitive. The characters ' ', '\t', '\r', '\n', '\"', '\'', '(', ')' are not allowed.

Session properties are initially associated with a session as follows:

When a client starts a session, it can optionally propose user-defined session properties (see SessionFactory.properties). Session properties proposed in this way must be accepted by the authenticator. This safeguard prevents abuse by a rogue, unprivileged client.
The server allocates all fixed property values.
The session is authenticated by registered authenticators. An authenticator that accepts a session can veto or change the user-defined session properties and add user-defined session properties. The authenticator can also change certain fixed properties.

Session Filters¶

Session filters are a mechanism of addressing a set of sessions by the values of their session properties.

Session filters are specified using a Domain Specific Language (DSL). For a full and detailed description of the session filters DSL see the user manual.

Session locks¶

The actions of multiple sessions can be coordinated using session locks. See SessionLock.

Note: This interface does not require user implementation and is only used to hide implementation details.

Note: Added in 6.6

Example:

session = diffusion.sessions().open("ws://localhost:8080")

state `property` ¶

state

Returns the current connection state of the session.

session_id `property` ¶

session_id

The current session ID.

services `property` ¶

services

The ServiceLocator instance responsible for retrieving services.

handlers `property` ¶

handlers

The collection of registered handlers.

data `property` ¶

data: dict

Internal data storage.

messaging `property` ¶

messaging: Messaging

Request-response messaging component.

topics `property` ¶

topics: Topics

Topics component.

session_trees `property` ¶

session_trees: SessionTrees

Session Trees component.

metrics `property` ¶

metrics: Metrics

Metrics component.

time_series `property` ¶

time_series: TimeSeries

Time Series component.

init ¶

__init__(
    url: StrictStr,
    principal: Optional[str] = None,
    credentials: Optional[Credentials] = None,
    properties: Optional[typing.Mapping[str, str]] = None,
    connection_factory: Optional[ConnectionFactory] = None,
    internal_session_factory: Optional[
        typing.Callable[
            [
                typing.Union[
                    AbstractConnection, ConnectionFactory
                ]
            ],
            InternalSession,
        ]
    ] = None,
    attributes: typing.Optional[SessionAttributes] = None,
)

A client session connected to a Diffusion server or a cluster of servers.

Parameters:

Name	Type	Description	Default
`url`	`StrictStr`	WebSockets URL of the Diffusion server to connect to.	required
`principal`	`Optional[str]`	The name of the security principal associated with the session.	`None`
`credentials`	`Optional[Credentials]`	Security information required to authenticate the connection.	`None`
`properties`	`Optional[typing.Mapping[str, str]]`	Session Properties.	`None`
`connection_factory`	`Optional[ConnectionFactory]`	Connection factory, for internal use.	`None`
`internal_session_factory`	`Optional[typing.Callable[[typing.Union[AbstractConnection, ConnectionFactory]], InternalSession]]`	Internal session factory, for internal use.	`None`
`attributes`	`typing.Optional[SessionAttributes]`	Session attributes.	`None`

The recommended method is to instantiate this class as an async context manager. Here is a minimal example:

>>> async with diffusion.Session("ws://diffusion.server:8080") as session:
...     # do some work with the session
...     pass

The context manager will make sure that the connection is properly closed at the end of the program. Alternatively, it is possible to open the connection explicitly, which can be useful if the session needs to be passed around, in this case the connection needs to be explicitly closed as well:

>>> import asyncio
>>> async def connect():
>>>     session = diffusion.Session("ws://diffusion.server:8080")
>>>     await session.connect()
>>>     # do some work with the session
>>>     await session.close()
>>> asyncio.run(connect())

connect `async` ¶

connect(properties: Optional[SessionProperties] = None)

Connect to the server.

Parameters:

Name	Type	Description	Default
`properties`	`Optional[SessionProperties]`	A dict of Diffusion session properties to set and/or update at connection.	`None`

close `async` ¶

close()

Closes the session.

ping_server `async` ¶

ping_server()

Send the user ping to the server.

lock `async` ¶

lock(
    lock_name: str,
    scope: SessionLockScope = SessionLockScope.UNLOCK_ON_SESSION_LOSS,
) -> SessionLock

Attempts to acquire a SessionLock with a given scope.

Notes

If the operation completes successfully, the result will be the requested SessionLock assigned to the calling session by the server. The session owns the returned lock and is responsible for unlocking it.

Acquiring the lock can take an arbitrarily long time if other sessions are competing for the lock. The server will retain the session's request for the lock until it is assigned to the session, the session is closed.

A session can call this method multiple times. If the lock is acquired, all calls will complete successfully with equal SessionLock.

A session that acquires a lock will remain its owner until it is unlocked (via SessionLock.unlock) or the session closes.

If called with a scope of SessionLockScope.UNLOCK_ON_SESSION_LOSS, this method behaves exactly like SessionLocks.lock.

If called with a scope of SessionLockScope.UNLOCK_ON_CONNECTION_LOSS, any lock that is returned will be unlocked if the session loses its connection to the server. This is useful to allow another session to take ownership of the lock while this session is reconnecting.

If a session makes multiple requests for a lock using different scopes, and the server assigns the lock to the session fulfilling the requests, the lock will be given the weakest scope (SessionLockScope.UNLOCK_ON_CONNECTION_LOSS).

Access control¶

To allow fine-grained access control, lock names are interpreted as path names, controlled with the update-topic/b, for example.

Since

6.10.

Parameters:

Name	Type	Description	Default
`lock_name`	`str`	The name of the session lock.	required
`scope`	`SessionLockScope`	The scope of the session lock.	`SessionLockScope.UNLOCK_ON_SESSION_LOSS`

Returns: A session lock object.

on_state_changed `async` ¶

on_state_changed(old_state: State, new_state: State)

Raises the StateChanged event from session. Bound as a handler on the InternalSession.on_state_changed event.

remove_listener ¶

remove_listener(listener: SessionListener)

Remove a given session state listener from the session Args: listener: the listener to remove

Raises:

Type	Description
`InvalidOperationError`	if no such listener is present

SessionListener ¶

Bases: EventStreamHandler

The session listener.

on_session_event `async` ¶

on_session_event(
    *,
    session: Session,
    old_state: typing.Optional[State],
    new_state: State
)

Called when a session changed state.

Parameters:

Name	Type	Description	Default
`session`	`Session`	The session that sent this event.	required
`old_state`	`typing.Optional[State]`	The old session state.	required
`new_state`	`State`	The session state.	required

exceptions ¶

SessionError ¶

Bases: DiffusionError

A Diffusion Session Error.

Name	Type	Description	Default
`lock_name`	`pydantic.StrictStr`	The name of the lock.	required
`scope`	`SessionLockScope`	The scope of the lock.	required

Name	Type	Description	Default
`session_details`	`SessionDetails`	the session details involved	required
`container_factory`	`SessionContainerFactory`	responsible for building the session and connecting it	required
`session_attributes`	`SessionAttributes`	the Session Attributes to use	required

Scheme	Transport Protocol
`ws`	WebSocket. See https://tools.ietf.org/html/rfc6455
`wss`	WebSocket over TLS.

Type	Description
`IllegalArgumentError`	if `url` is invalid
`IllegalStateError`	if any of the session attributes are found to be inconsistent. For example, if the default SSL context could not be loaded
`SessionEstablishmentError`	if an initial connection could not be established
`AuthenticationError`	if the client is insufficiently authorized to start the session, based on the supplied client credentials.

Session¶

diffusion.session ¶

ANONYMOUS_PRINCIPAL module-attribute ¶

Session ¶

Session lifecycle¶

Session Properties¶

Session Filters¶

Session locks¶

state property ¶

session_id property ¶

services property ¶

handlers property ¶

data property ¶

messaging property ¶

topics property ¶

session_trees property ¶

metrics property ¶

time_series property ¶

__init__ ¶

connect async ¶

close async ¶

ping_server async ¶

lock async ¶

Access control¶

on_state_changed async ¶

remove_listener ¶

SessionListener ¶

on_session_event async ¶

exceptions ¶

SessionError ¶

IncompatibleTopicError ¶

UpdateFailedError ¶

NoSuchTopicError ¶

NoTopicFoundError ¶

NoSuchEventError ¶

ExistingTopicError ¶

InvalidTopicPathError ¶

InvalidTopicSpecificationError ¶

TopicLicenseLimitError ¶

IncompatibleExistingTopicError ¶

AddTopicError ¶

UnsatisfiedConstraintError ¶

ExclusiveUpdaterConflictError ¶

InvalidPatchError ¶

FailedPatchError ¶

InvalidUpdateStreamError ¶

NotATimeSeriesTopicError ¶

IncompatibleTopicStateError ¶

InvalidQueryError ¶

SessionClosedError ¶

SessionSecurityError ¶

FatalConnectionError ¶

AuthenticationError ¶

ProxyAuthenticationError ¶

locks ¶

session_lock_acquisition ¶

SessionLockScope ¶

UNLOCK_ON_SESSION_LOSS class-attribute instance-attribute ¶

UNLOCK_ON_CONNECTION_LOSS class-attribute instance-attribute ¶

session_locks ¶

SessionLock ¶

Race conditions¶

unlock async ¶

set_released async ¶

SessionLocks ¶

__init__ ¶

lock async ¶

post_commit async ¶

unlock async ¶

properties ¶

ALL_FIXED_PROPERTIES module-attribute ¶

ALL_USER_PROPERTIES module-attribute ¶

retry_strategy ¶

RetryStrategy ¶

interval class-attribute instance-attribute ¶

attempts class-attribute instance-attribute ¶

session_attributes ¶

SessionAttributes dataclass ¶

DEFAULT_MAXIMUM_MESSAGE_SIZE class-attribute ¶

initial_retry_strategy class-attribute instance-attribute ¶

ANONYMOUS_PRINCIPAL `module-attribute` ¶

state `property` ¶

session_id `property` ¶

services `property` ¶

handlers `property` ¶

data `property` ¶

messaging `property` ¶

topics `property` ¶

session_trees `property` ¶

metrics `property` ¶

time_series `property` ¶

init ¶

connect `async` ¶

close `async` ¶

ping_server `async` ¶

lock `async` ¶

on_state_changed `async` ¶

on_session_event `async` ¶

UNLOCK_ON_SESSION_LOSS `class-attribute` `instance-attribute` ¶

UNLOCK_ON_CONNECTION_LOSS `class-attribute` `instance-attribute` ¶

unlock `async` ¶

set_released `async` ¶

init ¶

lock `async` ¶

post_commit `async` ¶

unlock `async` ¶

ALL_FIXED_PROPERTIES `module-attribute` ¶

ALL_USER_PROPERTIES `module-attribute` ¶

interval `class-attribute` `instance-attribute` ¶

attempts `class-attribute` `instance-attribute` ¶

SessionAttributes `dataclass` ¶

DEFAULT_MAXIMUM_MESSAGE_SIZE `class-attribute` ¶

initial_retry_strategy `class-attribute` `instance-attribute` ¶

ssl_context `class-attribute` `instance-attribute` ¶

maximum_message_size `class-attribute` `instance-attribute` ¶

server_host `class-attribute` `instance-attribute` ¶

server_port `class-attribute` `instance-attribute` ¶

server_url `property` ¶

secure_transport `class-attribute` `instance-attribute` ¶

SessionProperties `module-attribute` ¶

Connector `dataclass` ¶

session_details `instance-attribute` ¶

container_factory `instance-attribute` ¶

url `instance-attribute` ¶

init ¶

sessions `cached` ¶

password `property` ¶