Source code for azure.servicebus.aio._servicebus_session_async

# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import logging
from typing import Union
import six

from .._servicebus_session import BaseSession
from .._common.constants import (
    REQUEST_RESPONSE_GET_SESSION_STATE_OPERATION,
    REQUEST_RESPONSE_SET_SESSION_STATE_OPERATION,
    REQUEST_RESPONSE_RENEW_SESSION_LOCK_OPERATION,
    MGMT_RESPONSE_SESSION_STATE,
    MGMT_RESPONSE_RECEIVER_EXPIRATION,
    MGMT_REQUEST_SESSION_ID,
    MGMT_REQUEST_SESSION_STATE
)
from .._common import mgmt_handlers
from .._common.utils import utc_from_timestamp

_LOGGER = logging.getLogger(__name__)


[docs]class ServiceBusSession(BaseSession): """ The ServiceBusSession is used for manage session states and lock renewal. **Please use the instance variable `session` on the ServiceBusReceiver to get the corresponding ServiceBusSession object linked with the receiver instead of instantiating a ServiceBusSession object directly.** .. admonition:: Example: .. literalinclude:: ../samples/async_samples/sample_code_servicebus_async.py :start-after: [START get_session_async] :end-before: [END get_session_async] :language: python :dedent: 4 :caption: Get session from a receiver """
[docs] async def get_session_state(self): # type: () -> str """Get the session state. Returns None if no state has been set. :rtype: str .. admonition:: Example: .. literalinclude:: ../samples/async_samples/sample_code_servicebus_async.py :start-after: [START get_session_state_async] :end-before: [END get_session_state_async] :language: python :dedent: 4 :caption: Get the session state """ self._check_live() response = await self._receiver._mgmt_request_response_with_retry( # pylint: disable=protected-access REQUEST_RESPONSE_GET_SESSION_STATE_OPERATION, {MGMT_REQUEST_SESSION_ID: self.session_id}, mgmt_handlers.default ) session_state = response.get(MGMT_RESPONSE_SESSION_STATE) if isinstance(session_state, six.binary_type): session_state = session_state.decode('UTF-8') return session_state
[docs] async def set_session_state(self, state): # type: (Union[str, bytes, bytearray]) -> None """Set the session state. :param state: The state value. :type state: str, bytes or bytearray .. admonition:: Example: .. literalinclude:: ../samples/async_samples/sample_code_servicebus_async.py :start-after: [START set_session_state_async] :end-before: [END set_session_state_async] :language: python :dedent: 4 :caption: Set the session state """ self._check_live() state = state.encode(self._encoding) if isinstance(state, six.text_type) else state return await self._receiver._mgmt_request_response_with_retry( # pylint: disable=protected-access REQUEST_RESPONSE_SET_SESSION_STATE_OPERATION, {MGMT_REQUEST_SESSION_ID: self.session_id, MGMT_REQUEST_SESSION_STATE: bytearray(state)}, mgmt_handlers.default )
[docs] async def renew_lock(self): # type: () -> None """Renew the session lock. This operation must be performed periodically in order to retain a lock on the session to continue message processing. Once the lock is lost the connection will be closed. This operation can also be performed as a threaded background task by registering the session with an `azure.servicebus.aio.AutoLockRenew` instance. .. admonition:: Example: .. literalinclude:: ../samples/async_samples/sample_code_servicebus_async.py :start-after: [START session_renew_lock_async] :end-before: [END session_renew_lock_async] :language: python :dedent: 4 :caption: Renew the session lock before it expires """ self._check_live() expiry = await self._receiver._mgmt_request_response_with_retry( # pylint: disable=protected-access REQUEST_RESPONSE_RENEW_SESSION_LOCK_OPERATION, {MGMT_REQUEST_SESSION_ID: self.session_id}, mgmt_handlers.default ) self._locked_until_utc = utc_from_timestamp(expiry[MGMT_RESPONSE_RECEIVER_EXPIRATION]/1000.0)