Skip to content

ConnectionInfo

zmqtt.client.ConnectionInfo dataclass #

ConnectionInfo(
    *,
    connection_id: int,
    session_present: bool,
    return_code: int,
    properties: ConnAckProperties | None,
    effective_client_id: str,
    effective_keepalive: int,
    effective_session_expiry_interval: int | None,
)

Immutable snapshot of a successful CONNECT/CONNACK handshake.

connection_id instance-attribute #

connection_id: int

session_present instance-attribute #

session_present: bool

return_code instance-attribute #

return_code: int

properties instance-attribute #

properties: ConnAckProperties | None

effective_client_id instance-attribute #

effective_client_id: str

effective_keepalive instance-attribute #

effective_keepalive: int

effective_session_expiry_interval instance-attribute #

effective_session_expiry_interval: int | None

from_connack classmethod #

from_connack(
    connect_packet: Connect,
    connack: ConnAck,
    *,
    version: Literal["3.1.1", "5.0"],
    connection_id: int,
) -> Self

Build a snapshot from a successful CONNECT/CONNACK exchange.

Source code in zmqtt/client.py
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@classmethod
def from_connack(
    cls,
    connect_packet: Connect,
    connack: ConnAck,
    *,
    version: Literal["3.1.1", "5.0"],
    connection_id: int,
) -> Self:
    """Build a snapshot from a successful CONNECT/CONNACK exchange."""
    properties = connack.properties
    client_id = connect_packet.client_id
    keepalive = connect_packet.keepalive
    expiry = None
    if version == "5.0":
        expiry = 0
        if connect_packet.properties is not None and connect_packet.properties.session_expiry_interval is not None:
            expiry = connect_packet.properties.session_expiry_interval
    if properties is not None:
        if not client_id and properties.assigned_client_identifier is not None:
            client_id = properties.assigned_client_identifier
        if properties.server_keep_alive is not None:
            keepalive = properties.server_keep_alive
        if version == "5.0" and properties.session_expiry_interval is not None:
            expiry = properties.session_expiry_interval
    return cls(
        connection_id=connection_id,
        session_present=connack.session_present,
        return_code=connack.return_code,
        properties=properties,
        effective_client_id=client_id,
        effective_keepalive=keepalive,
        effective_session_expiry_interval=expiry,
    )