PROTON-2407 Introduce initial optional typing annotations to Python binding (#325)
diff --git a/python/proton/_common.py b/python/proton/_common.py
index 3715c6a..bc03b46 100644
--- a/python/proton/_common.py
+++ b/python/proton/_common.py
@@ -17,50 +17,35 @@
# under the License.
#
-
-#
-# Hacks to provide Python2 <---> Python3 compatibility
-#
-# The results are
-# | |long|unicode|
-# |python2|long|unicode|
-# |python3| int| str|
-try:
- long()
-except NameError:
- long = int
-try:
- unicode()
-except NameError:
- unicode = str
+from typing import Optional, Union, Any
-def isinteger(value):
- return isinstance(value, (int, long))
+def isinteger(value: Any) -> bool:
+ return isinstance(value, int)
-def isstring(value):
- return isinstance(value, (str, unicode))
+def isstring(value: Any) -> bool:
+ return isinstance(value, str)
class Constant(object):
- def __init__(self, name):
+ def __init__(self, name: str) -> None:
self.name = name
- def __repr__(self):
+ def __repr__(self) -> str:
return self.name
-def secs2millis(secs):
- return long(secs * 1000)
+def secs2millis(secs: Union[float, int]) -> int:
+ return int(secs * 1000)
-def millis2secs(millis):
+def millis2secs(millis: int) -> float:
return float(millis) / 1000.0
-def unicode2utf8(string):
+def unicode2utf8(string: Optional[str]) -> Optional[str]:
"""Some Proton APIs expect a null terminated string. Convert python text
types to UTF8 to avoid zero bytes introduced by other multi-byte encodings.
This method will throw if the string cannot be converted.
@@ -68,24 +53,18 @@
if string is None:
return None
elif isinstance(string, str):
- # Must be py2 or py3 str
- # The swig binding converts py3 str -> utf8 char* and back sutomatically
+ # The swig binding converts py3 str -> utf8 char* and back automatically
return string
- elif isinstance(string, unicode):
- # This must be python2 unicode as we already detected py3 str above
- return string.encode('utf-8')
# Anything else illegal - specifically python3 bytes
raise TypeError("Unrecognized string type: %r (%s)" % (string, type(string)))
-def utf82unicode(string):
+def utf82unicode(string: Optional[Union[str, bytes]]) -> Optional[str]:
"""Convert C strings returned from proton-c into python unicode"""
if string is None:
return None
- elif isinstance(string, unicode):
- # py2 unicode, py3 str (via hack definition)
+ elif isinstance(string, str):
return string
elif isinstance(string, bytes):
- # py2 str (via hack definition), py3 bytes
return string.decode('utf8')
raise TypeError("Unrecognized string type")
diff --git a/python/proton/_condition.py b/python/proton/_condition.py
index 338c776..445c912 100644
--- a/python/proton/_condition.py
+++ b/python/proton/_condition.py
@@ -17,7 +17,7 @@
# under the License.
#
-from __future__ import absolute_import
+from typing import Optional
from cproton import pn_condition_clear, pn_condition_set_name, pn_condition_set_description, pn_condition_info, \
pn_condition_is_set, pn_condition_get_name, pn_condition_get_description
@@ -62,12 +62,12 @@
self.description = description
self.info = info
- def __repr__(self):
+ def __repr__(self) -> str:
return "Condition(%s)" % ", ".join([repr(x) for x in
(self.name, self.description, self.info)
if x])
- def __eq__(self, o):
+ def __eq__(self, o: 'Condition') -> bool:
if not isinstance(o, Condition):
return False
return self.name == o.name and \
@@ -75,7 +75,7 @@
self.info == o.info
-def obj2cond(obj, cond):
+def obj2cond(obj, cond: Condition) -> None:
pn_condition_clear(cond)
if obj:
pn_condition_set_name(cond, str(obj.name))
@@ -85,7 +85,7 @@
info.put_object(obj.info)
-def cond2obj(cond):
+def cond2obj(cond) -> Optional[Condition]:
if pn_condition_is_set(cond):
return Condition(pn_condition_get_name(cond),
pn_condition_get_description(cond),
diff --git a/python/proton/_data.py b/python/proton/_data.py
index 7d63395..6e91a37 100644
--- a/python/proton/_data.py
+++ b/python/proton/_data.py
@@ -17,9 +17,14 @@
# under the License.
#
-from __future__ import absolute_import
-
import uuid
+from typing import Callable, List, Tuple, Union, Optional, Any, Dict, Iterable, overload, TypeVar
+try:
+ from typing import Literal
+except ImportError:
+ class Literal:
+ def __class_getitem__(cls, item):
+ pass
from cproton import PN_ARRAY, PN_BINARY, PN_BOOL, PN_BYTE, PN_CHAR, PN_DECIMAL128, PN_DECIMAL32, PN_DECIMAL64, \
PN_DESCRIBED, PN_DOUBLE, PN_FLOAT, PN_INT, PN_LIST, PN_LONG, PN_MAP, PN_NULL, PN_OVERFLOW, PN_SHORT, PN_STRING, \
@@ -43,13 +48,21 @@
long = int
unicode = str
+_T = TypeVar('_T')
+
+PythonAMQPData = Union[
+ Dict['PythonAMQPData', 'PythonAMQPData'],
+ List['PythonAMQPData'],
+ 'Described', 'Array', int, str, 'symbol', bytes, float, None]
+"""This type annotation represents Python data structures that can be encoded as AMQP Data"""
+
class UnmappedType:
- def __init__(self, msg):
+ def __init__(self, msg: str) -> None:
self.msg = msg
- def __repr__(self):
+ def __repr__(self) -> str:
return "UnmappedType(%s)" % self.msg
@@ -60,12 +73,12 @@
An unsigned 64 bit integer in the range :math:`0` to :math:`2^{64} - 1` inclusive.
"""
- def __init__(self, l):
- if (l < 0):
+ def __init__(self, l: int) -> None:
+ if l < 0:
raise AssertionError("initializing ulong with negative value")
super(ulong, self).__new__(ulong, l)
- def __repr__(self):
+ def __repr__(self) -> str:
return "ulong(%s)" % long.__repr__(self)
@@ -79,7 +92,7 @@
example, ``1311704463521`` represents the moment ``2011-07-26T18:21:03.521Z``.
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "timestamp(%s)" % long.__repr__(self)
@@ -90,7 +103,7 @@
Symbolic values from a constrained domain, represented by a sequence of ASCII characters.
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "symbol(%s)" % unicode.__repr__(self)
@@ -101,7 +114,7 @@
A 32 bit UTF-32BE encoded Unicode character.
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "char(%s)" % unicode.__repr__(self)
@@ -112,7 +125,7 @@
An 8 bit signed integer in the range :math:`-(2^7)` to :math:`2^7 - 1` inclusive.
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "byte(%s)" % int.__repr__(self)
@@ -123,7 +136,7 @@
A 16 bit signed integer in the range :math:`-(2^{15})` to :math:`2^{15} - 1` inclusive.
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "short(%s)" % int.__repr__(self)
@@ -134,7 +147,7 @@
A 32 bit signed integer in the range :math:`-(2^{31})` to :math:`2^{31} - 1` inclusive.
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "int32(%s)" % int.__repr__(self)
@@ -145,12 +158,12 @@
An 8 bit unsigned integer in the range :math:`0` to :math:`2^8 - 1` inclusive.
"""
- def __init__(self, i):
- if (i < 0):
+ def __init__(self, i: int) -> None:
+ if i < 0:
raise AssertionError("initializing ubyte with negative value")
super(ubyte, self).__new__(ubyte, i)
- def __repr__(self):
+ def __repr__(self) -> str:
return "ubyte(%s)" % int.__repr__(self)
@@ -161,12 +174,12 @@
A 16 bit unsigned integer in the range :math:`0` to :math:`2^{16} - 1` inclusive.
"""
- def __init__(self, i):
- if (i < 0):
+ def __init__(self, i: int) -> None:
+ if i < 0:
raise AssertionError("initializing ushort with negative value")
super(ushort, self).__new__(ushort, i)
- def __repr__(self):
+ def __repr__(self) -> str:
return "ushort(%s)" % int.__repr__(self)
@@ -177,12 +190,12 @@
A 32 bit unsigned integer in the range :math:`0` to :math:`2^{32} - 1` inclusive.
"""
- def __init__(self, l):
- if (l < 0):
+ def __init__(self, l: int) -> None:
+ if l < 0:
raise AssertionError("initializing uint with negative value")
super(uint, self).__new__(uint, l)
- def __repr__(self):
+ def __repr__(self) -> str:
return "uint(%s)" % long.__repr__(self)
@@ -193,7 +206,7 @@
A 32 bit floating point number (IEEE 754-2008 binary32).
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "float32(%s)" % float.__repr__(self)
@@ -204,7 +217,7 @@
A 32 bit decimal floating point number (IEEE 754-2008 decimal32).
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "decimal32(%s)" % int.__repr__(self)
@@ -215,7 +228,7 @@
A 64 bit decimal floating point number (IEEE 754-2008 decimal64).
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "decimal64(%s)" % long.__repr__(self)
@@ -226,7 +239,7 @@
A 128-bit decimal floating-point number (IEEE 754-2008 decimal128).
"""
- def __repr__(self):
+ def __repr__(self) -> str:
return "decimal128(%s)" % bytes.__repr__(self)
@@ -234,20 +247,22 @@
"""
A described AMQP type.
- :ivar descriptor: A symbol describing the value.
- :vartype descriptor: :class:`symbol`
+ :ivar descriptor: Any AMQP value can be a descriptor
:ivar value: The described value
- :vartype value: Any AMQP value
"""
- def __init__(self, descriptor, value):
+ def __init__(
+ self,
+ descriptor: PythonAMQPData,
+ value: PythonAMQPData,
+ ) -> None:
self.descriptor = descriptor
self.value = value
- def __repr__(self):
+ def __repr__(self) -> str:
return "Described(%r, %r)" % (self.descriptor, self.value)
- def __eq__(self, o):
+ def __eq__(self, o: Any) -> bool:
if isinstance(o, Described):
return self.descriptor == o.descriptor and self.value == o.value
else:
@@ -271,7 +286,12 @@
:ivar elements: A Python list of elements of the appropriate type.
"""
- def __init__(self, descriptor, type, *elements):
+ def __init__(
+ self,
+ descriptor: PythonAMQPData,
+ type: int,
+ *elements
+ ) -> None:
self.descriptor = descriptor
self.type = type
self.elements = elements
@@ -279,14 +299,14 @@
def __iter__(self):
return iter(self.elements)
- def __repr__(self):
+ def __repr__(self) -> str:
if self.elements:
els = ", %s" % (", ".join(map(repr, self.elements)))
else:
els = ""
return "Array(%r, %r%s)" % (self.descriptor, self.type, els)
- def __eq__(self, o):
+ def __eq__(self, o: Any) -> bool:
if isinstance(o, Array):
return self.descriptor == o.descriptor and \
self.type == o.type and self.elements == o.elements
@@ -294,40 +314,46 @@
return False
-def _check_type(s, allow_ulong=False, raise_on_error=True):
+def _check_type(
+ s: _T,
+ allow_ulong: bool = False,
+ raise_on_error: bool = True
+) -> Union[symbol, ulong, _T]:
if isinstance(s, symbol):
return s
if allow_ulong and isinstance(s, ulong):
return s
if isinstance(s, str):
- # Must be py2 or py3 str
return symbol(s)
- if isinstance(s, unicode):
- # This must be python2 unicode as we already detected py3 str above
- return symbol(s.encode('utf-8'))
if raise_on_error:
raise TypeError('Non-symbol type %s: %s' % (type(s), s))
return s
-def _check_is_symbol(s, raise_on_error=True):
+def _check_is_symbol(s: _T, raise_on_error: bool = True) -> Union[symbol, ulong, _T]:
return _check_type(s, allow_ulong=False, raise_on_error=raise_on_error)
-def _check_is_symbol_or_ulong(s, raise_on_error=True):
+def _check_is_symbol_or_ulong(s: _T, raise_on_error: bool = True) -> Union[symbol, ulong, _T]:
return _check_type(s, allow_ulong=True, raise_on_error=raise_on_error)
class RestrictedKeyDict(dict):
"""Parent class for :class:`PropertyDict` and :class:`AnnotationDict`"""
- def __init__(self, validation_fn, e=None, raise_on_error=True, **kwargs):
+ def __init__(
+ self,
+ validation_fn: Callable[[_T, bool], _T],
+ e: Optional[Any] = None,
+ raise_on_error: bool = True,
+ **kwargs
+ ) -> None:
super(RestrictedKeyDict, self).__init__()
self.validation_fn = validation_fn
self.raise_on_error = raise_on_error
self.update(e, **kwargs)
- def __setitem__(self, key, value):
+ def __setitem__(self, key: Union[symbol, str], value: Any) -> None:
"""Checks if the key is a :class:`symbol` type before setting the value"""
try:
return super(RestrictedKeyDict, self).__setitem__(self.validation_fn(key, self.raise_on_error), value)
@@ -336,7 +362,7 @@
# __setitem__() must raise a KeyError, not TypeError
raise KeyError('invalid non-symbol key: %s: %s' % (type(key), key))
- def update(self, e=None, **kwargs):
+ def update(self, e: Optional[Any] = None, **kwargs) -> None:
"""
Equivalent to dict.update(), but it was needed to call :meth:`__setitem__()`
instead of ``dict.__setitem__()``.
@@ -385,11 +411,10 @@
is encountered as a key in the initialization, or in a subsequent operation which
adds such an key. If ``False``, non-strings and non-symbols will be added as keys
to the dictionary without an error.
- :type raise_on_error: ``bool``
:param kwargs: Keyword args for initializing a ``dict`` of the form key1=val1, key2=val2, ...
"""
- def __init__(self, e=None, raise_on_error=True, **kwargs):
+ def __init__(self, e: Optional[Any] = None, raise_on_error: bool = True, **kwargs) -> None:
super(PropertyDict, self).__init__(_check_is_symbol, e, raise_on_error, **kwargs)
def __repr__(self):
@@ -424,17 +449,20 @@
>>> AnnotationDict({'one': 1, 2: 'two'}, raise_on_error=False)
AnnotationDict({2: 'two', symbol(u'one'): 1})
- :param e: Initialization for ``dict``
- :type e: ``dict`` or ``list`` of ``tuple`` or ``zip`` object
+ :param e: Initializer for ``dict``: a ``dict`` or ``list`` of ``tuple`` or ``zip`` object
:param raise_on_error: If ``True``, will raise an ``KeyError`` if a non-string, non-symbol or
non-ulong is encountered as a key in the initialization, or in a subsequent
operation which adds such an key. If ``False``, non-strings, non-ulongs and non-symbols
will be added as keys to the dictionary without an error.
- :type raise_on_error: ``bool``
:param kwargs: Keyword args for initializing a ``dict`` of the form key1=val1, key2=val2, ...
"""
- def __init__(self, e=None, raise_on_error=True, **kwargs):
+ def __init__(
+ self,
+ e: Optional[Union[Dict, List, Tuple, Iterable]] = None,
+ raise_on_error: bool = True,
+ **kwargs
+ ) -> None:
super(AnnotationDict, self).__init__(_check_is_symbol_or_ulong, e, raise_on_error, **kwargs)
def __repr__(self):
@@ -464,22 +492,24 @@
>>> SymbolList(['one', symbol('two'), 3], raise_on_error=False)
SymbolList([symbol(u'one'), symbol(u'two'), 3])
- :param t: Initialization for list
- :type t: ``list``
+ :param t: Initializer for list
:param raise_on_error: If ``True``, will raise an ``TypeError`` if a non-string or non-symbol is
encountered in the initialization list, or in a subsequent operation which adds such
an element. If ``False``, non-strings and non-symbols will be added to the list without
an error.
- :type raise_on_error: ``bool``
"""
- def __init__(self, t=None, raise_on_error=True):
+ def __init__(
+ self,
+ t: Optional[List[Any]] = None,
+ raise_on_error: bool = True
+ ) -> None:
super(SymbolList, self).__init__()
self.raise_on_error = raise_on_error
if t:
self.extend(t)
- def _check_list(self, t):
+ def _check_list(self, t: Iterable[Any]) -> List[Any]:
""" Check all items in list are :class:`symbol`s (or are converted to symbols). """
l = []
if t:
@@ -487,19 +517,19 @@
l.append(_check_is_symbol(v, self.raise_on_error))
return l
- def append(self, v):
+ def append(self, v: str) -> None:
""" Add a single value v to the end of the list """
return super(SymbolList, self).append(_check_is_symbol(v, self.raise_on_error))
- def extend(self, t):
+ def extend(self, t: Iterable[str]) -> None:
""" Add all elements of an iterable t to the end of the list """
return super(SymbolList, self).extend(self._check_list(t))
- def insert(self, i, v):
+ def insert(self, i: int, v: str) -> None:
""" Insert a value v at index i """
return super(SymbolList, self).insert(i, _check_is_symbol(v, self.raise_on_error))
- def __add__(self, t):
+ def __add__(self, t: Iterable[Any]) -> 'SymbolList':
""" Handles list1 + list2 """
return SymbolList(super(SymbolList, self).__add__(self._check_list(t)), raise_on_error=self.raise_on_error)
@@ -507,11 +537,11 @@
""" Handles list1 += list2 """
return super(SymbolList, self).__iadd__(self._check_list(t))
- def __setitem__(self, i, t):
+ def __setitem__(self, i: int, t: Any) -> None:
""" Handles list[i] = v """
return super(SymbolList, self).__setitem__(i, _check_is_symbol(t, self.raise_on_error))
- def __repr__(self):
+ def __repr__(self) -> str:
""" Representation of SymbolList """
return 'SymbolList(%s)' % super(SymbolList, self).__repr__()
@@ -626,17 +656,16 @@
"""
@classmethod
- def type_name(amqptype):
+ def type_name(cls, amqptype: int) -> str:
"""
Return a string name for an AMQP type.
:param amqptype: Numeric Proton AMQP type (`enum pn_type_t`)
- :type amqptype: integer
- :rtype: String describing the AMQP type with numeric value `amqptype`
+ :returns: String describing the AMQP type with numeric value `amqptype`
"""
return Data.type_names[amqptype]
- def __init__(self, capacity=16):
+ def __init__(self, capacity: int = 16) -> None:
if isinstance(capacity, (int, long)):
self._data = pn_data(capacity)
self._free = True
@@ -644,25 +673,25 @@
self._data = capacity
self._free = False
- def __del__(self):
+ def __del__(self) -> None:
if self._free and hasattr(self, "_data"):
pn_data_free(self._data)
del self._data
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, DataException)
raise exc("[%s]: %s" % (err, pn_error_text(pn_data_error(self._data))))
else:
return err
- def clear(self):
+ def clear(self) -> None:
"""
Clears the data object.
"""
pn_data_clear(self._data)
- def rewind(self):
+ def rewind(self) -> None:
"""
Clears current node and sets the parent to the root node. Clearing the
current node sets it _before_ the first node, calling next() will advance to
@@ -671,14 +700,13 @@
assert self._data is not None
pn_data_rewind(self._data)
- def next(self):
+ def next(self) -> Optional[int]:
"""
Advances the current node to its next sibling and returns its
type. If there is no next sibling the current node remains
unchanged and ``None`` is returned.
:return: Node type or ``None``
- :rtype: ``int`` or ``None``
"""
found = pn_data_next(self._data)
if found:
@@ -686,14 +714,13 @@
else:
return None
- def prev(self):
+ def prev(self) -> Optional[int]:
"""
Advances the current node to its previous sibling and returns its
type. If there is no previous sibling the current node remains
unchanged and ``None`` is returned.
:return: Node type or ``None``
- :rtype: ``int`` or ``None``
"""
found = pn_data_prev(self._data)
if found:
@@ -701,7 +728,7 @@
else:
return None
- def enter(self):
+ def enter(self) -> bool:
"""
Sets the parent node to the current node and clears the current node.
Clearing the current node sets it *before* the first child,
@@ -709,25 +736,23 @@
:return: ``True`` iff the pointers to the current/parent nodes are changed,
``False`` otherwise.
- :rtype: ``bool``
"""
return pn_data_enter(self._data)
- def exit(self):
+ def exit(self) -> bool:
"""
Sets the current node to the parent node and the parent node to
its own parent.
:return: ``True`` iff the pointers to the current/parent nodes are changed,
``False`` otherwise.
- :rtype: ``bool``
"""
return pn_data_exit(self._data)
- def lookup(self, name):
+ def lookup(self, name: str) -> bool:
return pn_data_lookup(self._data, name)
- def narrow(self):
+ def narrow(self) -> None:
"""
Modify this :class:`Data` object to behave as if the current node is the
root node of the tree. This impacts the behavior of :meth:`rewind`,
@@ -737,17 +762,16 @@
"""
pn_data_narrow(self._data)
- def widen(self):
+ def widen(self) -> None:
""" Reverse the effect of :meth:`narrow`. """
pn_data_widen(self._data)
- def type(self):
+ def type(self) -> Optional[int]:
"""
Returns the type of the current node. Returns `None` if there is no
current node.
:return: The current node type enumeration.
- :rtype: ``int`` or ``None``
"""
dtype = pn_data_type(self._data)
if dtype == -1:
@@ -755,12 +779,11 @@
else:
return dtype
- def encoded_size(self):
+ def encoded_size(self) -> int:
"""
Returns the size in bytes needed to encode the data in AMQP format.
:return: The size of the encoded data or an error code if data is invalid.
- :rtype: ``int``
"""
return pn_data_encoded_size(self._data)
@@ -793,7 +816,7 @@
"""
return self._check(pn_data_decode(self._data, encoded))
- def put_list(self):
+ def put_list(self) -> None:
"""
Puts a list value. Elements may be filled by entering the list
node and putting element values.
@@ -810,7 +833,7 @@
"""
self._check(pn_data_put_list(self._data))
- def put_map(self):
+ def put_map(self) -> None:
"""
Puts a map value. Elements may be filled by entering the map node
and putting alternating key value pairs.
@@ -826,7 +849,7 @@
"""
self._check(pn_data_put_map(self._data))
- def put_array(self, described, element_type):
+ def put_array(self, described: bool, element_type: int) -> None:
"""
Puts an array value. Elements may be filled by entering the array
node and putting the element values. The values must all be of the
@@ -859,7 +882,7 @@
"""
self._check(pn_data_put_array(self._data, described, element_type))
- def put_described(self):
+ def put_described(self) -> None:
"""
Puts a described value. A described node has two children, the
descriptor and the value. These are specified by entering the node
@@ -876,7 +899,7 @@
"""
self._check(pn_data_put_described(self._data))
- def put_null(self):
+ def put_null(self) -> None:
"""
Puts a null value.
@@ -884,236 +907,214 @@
"""
self._check(pn_data_put_null(self._data))
- def put_bool(self, b):
+ def put_bool(self, b: Union[bool, int]) -> None:
"""
Puts a boolean value.
:param b: a boolean value
- :type b: ``bool`` or ``int``
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_bool(self._data, b))
- def put_ubyte(self, ub):
+ def put_ubyte(self, ub: Union[ubyte, int]) -> None:
"""
Puts an unsigned byte value.
:param ub: an integral value in the range :math:`0` to :math:`2^8 - 1` inclusive
- :type ub: ``int``, :class:`ubyte`
:raise: * ``AssertionError`` if parameter is out of the range :math:`0` to :math:`2^8 - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_ubyte(self._data, ub))
- def put_byte(self, b):
+ def put_byte(self, b: Union[byte, int]) -> None:
"""
Puts a signed byte value.
:param b: an integral value in the range :math:`-(2^7)` to :math:`2^7 - 1` inclusive.
- :type b: ``int``, :class:`byte`
:raise: * ``AssertionError`` if parameter is out of the range :math:`-(2^7)` to :math:`2^7 - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_byte(self._data, b))
- def put_ushort(self, us):
+ def put_ushort(self, us: Union[ushort, int]) -> None:
"""
Puts an unsigned short value.
:param us: an integral value in the range :math:`0` to :math:`2^{16} - 1` inclusive.
- :type us: ``int``, :class:`ushort`
:raise: * ``AssertionError`` if parameter is out of the range :math:`0` to :math:`2^{16} - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_ushort(self._data, us))
- def put_short(self, s):
+ def put_short(self, s: Union[short, int]) -> None:
"""
Puts a signed short value.
:param s: an integral value in the range :math:`-(2^{15})` to :math:`2^{15} - 1` inclusive.
- :type s: ``int``, :class:`short`
:raise: * ``AssertionError`` if parameter is out of the range :math:`-(2^{15})` to :math:`2^{15} - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_short(self._data, s))
- def put_uint(self, ui):
+ def put_uint(self, ui: Union[uint, int]) -> None:
"""
Puts an unsigned int value.
:param ui: an integral value in the range :math:`0` to :math:`2^{32} - 1` inclusive.
- :type ui: ``int``, :class:`uint`
:raise: * ``AssertionError`` if parameter is out of the range :math:`0` to :math:`2^{32} - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_uint(self._data, ui))
- def put_int(self, i):
+ def put_int(self, i: Union[int32, int]) -> None:
"""
Puts a signed int value.
:param i: an integral value in the range :math:`-(2^{31})` to :math:`2^{31} - 1` inclusive.
- :type i: ``int``, :class:`int32`
:raise: * ``AssertionError`` if parameter is out of the range :math:`-(2^{31})` to :math:`2^{31} - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_int(self._data, i))
- def put_char(self, c):
+ def put_char(self, c: Union[char, str]) -> None:
"""
Puts a char value.
:param c: a single character
- :type c: ``str``, :class:`char`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_char(self._data, ord(c)))
- def put_ulong(self, ul):
+ def put_ulong(self, ul: Union[ulong, int]) -> None:
"""
Puts an unsigned long value.
:param ul: an integral value in the range :math:`0` to :math:`2^{64} - 1` inclusive.
- :type ul: ``int``, ``long``, :class:`ulong`
:raise: * ``AssertionError`` if parameter is out of the range :math:`0` to :math:`2^{64} - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_ulong(self._data, ul))
- def put_long(self, l):
+ def put_long(self, l: Union[long, int]) -> None:
"""
Puts a signed long value.
:param l: an integral value in the range :math:`-(2^{63})` to :math:`2^{63} - 1` inclusive.
- :type l: ``int``, ``long``
:raise: * ``AssertionError`` if parameter is out of the range :math:`-(2^{63})` to :math:`2^{63} - 1` inclusive.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_long(self._data, l))
- def put_timestamp(self, t):
+ def put_timestamp(self, t: Union[timestamp, int]) -> None:
"""
Puts a timestamp value.
:param t: a positive integral value
- :type t: ``int``, :class:`timestamp`
:raise: * ``AssertionError`` if parameter is negative.
* :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_timestamp(self._data, t))
- def put_float(self, f):
+ def put_float(self, f: Union[float32, float, int]) -> None:
"""
Puts a float value.
:param f: a floating point value
- :type f: ``float``, :class:`float32`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_float(self._data, f))
- def put_double(self, d):
+ def put_double(self, d: Union[float, int]) -> None:
"""
Puts a double value.
:param d: a floating point value.
- :type d: ``double``
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_double(self._data, d))
- def put_decimal32(self, d):
+ def put_decimal32(self, d: Union[decimal32, int]) -> None:
"""
Puts a decimal32 value.
:param d: a decimal32 number encoded in an 32-bit integral value.
- :type d: ``int``, :class:`decimal32`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_decimal32(self._data, d))
- def put_decimal64(self, d):
+ def put_decimal64(self, d: Union[decimal64, int]) -> None:
"""
Puts a decimal64 value.
:param d: a decimal64 number encoded in an 32-bit integral value.
- :type d: ``int``, ``long``, :class:`decimal64`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_decimal64(self._data, d))
- def put_decimal128(self, d):
+ def put_decimal128(self, d: Union[decimal128, bytes]) -> None:
"""
Puts a decimal128 value.
:param d: a decimal128 value encoded in a 16-byte binary value.
- :type d: ``bytes``, :class:`decimal128`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_decimal128(self._data, d))
- def put_uuid(self, u):
+ def put_uuid(self, u: uuid.UUID) -> None:
"""
Puts a UUID value.
:param u: a uuid value.
- :type u: ``uuid.UUID``
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_uuid(self._data, u.bytes))
- def put_binary(self, b):
+ def put_binary(self, b: bytes) -> None:
"""
Puts a binary value.
:param b: a binary value
- :type b: ``bytes``
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_binary(self._data, b))
- def put_memoryview(self, mv):
+ def put_memoryview(self, mv: memoryview) -> None:
"""
Put a Python memoryview object as an AMQP binary value.
:param mv: A Python memoryview object
- :type mv: ``memoryview``
:raise: :exc:`DataException` if there is a Proton error.
"""
self.put_binary(mv.tobytes())
- def put_buffer(self, buff):
+ def put_buffer(self, buff: Iterable[int]) -> None:
"""
Put a Python buffer object as an AMQP binary value.
:param buff: A Python buffer object (**CHECK THIS**)
- :type buff: Any object supporting the Python buffer interface.
:raise: :exc:`DataException` if there is a Proton error.
"""
self.put_binary(bytes(buff))
- def put_string(self, s):
+ def put_string(self, s: str) -> None:
"""
Puts a unicode value.
:param s: a unicode string
- :type s: ``str`` (Python 3.x) or ``unicode`` (Python 2.x)
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_string(self._data, s.encode("utf8")))
- def put_symbol(self, s):
+ def put_symbol(self, s: Union[str, symbol]) -> None:
"""
Puts a symbolic value.
:param s: the symbol name
- :type s: string, :class:`symbol`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_put_symbol(self._data, s.encode('ascii')))
- def get_list(self):
+ def get_list(self) -> int:
"""
If the current node is a list, return the number of elements,
otherwise return 0. List elements can be accessed by entering
@@ -1130,11 +1131,10 @@
>>> data.exit()
:return: the number of child elements of a list node
- :rtype: ``int``
"""
return pn_data_get_list(self._data)
- def get_map(self):
+ def get_map(self) -> int:
"""
If the current node is a map, return the number of child elements,
otherwise return 0. Key value pairs can be accessed by entering
@@ -1151,7 +1151,6 @@
>>> data.exit()
:return: the number of child elements of a map node
- :rtype: ``int``
"""
return pn_data_get_map(self._data)
@@ -1183,7 +1182,7 @@
type = None
return count, described, type
- def is_described(self):
+ def is_described(self) -> bool:
"""
Checks if the current node is a described value. The descriptor
and value may be accessed by entering the described value.
@@ -1198,223 +1197,199 @@
>>> data.exit()
:return: ``True`` if the current node is a described type, ``False`` otherwise.
- :rtype: ``bool``
"""
return pn_data_is_described(self._data)
- def is_null(self):
+ def is_null(self) -> bool:
"""
Checks if the current node is the AMQP null type.
:return: ``True`` if the current node is the AMQP null type, ``False`` otherwise.
- :rtype: ``bool``
"""
return pn_data_is_null(self._data)
- def get_bool(self):
+ def get_bool(self) -> bool:
"""
Get the current node value as a ``bool``.
:return: If the current node is a boolean type, returns its value,
``False`` otherwise.
- :rtype: ``bool``
"""
return pn_data_get_bool(self._data)
- def get_ubyte(self):
+ def get_ubyte(self) -> ubyte:
"""
Get the current node value as a :class:`ubyte`.
:return: If the current node is an unsigned byte, its value, 0 otherwise.
- :rtype: :class:`ubyte`
"""
return ubyte(pn_data_get_ubyte(self._data))
- def get_byte(self):
+ def get_byte(self) -> byte:
"""
Get the current node value as a :class:`byte`.
:return: If the current node is a signed byte, its value, 0 otherwise.
- :rtype: :class:`byte`
"""
return byte(pn_data_get_byte(self._data))
- def get_ushort(self):
+ def get_ushort(self) -> ushort:
"""
Get the current node value as a :class:`ushort`.
:return: If the current node is an unsigned short, its value, 0 otherwise.
- :rtype: :class:`ushort`
"""
return ushort(pn_data_get_ushort(self._data))
- def get_short(self):
+ def get_short(self) -> short:
"""
Get the current node value as a :class:`short`.
:return: If the current node is a signed short, its value, 0 otherwise.
- :rtype: :class:`short`
"""
return short(pn_data_get_short(self._data))
- def get_uint(self):
+ def get_uint(self) -> uint:
"""
Get the current node value as a :class:`uint`.
:return: If the current node is an unsigned int, its value, 0 otherwise.
- :rtype: :class:`uint`
"""
return uint(pn_data_get_uint(self._data))
- def get_int(self):
+ def get_int(self) -> int32:
"""
Get the current node value as a :class:`int32`.
:return: If the current node is a signed int, its value, 0 otherwise.
- :rtype: :class:`int32`
"""
return int32(pn_data_get_int(self._data))
- def get_char(self):
+ def get_char(self) -> char:
"""
Get the current node value as a :class:`char`.
:return: If the current node is a char, its value, 0 otherwise.
- :rtype: :class:`char`
"""
return char(chr(pn_data_get_char(self._data)))
- def get_ulong(self):
- """
- Get the current node value as a :class:`long`.
-
- :return: If the current node is an unsigned long, its value, 0 otherwise.
- :rtype: :class:`ulong`
- """
- return ulong(pn_data_get_ulong(self._data))
-
- def get_long(self):
+ def get_ulong(self) -> ulong:
"""
Get the current node value as a :class:`ulong`.
+ :return: If the current node is an unsigned long, its value, 0 otherwise.
+ """
+ return ulong(pn_data_get_ulong(self._data))
+
+ def get_long(self) -> long:
+ """
+ Get the current node value as a :class:`long`.
+
:return: If the current node is an signed long, its value, 0 otherwise.
- :rtype: :class:`long`
"""
return long(pn_data_get_long(self._data))
- def get_timestamp(self):
+ def get_timestamp(self) -> timestamp:
"""
Get the current node value as a :class:`timestamp`.
:return: If the current node is a timestamp, its value, 0 otherwise.
- :rtype: :class:`timestamp`
"""
return timestamp(pn_data_get_timestamp(self._data))
- def get_float(self):
+ def get_float(self) -> float32:
"""
Get the current node value as a :class:`float32`.
:return: If the current node is a float, its value, 0 otherwise.
- :rtype: :class:`float32`
"""
return float32(pn_data_get_float(self._data))
- def get_double(self):
+ def get_double(self) -> float:
"""
Get the current node value as a ``double``.
:return: If the current node is a double, its value, 0 otherwise.
- :rtype: ``double``
"""
return pn_data_get_double(self._data)
# XXX: need to convert
- def get_decimal32(self):
+ def get_decimal32(self) -> decimal32:
"""
Get the current node value as a :class:`decimal32`.
:return: If the current node is a decimal32, its value, 0 otherwise.
- :rtype: :class:`decimal32`
"""
return decimal32(pn_data_get_decimal32(self._data))
# XXX: need to convert
- def get_decimal64(self):
+ def get_decimal64(self) -> decimal64:
"""
Get the current node value as a :class:`decimal64`.
:return: If the current node is a decimal64, its value, 0 otherwise.
- :rtype: :class:`decimal64`
"""
return decimal64(pn_data_get_decimal64(self._data))
# XXX: need to convert
- def get_decimal128(self):
+ def get_decimal128(self) -> decimal128:
"""
Get the current node value as a :class:`decimal128`.
:return: If the current node is a decimal128, its value, 0 otherwise.
- :rtype: :class:`decimal128`
"""
return decimal128(pn_data_get_decimal128(self._data))
- def get_uuid(self):
+ def get_uuid(self) -> Optional[uuid.UUID]:
"""
Get the current node value as a ``uuid.UUID``.
:return: If the current node is a UUID, its value, ``None`` otherwise.
- :rtype: ``uuid.UUID`` or ``None``
"""
if pn_data_type(self._data) == Data.UUID:
return uuid.UUID(bytes=pn_data_get_uuid(self._data))
else:
return None
- def get_binary(self):
+ def get_binary(self) -> bytes:
"""
Get the current node value as ``bytes``.
- :return: If the current node is binary, its value, ``""`` otherwise.
- :rtype: ``bytes``
+ :return: If the current node is binary, its value, ``b""`` otherwise.
"""
return pn_data_get_binary(self._data)
- def get_string(self):
+ def get_string(self) -> str:
"""
Get the current node value as ``str``.
:return: If the current node is a string, its value, ``""`` otherwise.
- :rtype: ``str``
"""
return pn_data_get_string(self._data).decode("utf8")
- def get_symbol(self):
+ def get_symbol(self) -> symbol:
"""
Get the current node value as :class:`symbol`.
:return: If the current node is a symbol, its value, ``""`` otherwise.
- :rtype: :class:`symbol`
"""
return symbol(pn_data_get_symbol(self._data).decode('ascii'))
- def copy(self, src):
+ def copy(self, src: 'Data') -> None:
"""
Copy the contents of another pn_data_t object. Any values in the
data object will be lost.
:param src: The source object from which to copy
- :type src: :class:`Data`
:raise: :exc:`DataException` if there is a Proton error.
"""
self._check(pn_data_copy(self._data, src._data))
- def format(self):
+ def format(self) -> str:
"""
Formats the contents of this :class:`Data` object in a human readable way.
:return: A Formatted string containing contents of this :class:`Data` object.
- :rtype: ``str``
:raise: :exc:`DataException` if there is a Proton error.
"""
sz = 16
@@ -1427,7 +1402,7 @@
self._check(err)
return result
- def dump(self):
+ def dump(self) -> None:
"""
Dumps a debug representation of the internal state of this :class:`Data`
object that includes its navigational state to ``cout`` (``stdout``) for
@@ -1435,13 +1410,12 @@
"""
pn_data_dump(self._data)
- def put_dict(self, d):
+ def put_dict(self, d: Dict[Any, Any]) -> None:
"""
A convenience method for encoding the contents of a Python ``dict``
as an AMQP map.
:param d: The dictionary to be encoded
- :type d: ``dict``
:raise: :exc:`DataException` if there is a Proton error.
"""
self.put_map()
@@ -1453,12 +1427,11 @@
finally:
self.exit()
- def get_dict(self):
+ def get_dict(self) -> Dict[Any, Any]:
"""
A convenience method for decoding an AMQP map as a Python ``dict``.
:returns: The decoded dictionary.
- :rtype: ``dict``
"""
if self.enter():
try:
@@ -1474,13 +1447,12 @@
self.exit()
return result
- def put_sequence(self, s):
+ def put_sequence(self, s: List[Any]) -> None:
"""
A convenience method for encoding a Python ``list`` as an
AMQP list.
:param s: The sequence to be encoded
- :type s: ``list``
:raise: :exc:`DataException` if there is a Proton error.
"""
self.put_list()
@@ -1491,12 +1463,11 @@
finally:
self.exit()
- def get_sequence(self):
+ def get_sequence(self) -> List[Any]:
"""
A convenience method for decoding an AMQP list as a Python ``list``.
:returns: The decoded list.
- :rtype: ``list``
"""
if self.enter():
try:
@@ -1507,12 +1478,11 @@
self.exit()
return result
- def get_py_described(self):
+ def get_py_described(self) -> Described:
"""
A convenience method for decoding an AMQP described value.
:returns: The decoded AMQP descriptor.
- :rtype: :class:`Described`
"""
if self.enter():
try:
@@ -1524,7 +1494,7 @@
self.exit()
return Described(descriptor, value)
- def put_py_described(self, d):
+ def put_py_described(self, d: Described) -> None:
"""
A convenience method for encoding a :class:`Described` object
as an AMQP described value. This method encapsulates all the steps
@@ -1542,7 +1512,7 @@
finally:
self.exit()
- def get_py_array(self):
+ def get_py_array(self) -> Optional[Array]:
"""
A convenience method for decoding an AMQP array into an
:class:`Array` object. This method encapsulates all the
@@ -1552,7 +1522,6 @@
representing the array and its contents. Otherwise return ``None``.
:returns: The decoded AMQP array.
- :rtype: :class:`Array`
"""
count, described, type = self.get_array()
@@ -1572,14 +1541,13 @@
self.exit()
return Array(descriptor, type, *elements)
- def put_py_array(self, a):
+ def put_py_array(self, a: Array) -> None:
"""
A convenience method for encoding an :class:`Array` object as
an AMQP array. This method encapsulates the steps described in
:func:`put_array` into a single function.
:param a: The array object to be encoded
- :type a: :class:`Array`
:raise: :exc:`DataException` if there is a Proton error.
"""
described = a.descriptor != UNDESCRIBED
@@ -1653,11 +1621,11 @@
MAP: get_dict
}
- def put_object(self, obj):
+ def put_object(self, obj: Any) -> None:
putter = self.put_mappings[obj.__class__]
putter(self, obj)
- def get_object(self):
+ def get_object(self) -> Optional[Any]:
type = self.type()
if type is None:
return None
diff --git a/python/proton/_delivery.py b/python/proton/_delivery.py
index a9243df..7104753 100644
--- a/python/proton/_delivery.py
+++ b/python/proton/_delivery.py
@@ -17,8 +17,6 @@
# under the License.
#
-from __future__ import absolute_import
-
from cproton import PN_ACCEPTED, PN_MODIFIED, PN_RECEIVED, PN_REJECTED, PN_RELEASED, pn_delivery_abort, \
pn_delivery_aborted, pn_delivery_attachments, pn_delivery_link, pn_delivery_local, pn_delivery_local_state, \
pn_delivery_partial, pn_delivery_pending, pn_delivery_readable, pn_delivery_remote, pn_delivery_remote_state, \
@@ -32,22 +30,28 @@
from ._data import dat2obj, obj2dat
from ._wrapper import Wrapper
+from typing import Dict, List, Optional, Type, Union, TYPE_CHECKING
+if TYPE_CHECKING:
+ from ._condition import Condition
+ from ._endpoints import Receiver, Sender # circular import
+ from ._reactor import Connection, Session, Transport
+
class NamedInt(int):
- values = {} # type: Dict[int, str] # noqa # TODO(PROTON-2323) typing.Dict is not available on Python 2.7
+ values: Dict[int, str] = {}
def __new__(cls, i, name):
ni = super(NamedInt, cls).__new__(cls, i)
cls.values[i] = ni
return ni
- def __init__(self, i, name):
+ def __init__(self, i: int, name: str) -> None:
self.name = name
- def __repr__(self):
+ def __repr__(self) -> str:
return self.name
- def __str__(self):
+ def __str__(self) -> str:
return self.name
@classmethod
@@ -311,7 +315,7 @@
def __init__(self, impl):
Wrapper.__init__(self, impl, pn_delivery_attachments)
- def _init(self):
+ def _init(self) -> None:
self.local = Disposition(pn_delivery_local(self._impl), True)
self.remote = Disposition(pn_delivery_remote(self._impl), False)
@@ -325,43 +329,36 @@
return pn_delivery_tag(self._impl)
@property
- def writable(self):
+ def writable(self) -> bool:
"""
``True`` for an outgoing delivery to which data can now be written,
``False`` otherwise..
-
- :type: ``bool``
"""
return pn_delivery_writable(self._impl)
@property
- def readable(self):
+ def readable(self) -> bool:
"""
``True`` for an incoming delivery that has data to read,
``False`` otherwise..
-
- :type: ``bool``
"""
return pn_delivery_readable(self._impl)
@property
- def updated(self):
+ def updated(self) -> bool:
"""
``True`` if the state of the delivery has been updated
(e.g. it has been settled and/or accepted, rejected etc),
``False`` otherwise.
-
- :type: ``bool``
"""
return pn_delivery_updated(self._impl)
- def update(self, state):
+ def update(self, state: Union[int, DispositionType]) -> None:
"""
Set the local state of the delivery e.g. :const:`ACCEPTED`,
:const:`REJECTED`, :const:`RELEASED`.
:param state: State of delivery
- :type state: ``int``
"""
obj2dat(self.local._data, pn_disposition_data(self.local._impl))
obj2dat(self.local._annotations, pn_disposition_annotations(self.local._impl))
@@ -369,21 +366,17 @@
pn_delivery_update(self._impl, state)
@property
- def pending(self):
+ def pending(self) -> int:
"""
The amount of pending message data for a delivery.
-
- :type: ``int``
"""
return pn_delivery_pending(self._impl)
@property
- def partial(self):
+ def partial(self) -> bool:
"""
``True`` for an incoming delivery if not all the data is
yet available, ``False`` otherwise.
-
- :type: ``bool``
"""
return pn_delivery_partial(self._impl)
@@ -407,16 +400,14 @@
return DispositionType.get(pn_delivery_remote_state(self._impl))
@property
- def settled(self):
+ def settled(self) -> bool:
"""
``True`` if the delivery has been settled by the remote peer,
``False`` otherwise.
-
- :type: ``bool``
"""
return pn_delivery_settled(self._impl)
- def settle(self):
+ def settle(self) -> None:
"""
Settles the delivery locally. This indicates the application
considers the delivery complete and does not wish to receive any
@@ -425,15 +416,13 @@
pn_delivery_settle(self._impl)
@property
- def aborted(self):
+ def aborted(self) -> bool:
"""
``True`` if the delivery has been aborted, ``False`` otherwise.
-
- :type: ``bool``
"""
return pn_delivery_aborted(self._impl)
- def abort(self):
+ def abort(self) -> None:
"""
Aborts the delivery. This indicates the application wishes to
invalidate any data that may have already been sent on this delivery.
@@ -442,51 +431,41 @@
pn_delivery_abort(self._impl)
@property
- def work_next(self):
+ def work_next(self) -> Optional['Delivery']:
"""Deprecated: use on_message(), on_accepted(), on_rejected(),
on_released(), and on_settled() instead.
The next :class:`Delivery` on the connection that has pending
operations.
-
- :type: :class:`Delivery`
"""
return Delivery.wrap(pn_work_next(self._impl))
@property
- def link(self):
+ def link(self) -> Union['Receiver', 'Sender']:
"""
The :class:`Link` on which the delivery was sent or received.
-
- :type: :class:`Link`
"""
from . import _endpoints
return _endpoints.Link.wrap(pn_delivery_link(self._impl))
@property
- def session(self):
+ def session(self) -> 'Session':
"""
The :class:`Session` over which the delivery was sent or received.
-
- :type: :class:`Session`
"""
return self.link.session
@property
- def connection(self):
+ def connection(self) -> 'Connection':
"""
The :class:`Connection` over which the delivery was sent or received.
-
- :type: :class:`Connection`
"""
return self.session.connection
@property
- def transport(self):
+ def transport(self) -> 'Transport':
"""
The :class:`Transport` bound to the :class:`Connection` over which
the delivery was sent or received.
-
- :type: :class:`Transport`
"""
return self.connection.transport
diff --git a/python/proton/_endpoints.py b/python/proton/_endpoints.py
index 7602d17..9520d68 100644
--- a/python/proton/_endpoints.py
+++ b/python/proton/_endpoints.py
@@ -21,8 +21,6 @@
The proton.endpoints module
"""
-from __future__ import absolute_import
-
import weakref
from cproton import PN_CONFIGURATION, PN_COORDINATOR, PN_DELIVERIES, PN_DIST_MODE_COPY, PN_DIST_MODE_MOVE, \
@@ -65,6 +63,12 @@
from ._exceptions import ConnectionException, EXCEPTIONS, LinkException, SessionException
from ._transport import Transport
from ._wrapper import Wrapper
+from typing import Callable, Dict, List, Optional, Union, TYPE_CHECKING
+if TYPE_CHECKING:
+ from ._condition import Condition
+ from ._data import Array, symbol
+ from ._events import Collector, Handler
+ from ._message import Message
class Endpoint(object):
@@ -106,20 +110,18 @@
REMOTE_CLOSED = PN_REMOTE_CLOSED
""" The remote endpoint state is closed. """
- def _init(self):
- self.condition = None
- self._handler = None
+ def _init(self) -> None:
+ self.condition: Optional['Condition'] = None
+ self._handler: Optional['Handler'] = None
- def _update_cond(self):
+ def _update_cond(self) -> None:
obj2cond(self.condition, self._get_cond_impl())
@property
- def remote_condition(self):
+ def remote_condition(self) -> Optional['Condition']:
"""
The remote condition associated with the connection endpoint.
See :class:`Condition` for more information.
-
- :type: :class:`Condition`
"""
return cond2obj(self._get_remote_cond_impl())
@@ -130,10 +132,16 @@
def _get_remote_cond_impl(self):
assert False, "Subclass must override this!"
- def _get_handler(self):
+ @property
+ def handler(self) -> Optional['Handler']:
+ """Handler for events.
+
+ :getter: Get the event handler, or return ``None`` if no handler has been set.
+ :setter: Set the event handler."""
return self._handler
- def _set_handler(self, handler):
+ @handler.setter
+ def handler(self, handler: Optional['Handler']) -> None:
# TODO Hack This is here for some very odd (IMO) backwards compat behaviour
from ._events import Handler
if handler is None:
@@ -144,14 +152,6 @@
self._handler = Handler()
self._handler.add(handler)
- handler = property(_get_handler, _set_handler, doc="""
- Handler for events.
-
- :getter: Get the event handler, or return ``None`` if no handler has been set.
- :setter: Set the event handler.
- :type: :class:`Handler` or ``None``
- """)
-
class Connection(Wrapper, Endpoint):
"""
@@ -168,7 +168,7 @@
def __init__(self, impl=pn_connection):
Wrapper.__init__(self, impl, pn_connection_attachments)
- def _init(self):
+ def _init(self) -> None:
Endpoint._init(self)
self.offered_capabilities = None
self.desired_capabilities = None
@@ -180,25 +180,21 @@
return pn_connection_attachments(self._impl)
@property
- def connection(self):
+ def connection(self) -> 'Connection':
"""
Get this connection.
-
- :type: :class:`Connection`
"""
return self
@property
- def transport(self):
+ def transport(self) -> Optional[Transport]:
"""
The transport bound to this connection. If the connection
is unbound, then this operation will return ``None``.
-
- :type: :class:`Transport` or ``None``
"""
return Transport.wrap(pn_connection_transport(self._impl))
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, ConnectionException)
raise exc("[%s]: %s" % (err, pn_connection_error(self._impl)))
@@ -212,7 +208,7 @@
return pn_connection_remote_condition(self._impl)
# TODO: Blacklisted API call
- def collect(self, collector):
+ def collect(self, collector: 'Collector') -> None:
if collector is None:
pn_connection_collect(self._impl, None)
else:
@@ -402,7 +398,7 @@
obj2dat(self.properties, pn_connection_properties(self._impl))
pn_connection_open(self._impl)
- def close(self):
+ def close(self) -> None:
"""
Closes the connection.
@@ -429,7 +425,7 @@
s.update()
@property
- def state(self):
+ def state(self) -> int:
"""
The state of the connection as a bit field. The state has a local
and a remote component. Each of these can be in one of three
@@ -439,12 +435,11 @@
"""
return pn_connection_state(self._impl)
- def session(self):
+ def session(self) -> 'Session':
"""
Returns a new session on this connection.
:return: New session
- :rtype: :class:`Session`
:raises: :class:`SessionException`
"""
ssn = pn_session(self._impl)
@@ -453,7 +448,7 @@
else:
return Session(ssn)
- def session_head(self, mask):
+ def session_head(self, mask: int) -> Optional['Session']:
"""
Retrieve the first session from a given connection that matches the
specified state mask.
@@ -468,11 +463,10 @@
:param mask: State mask to match
:return: The first session owned by the connection that matches the
mask, else ``None`` if no sessions matches.
- :rtype: :class:`Session` or ``None``
"""
return Session.wrap(pn_session_head(self._impl, mask))
- def link_head(self, mask):
+ def link_head(self, mask: int) -> Optional[Union['Sender', 'Receiver']]:
"""
Retrieve the first link that matches the given state mask.
@@ -484,15 +478,13 @@
set respectively. ``state==0`` matches all links.
:param mask: State mask to match
- :type mask: ``int``
:return: The first link owned by the connection that matches the
mask, else ``None`` if no link matches.
- :rtype: :class:`Link` or ``None``
"""
return Link.wrap(pn_link_head(self._impl, mask))
@property
- def work_head(self):
+ def work_head(self) -> Optional[Delivery]:
"""Deprecated: use on_message(), on_accepted(), on_rejected(),
on_released(), and on_settled() instead.
@@ -526,7 +518,7 @@
"""
return pn_error_code(pn_connection_error(self._impl))
- def free(self):
+ def free(self) -> None:
"""
Releases this connection object.
@@ -659,22 +651,20 @@
return pn_session_outgoing_bytes(self._impl)
@property
- def incoming_bytes(self):
+ def incoming_bytes(self) -> int:
"""
The number of incoming bytes currently buffered.
-
- :type: ``int`` (bytes)
"""
return pn_session_incoming_bytes(self._impl)
- def open(self):
+ def open(self) -> None:
"""
Open a session. Once this operation has completed, the
:const:`LOCAL_ACTIVE` state flag will be set.
"""
pn_session_open(self._impl)
- def close(self):
+ def close(self) -> None:
"""
Close a session.
@@ -705,56 +695,44 @@
return Session.wrap(pn_session_next(self._impl, mask))
@property
- def state(self):
+ def state(self) -> int:
"""
The endpoint state flags for this session. See :class:`Endpoint` for
details of the flags.
-
- :type: ``int``
"""
return pn_session_state(self._impl)
@property
- def connection(self):
+ def connection(self) -> Connection:
"""
The parent connection for this session.
-
- :type: :class:`Connection`
"""
return Connection.wrap(pn_session_connection(self._impl))
@property
- def transport(self):
+ def transport(self) -> Transport:
"""
The transport bound to the parent connection for this session.
-
- :type: :class:`Transport`
"""
return self.connection.transport
- def sender(self, name):
+ def sender(self, name: str) -> 'Sender':
"""
Create a new :class:`Sender` on this session.
:param name: Name of sender
- :type name: ``str``
- :return: New Sender object
- :rtype: :class:`Sender`
"""
return Sender(pn_sender(self._impl, unicode2utf8(name)))
- def receiver(self, name):
+ def receiver(self, name: str) -> 'Receiver':
"""
Create a new :class:`Receiver` on this session.
:param name: Name of receiver
- :type name: ``str``
- :return: New Receiver object
- :rtype: :class:`Receiver`
"""
return Receiver(pn_receiver(self._impl, unicode2utf8(name)))
- def free(self):
+ def free(self) -> None:
"""
Free this session. When a session is freed it will no
longer be retained by the connection once any internal
@@ -796,14 +774,14 @@
def __init__(self, impl):
Wrapper.__init__(self, impl, pn_link_attachments)
- def _init(self):
+ def _init(self) -> None:
Endpoint._init(self)
self.properties = None
def _get_attachments(self):
return pn_link_attachments(self._impl)
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, LinkException)
raise exc("[%s]: %s" % (err, pn_error_text(pn_link_error(self._impl))))
@@ -816,7 +794,7 @@
def _get_remote_cond_impl(self):
return pn_link_remote_condition(self._impl)
- def open(self):
+ def open(self) -> None:
"""
Opens the link.
@@ -828,7 +806,7 @@
obj2dat(self.properties, pn_link_properties(self._impl))
pn_link_open(self._impl)
- def close(self):
+ def close(self) -> None:
"""
Closes the link.
@@ -844,7 +822,7 @@
pn_link_close(self._impl)
@property
- def state(self):
+ def state(self) -> int:
"""
The state of the link as a bit field. The state has a local
and a remote component. Each of these can be in one of three
@@ -853,81 +831,65 @@
:const:`LOCAL_ACTIVE`, :const:`LOCAL_CLOSED`,
:const:`REMOTE_UNINIT`, :const:`REMOTE_ACTIVE` and
:const:`REMOTE_CLOSED`.
-
- :type: ``int``
"""
return pn_link_state(self._impl)
@property
- def source(self):
+ def source(self) -> 'Terminus':
"""
The source of the link as described by the local peer. The
returned object is valid until the link is freed.
-
- :type: :class:`Terminus`
"""
return Terminus(pn_link_source(self._impl))
@property
- def target(self):
+ def target(self) -> 'Terminus':
"""
The target of the link as described by the local peer. The
returned object is valid until the link is freed.
-
- :type: :class:`Terminus`
"""
return Terminus(pn_link_target(self._impl))
@property
- def remote_source(self):
+ def remote_source(self) -> 'Terminus':
"""
The source of the link as described by the remote peer. The
returned object is valid until the link is freed. The remote
:class:`Terminus` object will be empty until the link is
remotely opened as indicated by the :const:`REMOTE_ACTIVE`
flag.
-
- :type: :class:`Terminus`
"""
return Terminus(pn_link_remote_source(self._impl))
@property
- def remote_target(self):
+ def remote_target(self) -> 'Terminus':
"""
The target of the link as described by the remote peer. The
returned object is valid until the link is freed. The remote
:class:`Terminus` object will be empty until the link is
remotely opened as indicated by the :const:`REMOTE_ACTIVE`
flag.
-
- :type: :class:`Terminus`
"""
return Terminus(pn_link_remote_target(self._impl))
@property
- def session(self):
+ def session(self) -> Session:
"""
The parent session for this link.
-
- :type: :class:`Session`
"""
return Session.wrap(pn_link_session(self._impl))
@property
- def connection(self):
+ def connection(self) -> Connection:
"""
The connection on which this link was attached.
-
- :type: :class:`Connection`
"""
return self.session.connection
@property
- def transport(self):
+ def transport(self) -> Transport:
"""
The transport bound to the connection on which this link was attached.
-
- :type: :class:`Transport`
"""
return self.session.transport
@@ -945,7 +907,7 @@
return Delivery(pn_delivery(self._impl, tag))
@property
- def current(self):
+ def current(self) -> Optional[Delivery]:
"""
The current delivery for this link.
@@ -958,12 +920,10 @@
current delivery remains the same until it is changed through
use of :meth:`advance` or until it is settled via
:meth:`Delivery.settle`.
-
- :rtype: :class:`Delivery`
"""
return Delivery.wrap(pn_link_current(self._impl))
- def advance(self):
+ def advance(self) -> bool:
"""
Advance the current delivery of this link to the next delivery.
@@ -984,21 +944,18 @@
:return: ``True`` if the value of the current delivery changed (even
if it was set to ``NULL``, ``False`` otherwise.
- :rtype: ``bool``
"""
return pn_link_advance(self._impl)
@property
- def unsettled(self):
+ def unsettled(self) -> int:
"""
The number of unsettled deliveries for this link.
-
- :type: ``int``
"""
return pn_link_unsettled(self._impl)
@property
- def credit(self):
+ def credit(self) -> int:
"""
The amount of outstanding credit on this link.
@@ -1018,13 +975,11 @@
up being buffered by the link until enough credit is obtained from
the receiver to send them over the wire. In this case the balance
reported by :attr:`credit` will go negative.
-
- :type: ``int``
"""
return pn_link_credit(self._impl)
@property
- def available(self):
+ def available(self) -> int:
"""
The available deliveries hint for this link.
@@ -1032,13 +987,11 @@
deliveries that might be able to be sent if sufficient credit were
issued by the receiving link endpoint. See :meth:`Sender.offered` for
more details.
-
- :type: ``int``
"""
return pn_link_available(self._impl)
@property
- def queued(self):
+ def queued(self) -> int:
"""
The number of queued deliveries for a link.
@@ -1047,12 +1000,10 @@
:meth:`credit`), or they simply may not have yet had a chance to
be written to the wire. This operation will return the number of
queued deliveries on a link.
-
- :type: ``int``
"""
return pn_link_queued(self._impl)
- def next(self, mask):
+ def next(self, mask: int) -> Optional[Union['Sender', 'Receiver']]:
"""
Retrieve the next link that matches the given state mask.
@@ -1062,37 +1013,29 @@
match behavior.
:param mask: State mask to match
- :type mask: ``int``
:return: The next link that matches the given state mask, or
``None`` if no link matches.
- :rtype: :class:`Link`
"""
return Link.wrap(pn_link_next(self._impl, mask))
@property
- def name(self):
+ def name(self) -> str:
"""
The name of the link.
-
- :type: ``str``
"""
return utf82unicode(pn_link_name(self._impl))
@property
- def is_sender(self):
+ def is_sender(self) -> bool:
"""
``True`` if this link is a sender, ``False`` otherwise.
-
- :type: ``bool``
"""
return pn_link_is_sender(self._impl)
@property
- def is_receiver(self):
+ def is_receiver(self) -> bool:
"""
``True`` if this link is a receiver, ``False`` otherwise.
-
- :type: ``bool``
"""
return pn_link_is_receiver(self._impl)
@@ -1164,7 +1107,7 @@
:type: ``bool``
""")
- def drained(self):
+ def drained(self) -> int:
"""
Drain excess credit for this link.
@@ -1183,20 +1126,17 @@
to the receiver.
:return: The number of credits drained.
- :rtype: ``int``
"""
return pn_link_drained(self._impl)
@property
- def remote_max_message_size(self):
+ def remote_max_message_size(self) -> int:
"""
Get the remote view of the maximum message size for this link.
.. warning:: **Unsettled API**
A zero value means the size is unlimited.
-
- :type: ``long``
"""
return pn_link_remote_max_message_size(self._impl)
@@ -1270,22 +1210,20 @@
A link over which messages are sent.
"""
- def offered(self, n):
+ def offered(self, n: int) -> None:
"""
Signal the availability of deliveries for this Sender.
:param n: Credit the number of deliveries potentially
available for transfer.
- :type n: ``int``
"""
pn_link_offered(self._impl, n)
- def stream(self, data):
+ def stream(self, data: bytes) -> int:
"""
Send specified data as part of the current delivery.
:param data: Data to send
- :type data: ``binary``
"""
return self._check(pn_link_send(self._impl, data))
@@ -1327,16 +1265,15 @@
A link over which messages are received.
"""
- def flow(self, n):
+ def flow(self, n: int) -> None:
"""
Increases the credit issued to the remote sender by the specified number of messages.
:param n: The credit to be issued to the remote sender.
- :type n: ``int``
"""
pn_link_flow(self._impl, n)
- def recv(self, limit):
+ def recv(self, limit: int) -> Optional[bytes]:
"""
Receive message data for the current delivery on this receiver.
@@ -1347,10 +1284,8 @@
``None`` is returned.
:param limit: the max data size to receive of this message
- :type limit: ``int``
:return: The received message data, or ``None`` if the message
has been completely received.
- :rtype: ``binary`` or ``None``
:raise: * :class:`Timeout` if timed out
* :class:`Interrupt` if interrupted
* :class:`LinkException` for all other exceptions
@@ -1362,7 +1297,7 @@
self._check(n)
return binary
- def drain(self, n):
+ def drain(self, n: int) -> None:
"""
Grant credit for incoming deliveries on this receiver, and
set drain mode to true.
@@ -1370,18 +1305,16 @@
Use :attr:`drain_mode` to set the drain mode explicitly.
:param n: The amount by which to increment the link credit
- :type n: ``int``
"""
pn_link_drain(self._impl, n)
- def draining(self):
+ def draining(self) -> bool:
"""
Check if a link is currently draining. A link is defined
to be draining when drain mode is set to ``True``, and the
sender still has excess credit.
:return: ``True`` if the link is currently draining, ``False`` otherwise.
- :rtype: ``bool``
"""
return pn_link_draining(self._impl)
@@ -1425,7 +1358,7 @@
def __init__(self, impl):
self._impl = impl
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, LinkException)
raise exc("[%s]" % err)
@@ -1565,12 +1498,11 @@
"""
return Data(pn_terminus_filter(self._impl))
- def copy(self, src):
+ def copy(self, src: 'Terminus') -> None:
"""
Copy another terminus object.
:param src: The terminus to be copied from
- :type src: :class:`Terminus`
:raises: :class:`LinkException` if there is an error
"""
self._check(pn_terminus_copy(self._impl, src._impl))
diff --git a/python/proton/_events.py b/python/proton/_events.py
index ddac28c..bee0bf7 100644
--- a/python/proton/_events.py
+++ b/python/proton/_events.py
@@ -17,8 +17,6 @@
# under the License.
#
-from __future__ import absolute_import
-
import threading
from cproton import PN_CONNECTION_BOUND, PN_CONNECTION_FINAL, PN_CONNECTION_INIT, PN_CONNECTION_LOCAL_CLOSE, \
@@ -35,40 +33,47 @@
from ._delivery import Delivery
from ._endpoints import Connection, Link, Session
from ._transport import Transport
+from typing import Any, List, Optional, Union, TYPE_CHECKING, Callable
+
+if TYPE_CHECKING:
+ from ._reactor import Container
+ from ._endpoints import Receiver, Sender
+ from ._handlers import ConnectSelectable
+ from ._selectable import Selectable
class Collector:
- def __init__(self):
+ def __init__(self) -> None:
self._impl = pn_collector()
def put(self, obj, etype):
pn_collector_put(self._impl, PN_PYREF, pn_py2void(obj), etype.number)
- def peek(self):
+ def peek(self) -> Optional['Event']:
return Event.wrap(pn_collector_peek(self._impl))
- def more(self):
+ def more(self) -> bool:
return pn_collector_more(self._impl)
- def pop(self):
+ def pop(self) -> None:
ev = self.peek()
pn_collector_pop(self._impl)
- def release(self):
+ def release(self) -> None:
pn_collector_release(self._impl)
- def __del__(self):
+ def __del__(self) -> None:
pn_collector_free(self._impl)
del self._impl
if "TypeExtender" not in globals():
class TypeExtender:
- def __init__(self, number):
+ def __init__(self, number: int) -> None:
self.number = number
- def next(self):
+ def next(self) -> int:
try:
return self.number
finally:
@@ -88,7 +93,7 @@
_extended = TypeExtender(10000)
TYPES = {}
- def __init__(self, name=None, number=None, method=None):
+ def __init__(self, name: Optional[str] = None, number: Optional[int] = None, method: Optional[str] = None) -> None:
if name is None and number is None:
raise TypeError("extended events require a name")
try:
@@ -127,7 +132,7 @@
class EventBase(object):
- def __init__(self, type):
+ def __init__(self, type: EventType) -> None:
self._type = type
@property
@@ -164,15 +169,15 @@
for h in handler.handlers:
self.dispatch(h, type)
- def __repr__(self):
+ def __repr__(self) -> str:
return "%s(%r)" % (self._type, self.context)
-def _core(number, method):
+def _core(number: int, method: str) -> EventType:
return EventType(number=number, method=method)
-def _internal(name):
+def _internal(name: str) -> EventType:
return EventType(name=name)
@@ -458,11 +463,9 @@
self._transport = Transport.wrap(pn_event_transport(impl))
@property
- def clazz(self):
+ def clazz(self) -> str:
"""
The name of the class associated with the event context.
-
- :type: ``str``
"""
return self._clsname
@@ -515,14 +518,14 @@
return h
@property
- def reactor(self):
+ def reactor(self) -> 'Container':
"""
**Deprecated** - The :class:`reactor.Container` (was reactor) associated with the event.
"""
return self.container
@property
- def container(self):
+ def container(self) -> 'Container':
"""
The :class:`reactor.Container` associated with the event.
"""
@@ -542,54 +545,44 @@
return getattr(c, name, None)
@property
- def transport(self):
+ def transport(self) -> Optional[Transport]:
"""
The transport associated with the event, or ``None`` if none
is associated with it.
-
- :type: :class:`Transport`
"""
return self._transport
@property
- def connection(self):
+ def connection(self) -> Optional[Connection]:
"""
The connection associated with the event, or ``None`` if none
is associated with it.
-
- :type: :class:`Connection`
"""
return self._connection
@property
- def session(self):
+ def session(self) -> Optional[Session]:
"""
The session associated with the event, or ``None`` if none
is associated with it.
-
- :type: :class:`Session`
"""
return self._session
@property
- def link(self):
+ def link(self) -> Optional[Union['Receiver', 'Sender']]:
"""
The link associated with the event, or ``None`` if none
is associated with it.
-
- :type: :class:`Link`
"""
return self._link
@property
- def sender(self):
+ def sender(self) -> Optional['Sender']:
"""
The sender link associated with the event, or ``None`` if
none is associated with it. This is essentially an alias for
link(), that does an additional check on the type of the
link.
-
- :type: :class:`Sender` (**<-- CHECK!**)
"""
l = self.link
if l and l.is_sender:
@@ -598,13 +591,11 @@
return None
@property
- def receiver(self):
+ def receiver(self) -> Optional['Receiver']:
"""
The receiver link associated with the event, or ``None`` if
none is associated with it. This is essentially an alias for
link(), that does an additional check on the type of the link.
-
- :type: :class:`Receiver` (**<-- CHECK!**)
"""
l = self.link
if l and l.is_receiver:
@@ -613,12 +604,10 @@
return None
@property
- def delivery(self):
+ def delivery(self) -> Optional[Delivery]:
"""
The delivery associated with the event, or ``None`` if none
is associated with it.
-
- :type: :class:`Delivery`
"""
return self._delivery
@@ -649,13 +638,12 @@
"""
self.handlers.append(handler)
- def on_unhandled(self, method, *args):
+ def on_unhandled(self, method: str, *args) -> None:
"""
The callback for handling events which are not handled by
any other handler.
:param method: The name of the intended handler method.
- :type method: ``str``
:param args: Arguments for the intended handler method.
"""
pass
diff --git a/python/proton/_handlers.py b/python/proton/_handlers.py
index 963a57a..0191b98 100644
--- a/python/proton/_handlers.py
+++ b/python/proton/_handlers.py
@@ -17,8 +17,6 @@
# under the License.
#
-from __future__ import absolute_import
-
import errno
import logging
import socket
@@ -35,6 +33,11 @@
from ._selectable import Selectable
from ._transport import Transport
from ._url import Url
+from typing import Any, List, Optional, Tuple, Union, TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+ from ._reactor import Container, Transaction
+ from ._endpoints import Sender, Receiver
log = logging.getLogger("proton")
@@ -54,13 +57,13 @@
self.auto_settle = auto_settle
self.delegate = delegate
- def on_link_flow(self, event):
+ def on_link_flow(self, event: Event):
if event.link.is_sender and event.link.credit \
and event.link.state & Endpoint.LOCAL_ACTIVE \
and event.link.state & Endpoint.REMOTE_ACTIVE:
self.on_sendable(event)
- def on_delivery(self, event):
+ def on_delivery(self, event: Event):
dlv = event.delivery
if dlv.link.is_sender and dlv.updated:
if dlv.remote_state == Delivery.ACCEPTED:
@@ -74,41 +77,38 @@
if self.auto_settle:
dlv.settle()
- def on_sendable(self, event):
+ def on_sendable(self, event: Event):
"""
Called when the sender link has credit and messages can
therefore be transferred.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_sendable', event)
- def on_accepted(self, event):
+ def on_accepted(self, event: Event):
"""
Called when the remote peer accepts an outgoing message.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_accepted', event)
- def on_rejected(self, event):
+ def on_rejected(self, event: Event):
"""
Called when the remote peer rejects an outgoing message.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_rejected', event)
- def on_released(self, event):
+ def on_released(self, event: Event):
"""
Called when the remote peer releases an outgoing message. Note
that this may be in response to either the ``RELEASE`` or ``MODIFIED``
@@ -116,12 +116,11 @@
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_released', event)
- def on_settled(self, event):
+ def on_settled(self, event: Event):
"""
Called when the remote peer has settled the outgoing
message. This is the point at which it should never be
@@ -129,13 +128,12 @@
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_settled', event)
-def recv_msg(delivery):
+def recv_msg(delivery: Delivery) -> Message:
msg = Message()
msg.decode(delivery.link.recv(delivery.pending))
delivery.link.advance()
@@ -161,7 +159,7 @@
A class containing methods for handling received messages.
"""
- def accept(self, delivery):
+ def accept(self, delivery: Delivery):
"""
Accepts a received message.
@@ -170,11 +168,10 @@
for transactional methods.
:param delivery: The message delivery tracking object
- :type delivery: :class:`proton.Delivery`
"""
self.settle(delivery, Delivery.ACCEPTED)
- def reject(self, delivery):
+ def reject(self, delivery: Delivery):
"""
Rejects a received message that is considered invalid or
unprocessable.
@@ -184,11 +181,10 @@
for transactional methods.
:param delivery: The message delivery tracking object
- :type delivery: :class:`proton.Delivery`
"""
self.settle(delivery, Delivery.REJECTED)
- def release(self, delivery, delivered=True):
+ def release(self, delivery: Delivery, delivered: bool = True):
"""
Releases a received message, making it available at the source
for any (other) interested receiver. The ``delivered``
@@ -200,13 +196,11 @@
for transactional methods.
:param delivery: The message delivery tracking object
- :type delivery: :class:`proton.Delivery`
:param delivered: If ``True``, the message will be annotated
with a delivery attempt (setting delivery flag
:const:`proton.Delivery.MODIFIED`). Otherwise, the message
will be returned without the annotation and released (setting
delivery flag :const:`proton.Delivery.RELEASED`
- :type delivered: ``bool``
"""
if delivered:
self.settle(delivery, Delivery.MODIFIED)
@@ -244,7 +238,7 @@
self.delegate = delegate
self.auto_accept = auto_accept
- def on_delivery(self, event):
+ def on_delivery(self, event: Event) -> None:
dlv = event.delivery
if not dlv.link.is_receiver:
return
@@ -272,7 +266,7 @@
elif dlv.updated and dlv.settled:
self.on_settled(event)
- def on_message(self, event):
+ def on_message(self, event: Event):
"""
Called when a message is received. The message itself can be
obtained as a property on the event. For the purpose of
@@ -282,29 +276,26 @@
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_message', event)
- def on_settled(self, event):
+ def on_settled(self, event: Event):
"""
Callback for when a message delivery is settled by the remote peer.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_settled', event)
- def on_aborted(self, event):
+ def on_aborted(self, event: Event):
"""
Callback for when a message delivery is aborted by the remote peer.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_aborted', event)
@@ -336,7 +327,7 @@
@classmethod
def is_local_open(cls, endpoint):
"""
- Test if local ``enpoint`` is open (ie has state
+ Test if local ``endpoint`` is open (ie has state
:const:`proton.Endpoint.LOCAL_ACTIVE`).
:param endpoint: The local endpoint to be tested.
@@ -350,7 +341,7 @@
@classmethod
def is_local_uninitialised(cls, endpoint):
"""
- Test if local ``enpoint`` is uninitialised (ie has state
+ Test if local ``endpoint`` is uninitialised (ie has state
:const:`proton.Endpoint.LOCAL_UNINIT`).
:param endpoint: The local endpoint to be tested.
@@ -364,7 +355,7 @@
@classmethod
def is_local_closed(cls, endpoint):
"""
- Test if local ``enpoint`` is closed (ie has state
+ Test if local ``endpoint`` is closed (ie has state
:const:`proton.Endpoint.LOCAL_CLOSED`).
:param endpoint: The local endpoint to be tested.
@@ -392,7 +383,7 @@
@classmethod
def is_remote_closed(cls, endpoint):
"""
- Test if remote ``enpoint`` is closed (ie has state
+ Test if remote ``endpoint`` is closed (ie has state
:const:`proton.Endpoint.REMOTE_CLOSED`).
:param endpoint: The remote endpoint to be tested.
@@ -404,22 +395,20 @@
return endpoint.state & Endpoint.REMOTE_CLOSED
@classmethod
- def print_error(cls, endpoint, endpoint_type):
+ def print_error(cls, endpoint: Endpoint, endpoint_type: str) -> None:
"""
Logs an error message related to an error condition at an endpoint.
:param endpoint: The endpoint to be tested
- :type endpoint: :class:`proton.Endpoint`
:param endpoint_type: The endpoint type as a string to be printed
in the log message.
- :type endpoint_type: ``str``
"""
if endpoint.remote_condition:
log.error(endpoint.remote_condition.description or endpoint.remote_condition.name)
elif cls.is_local_open(endpoint) and cls.is_remote_closed(endpoint):
log.error("%s closed by peer" % endpoint_type)
- def on_link_remote_close(self, event):
+ def on_link_remote_close(self, event: Event) -> None:
if event.link.remote_condition:
self.on_link_error(event)
elif self.is_local_closed(event.link):
@@ -428,7 +417,7 @@
self.on_link_closing(event)
event.link.close()
- def on_session_remote_close(self, event):
+ def on_session_remote_close(self, event: Event) -> None:
if event.session.remote_condition:
self.on_session_error(event)
elif self.is_local_closed(event.session):
@@ -437,7 +426,7 @@
self.on_session_closing(event)
event.session.close()
- def on_connection_remote_close(self, event):
+ def on_connection_remote_close(self, event: Event) -> None:
if event.connection.remote_condition:
if event.connection.remote_condition.name == "amqp:connection:forced":
# Treat this the same as just having the transport closed by the peer without
@@ -450,40 +439,40 @@
self.on_connection_closing(event)
event.connection.close()
- def on_connection_local_open(self, event):
+ def on_connection_local_open(self, event: Event) -> None:
if self.is_remote_open(event.connection):
self.on_connection_opened(event)
- def on_connection_remote_open(self, event):
+ def on_connection_remote_open(self, event: Event) -> None:
if self.is_local_open(event.connection):
self.on_connection_opened(event)
elif self.is_local_uninitialised(event.connection):
self.on_connection_opening(event)
event.connection.open()
- def on_session_local_open(self, event):
+ def on_session_local_open(self, event: Event) -> None:
if self.is_remote_open(event.session):
self.on_session_opened(event)
- def on_session_remote_open(self, event):
+ def on_session_remote_open(self, event: Event) -> None:
if self.is_local_open(event.session):
self.on_session_opened(event)
elif self.is_local_uninitialised(event.session):
self.on_session_opening(event)
event.session.open()
- def on_link_local_open(self, event):
+ def on_link_local_open(self, event: Event) -> None:
if self.is_remote_open(event.link):
self.on_link_opened(event)
- def on_link_remote_open(self, event):
+ def on_link_remote_open(self, event: Event) -> None:
if self.is_local_open(event.link):
self.on_link_opened(event)
elif self.is_local_uninitialised(event.link):
self.on_link_opening(event)
event.link.open()
- def on_connection_opened(self, event):
+ def on_connection_opened(self, event: Event) -> None:
"""
Callback for when both the local and remote endpoints of a
connection have opened.
@@ -495,86 +484,79 @@
if self.delegate is not None:
_dispatch(self.delegate, 'on_connection_opened', event)
- def on_session_opened(self, event):
+ def on_session_opened(self, event: Event) -> None:
"""
Callback for when both the local and remote endpoints of a
session have opened.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_session_opened', event)
- def on_link_opened(self, event):
+ def on_link_opened(self, event: Event) -> None:
"""
Callback for when both the local and remote endpoints of a
link have opened.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_link_opened', event)
- def on_connection_opening(self, event):
+ def on_connection_opening(self, event: Event) -> None:
"""
Callback for when a remote peer initiates the opening of
a connection.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_connection_opening', event)
- def on_session_opening(self, event):
+ def on_session_opening(self, event: Event) -> None:
"""
Callback for when a remote peer initiates the opening of
a session.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_session_opening', event)
- def on_link_opening(self, event):
+ def on_link_opening(self, event: Event) -> None:
"""
Callback for when a remote peer initiates the opening of
a link.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_link_opening', event)
- def on_connection_error(self, event):
+ def on_connection_error(self, event: Event) -> None:
"""
Callback for when an initiated connection open fails.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_connection_error', event)
else:
self.print_error(event.connection, "connection")
- def on_session_error(self, event):
+ def on_session_error(self, event: Event) -> None:
"""
Callback for when an initiated session open fails.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_session_error', event)
@@ -582,13 +564,12 @@
self.print_error(event.session, "session")
event.connection.close()
- def on_link_error(self, event):
+ def on_link_error(self, event: Event) -> None:
"""
Callback for when an initiated link open fails.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_link_error', event)
@@ -596,103 +577,95 @@
self.print_error(event.link, "link")
event.connection.close()
- def on_connection_closed(self, event):
+ def on_connection_closed(self, event: Event) -> None:
"""
Callback for when both the local and remote endpoints of a
connection have closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_connection_closed', event)
- def on_session_closed(self, event):
+ def on_session_closed(self, event: Event) -> None:
"""
Callback for when both the local and remote endpoints of a
session have closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_session_closed', event)
- def on_link_closed(self, event):
+ def on_link_closed(self, event: Event) -> None:
"""
Callback for when both the local and remote endpoints of a
link have closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_link_closed', event)
- def on_connection_closing(self, event):
+ def on_connection_closing(self, event: Event) -> None:
"""
Callback for when a remote peer initiates the closing of
a connection.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_connection_closing', event)
elif self.peer_close_is_error:
self.on_connection_error(event)
- def on_session_closing(self, event):
+ def on_session_closing(self, event: Event) -> None:
"""
Callback for when a remote peer initiates the closing of
a session.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_session_closing', event)
elif self.peer_close_is_error:
self.on_session_error(event)
- def on_link_closing(self, event):
+ def on_link_closing(self, event: Event) -> None:
"""
Callback for when a remote peer initiates the closing of
a link.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None:
_dispatch(self.delegate, 'on_link_closing', event)
elif self.peer_close_is_error:
self.on_link_error(event)
- def on_transport_tail_closed(self, event):
+ def on_transport_tail_closed(self, event: Event) -> None:
"""
Callback for when the transport tail has closed (ie no further input will
be accepted by the transport).
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
self.on_transport_closed(event)
- def on_transport_closed(self, event):
+ def on_transport_closed(self, event: Event) -> None:
"""
Callback for when the transport has closed - ie both the head (input) and
tail (output) of the transport pipeline are closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if self.delegate is not None and event.connection and self.is_local_open(event.connection):
_dispatch(self.delegate, 'on_disconnected', event)
@@ -705,20 +678,22 @@
cases.
:param prefetch: Initial flow credit for receiving messages, defaults to 10.
- :type prefetch: ``int``
:param auto_accept: If ``True``, accept all messages (default). Otherwise
messages must be individually accepted or rejected.
- :type auto_accept: ``bool``
:param auto_settle: If ``True``, settle all messages (default). Otherwise
messages must be explicitly settled.
- :type auto_settle: ``bool``
:param peer_close_is_error: If ``True``, a peer endpoint closing will be
treated as an error with an error callback. Otherwise (default), the
normal callbacks for the closing will occur.
- :type peer_close_is_error: ``bool``
"""
- def __init__(self, prefetch=10, auto_accept=True, auto_settle=True, peer_close_is_error=False):
+ def __init__(
+ self,
+ prefetch: int = 10,
+ auto_accept: bool = True,
+ auto_settle: bool = True,
+ peer_close_is_error: bool = False
+ ) -> None:
self.handlers = []
if prefetch:
self.handlers.append(FlowController(prefetch))
@@ -727,7 +702,7 @@
self.handlers.append(OutgoingMessageHandler(auto_settle, weakref.proxy(self)))
self.fatal_conditions = ["amqp:unauthorized-access"]
- def on_transport_error(self, event):
+ def on_transport_error(self, event: Event) -> None:
"""
Called when some error is encountered with the transport over
which the AMQP connection is to be established. This includes
@@ -735,7 +710,6 @@
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if event.transport.condition:
if event.transport.condition.info:
@@ -749,7 +723,7 @@
else:
logging.error("Unspecified transport error")
- def on_connection_error(self, event):
+ def on_connection_error(self, event: Event) -> None:
"""
Called when the peer closes the connection with an error condition.
@@ -759,152 +733,138 @@
"""
EndpointStateHandler.print_error(event.connection, "connection")
- def on_session_error(self, event):
+ def on_session_error(self, event: Event) -> None:
"""
Called when the peer closes the session with an error condition.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
EndpointStateHandler.print_error(event.session, "session")
event.connection.close()
- def on_link_error(self, event):
+ def on_link_error(self, event: Event) -> None:
"""
Called when the peer closes the link with an error condition.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
EndpointStateHandler.print_error(event.link, "link")
event.connection.close()
- def on_reactor_init(self, event):
+ def on_reactor_init(self, event: Event) -> None:
"""
Called when the event loop - the reactor - starts.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
if hasattr(event.reactor, 'subclass'):
setattr(event, event.reactor.subclass.__name__.lower(), event.reactor)
self.on_start(event)
- def on_start(self, event):
+ def on_start(self, event: Event) -> None:
"""
Called when the event loop starts. (Just an alias for on_reactor_init)
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_connection_closed(self, event):
+ def on_connection_closed(self, event: Event) -> None:
"""
Called when the connection is closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_session_closed(self, event):
+ def on_session_closed(self, event: Event) -> None:
"""
Called when the session is closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_link_closed(self, event):
+ def on_link_closed(self, event: Event) -> None:
"""
Called when the link is closed.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_connection_closing(self, event):
+ def on_connection_closing(self, event: Event) -> None:
"""
Called when the peer initiates the closing of the connection.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_session_closing(self, event):
+ def on_session_closing(self, event: Event) -> None:
"""
Called when the peer initiates the closing of the session.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_link_closing(self, event):
+ def on_link_closing(self, event: Event) -> None:
"""
Called when the peer initiates the closing of the link.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_disconnected(self, event):
+ def on_disconnected(self, event: Event) -> None:
"""
Called when the socket is disconnected.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_sendable(self, event):
+ def on_sendable(self, event: Event) -> None:
"""
Called when the sender link has credit and messages can
therefore be transferred.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_accepted(self, event):
+ def on_accepted(self, event: Event) -> None:
"""
Called when the remote peer accepts an outgoing message.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_rejected(self, event):
+ def on_rejected(self, event: Event) -> None:
"""
Called when the remote peer rejects an outgoing message.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_released(self, event):
+ def on_released(self, event: Event) -> None:
"""
Called when the remote peer releases an outgoing message. Note
that this may be in response to either the RELEASE or MODIFIED
@@ -912,11 +872,10 @@
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_settled(self, event):
+ def on_settled(self, event: Event) -> None:
"""
Called when the remote peer has settled the outgoing
message. This is the point at which it should never be
@@ -924,11 +883,10 @@
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_message(self, event):
+ def on_message(self, event: Event) -> None:
"""
Called when a message is received. The message itself can be
obtained as a property on the event. For the purpose of
@@ -939,7 +897,6 @@
:param event: The underlying event object. Use this to obtain further
information on the event. In particular, the message itself may
be obtained by accessing ``event.message``.
- :type event: :class:`proton.Event`
"""
pass
@@ -950,56 +907,51 @@
be notified of state changes related to a transaction.
"""
- def on_transaction_declared(self, event):
+ def on_transaction_declared(self, event: Event) -> None:
"""
Called when a local transaction is declared.
:param event: The underlying event object. Use this to obtain further
information on the event. In particular, the :class:`proton.reactor.Transaction`
object may be obtained by accessing ``event.transaction``.
- :type event: :class:`proton.Event`
"""
pass
- def on_transaction_committed(self, event):
+ def on_transaction_committed(self, event: Event) -> None:
"""
Called when a local transaction is discharged successfully
(committed).
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_transaction_aborted(self, event):
+ def on_transaction_aborted(self, event: Event) -> None:
"""
Called when a local transaction is discharged unsuccessfully
(aborted).
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_transaction_declare_failed(self, event):
+ def on_transaction_declare_failed(self, event: Event) -> None:
"""
Called when a local transaction declare fails.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
- def on_transaction_commit_failed(self, event):
+ def on_transaction_commit_failed(self, event: Event) -> None:
"""
Called when the commit of a local transaction fails.
:param event: The underlying event object. Use this to obtain further
information on the event.
- :type event: :class:`proton.Event`
"""
pass
@@ -1013,34 +965,34 @@
a transactional acceptance of received messages.
:param prefetch: Initial flow credit for receiving messages, defaults to 10.
- :type prefetch: ``int``
:param auto_accept: If ``True``, accept all messages (default). Otherwise messages
must be individually accepted or rejected.
- :type auto_accept: ``bool``
:param auto_settle: If ``True``, settle all messages (default). Otherwise
messages must be explicitly settled.
- :type auto_settle: ``bool``
:param peer_close_is_error: If ``True``, a peer endpoint closing will be
treated as an error with an error callback. Otherwise (default), the
normal callbacks for the closing will occur.
- :type peer_close_is_error: ``bool``
"""
- def __init__(self, prefetch=10, auto_accept=False, auto_settle=True, peer_close_is_error=False):
+ def __init__(
+ self,
+ prefetch: int = 10,
+ auto_accept: bool = False,
+ auto_settle: bool = True,
+ peer_close_is_error: bool = False
+ ) -> None:
super(TransactionalClientHandler, self).__init__(prefetch, auto_accept, auto_settle, peer_close_is_error)
- def accept(self, delivery, transaction=None):
+ def accept(self, delivery: Delivery, transaction: Optional['Transaction'] = None):
"""
A convenience method for accepting a received message as part of a
transaction. If no transaction object is supplied, a regular
non-transactional acceptance will be performed.
:param delivery: Delivery tracking object for received message.
- :type delivery: :class:`proton.Delivery`
:param transaction: Transaction tracking object which is required if
the message is being accepted under the transaction. If ``None`` (default),
then a normal non-transactional accept occurs.
- :type transaction: :class:`proton.reactor.Transaction`
"""
if transaction:
transaction.accept(delivery)
@@ -1049,23 +1001,23 @@
class FlowController(Handler):
- def __init__(self, window=1024):
+ def __init__(self, window: int = 1024) -> None:
self._window = window
self._drained = 0
- def on_link_local_open(self, event):
+ def on_link_local_open(self, event: Event) -> None:
self._flow(event.link)
- def on_link_remote_open(self, event):
+ def on_link_remote_open(self, event: Event) -> None:
self._flow(event.link)
- def on_link_flow(self, event):
+ def on_link_flow(self, event: Event) -> None:
self._flow(event.link)
- def on_delivery(self, event):
+ def on_delivery(self, event: Event) -> None:
self._flow(event.link)
- def _flow(self, link):
+ def _flow(self, link: Union['Sender', 'Receiver']) -> None:
if link.is_receiver:
self._drained += link.drained()
if self._drained == 0:
@@ -1076,19 +1028,19 @@
class Handshaker(Handler):
@staticmethod
- def on_connection_remote_open(event):
+ def on_connection_remote_open(event: Event) -> None:
conn = event.connection
if conn.state & Endpoint.LOCAL_UNINIT:
conn.open()
@staticmethod
- def on_session_remote_open(event):
+ def on_session_remote_open(event: Event) -> None:
ssn = event.session
if ssn.state & Endpoint.LOCAL_UNINIT:
ssn.open()
@staticmethod
- def on_link_remote_open(event):
+ def on_link_remote_open(event: Event) -> None:
link = event.link
if link.state & Endpoint.LOCAL_UNINIT:
link.source.copy(link.remote_source)
@@ -1096,19 +1048,19 @@
link.open()
@staticmethod
- def on_connection_remote_close(event):
+ def on_connection_remote_close(event: Event) -> None:
conn = event.connection
if not conn.state & Endpoint.LOCAL_CLOSED:
conn.close()
@staticmethod
- def on_session_remote_close(event):
+ def on_session_remote_close(event: Event) -> None:
ssn = event.session
if not ssn.state & Endpoint.LOCAL_CLOSED:
ssn.close()
@staticmethod
- def on_link_remote_close(event):
+ def on_link_remote_close(event: Event) -> None:
link = event.link
if not link.state & Endpoint.LOCAL_CLOSED:
link.close()
@@ -1121,26 +1073,26 @@
class PythonIO:
- def __init__(self):
+ def __init__(self) -> None:
self.selectables = []
self.delegate = IOHandler()
- def on_unhandled(self, method, event):
+ def on_unhandled(self, method: str, event: Event) -> None:
event.dispatch(self.delegate)
- def on_selectable_init(self, event):
+ def on_selectable_init(self, event: Event) -> None:
self.selectables.append(event.context)
- def on_selectable_updated(self, event):
+ def on_selectable_updated(self, event: Event) -> None:
pass
- def on_selectable_final(self, event):
+ def on_selectable_final(self, event: Event) -> None:
sel = event.context
if sel.is_terminal:
self.selectables.remove(sel)
sel.close()
- def on_reactor_quiesced(self, event):
+ def on_reactor_quiesced(self, event: Event) -> None:
reactor = event.reactor
# check if we are still quiesced, other handlers of
# on_reactor_quiesced could have produced events to process
@@ -1186,25 +1138,25 @@
# For C style IO handler need to implement Selector
class IOHandler(Handler):
- def __init__(self):
+ def __init__(self) -> None:
self._selector = IO.Selector()
- def on_selectable_init(self, event):
+ def on_selectable_init(self, event: Event) -> None:
s = event.selectable
self._selector.add(s)
s._reactor._selectables += 1
- def on_selectable_updated(self, event):
+ def on_selectable_updated(self, event: Event) -> None:
s = event.selectable
self._selector.update(s)
- def on_selectable_final(self, event):
+ def on_selectable_final(self, event: Event) -> None:
s = event.selectable
self._selector.remove(s)
s._reactor._selectables -= 1
s.close()
- def on_reactor_quiesced(self, event):
+ def on_reactor_quiesced(self, event: Event) -> None:
r = event.reactor
if not r.quiesced:
@@ -1224,7 +1176,7 @@
r.yield_()
- def on_selectable_readable(self, event):
+ def on_selectable_readable(self, event: Event) -> None:
s = event.selectable
t = s._transport
@@ -1252,7 +1204,7 @@
r = s._reactor
self.update(t, s, r.now)
- def on_selectable_writable(self, event):
+ def on_selectable_writable(self, event: Event) -> None:
s = event.selectable
t = s._transport
@@ -1277,7 +1229,7 @@
r = s._reactor
self.update(t, s, r.now)
- def on_selectable_error(self, event):
+ def on_selectable_error(self, event: Event) -> None:
s = event.selectable
t = s._transport
@@ -1288,14 +1240,14 @@
t._selectable = None
s.update()
- def on_selectable_expired(self, event):
+ def on_selectable_expired(self, event: Event) -> None:
s = event.selectable
t = s._transport
r = s._reactor
self.update(t, s, r.now)
- def on_connection_local_open(self, event):
+ def on_connection_local_open(self, event: Event) -> None:
c = event.connection
if not c.state & Endpoint.REMOTE_UNINIT:
return
@@ -1306,7 +1258,7 @@
# bound the transport and connection!
t.bind_nothrow(c)
- def on_connection_bound(self, event):
+ def on_connection_bound(self, event: Event) -> None:
c = event.connection
t = event.transport
@@ -1347,7 +1299,7 @@
t._selectable = None
@staticmethod
- def update(transport, selectable, now):
+ def update(transport: Transport, selectable: Selectable, now: float) -> None:
try:
capacity = transport.capacity()
selectable.reading = capacity > 0
@@ -1367,14 +1319,14 @@
selectable.deadline = transport.tick(now)
selectable.update()
- def on_transport(self, event):
+ def on_transport(self, event: Event) -> None:
t = event.transport
r = t._reactor
s = t._selectable
if s and not s.is_terminal:
self.update(t, s, r.now)
- def on_transport_closed(self, event):
+ def on_transport_closed(self, event: Event) -> None:
t = event.transport
r = t._reactor
s = t._selectable
@@ -1395,10 +1347,10 @@
self._iohandler = iohandler
transport._connect_selectable = self
- def readable(self):
+ def readable(self) -> None:
pass
- def writable(self):
+ def writable(self) -> None:
e = self._delegate.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
t = self._transport
t._connect_selectable = None
diff --git a/python/proton/_io.py b/python/proton/_io.py
index 400b575..2758587 100644
--- a/python/proton/_io.py
+++ b/python/proton/_io.py
@@ -24,6 +24,10 @@
import select
import time
+from typing import TYPE_CHECKING, Tuple, List
+
+if TYPE_CHECKING:
+ from proton._selectable import Selectable
PN_INVALID_SOCKET = -1
@@ -31,12 +35,12 @@
class IO(object):
@staticmethod
- def _setupsocket(s):
+ def _setupsocket(s: socket) -> None:
s.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, True)
s.setblocking(False)
@staticmethod
- def close(s):
+ def close(s: socket) -> None:
s.close()
@staticmethod
@@ -49,7 +53,7 @@
return s
@staticmethod
- def accept(s):
+ def accept(s: socket):
n = s.accept()
IO._setupsocket(n[0])
return n
@@ -70,19 +74,19 @@
return select.select(*args, **kwargs)
@staticmethod
- def sleep(t):
+ def sleep(t: float) -> None:
time.sleep(t)
return
class Selector(object):
- def __init__(self):
+ def __init__(self) -> None:
self._selectables = set()
self._reading = set()
self._writing = set()
self._deadline = None
- def add(self, selectable):
+ def add(self, selectable: 'Selectable') -> None:
self._selectables.add(selectable)
if selectable.reading:
self._reading.add(selectable)
@@ -94,17 +98,17 @@
else:
self._deadline = min(selectable.deadline, self._deadline)
- def remove(self, selectable):
+ def remove(self, selectable: 'Selectable') -> None:
self._selectables.discard(selectable)
self._reading.discard(selectable)
self._writing.discard(selectable)
self.update_deadline()
@property
- def selectables(self):
+ def selectables(self) -> int:
return len(self._selectables)
- def update_deadline(self):
+ def update_deadline(self) -> None:
for sel in self._selectables:
if sel.deadline:
if self._deadline is None:
@@ -112,7 +116,7 @@
else:
self._deadline = min(sel.deadline, self._deadline)
- def update(self, selectable):
+ def update(self, selectable: 'Selectable') -> None:
self._reading.discard(selectable)
self._writing.discard(selectable)
if selectable.reading:
diff --git a/python/proton/_message.py b/python/proton/_message.py
index f4a73ff..3d12a04 100644
--- a/python/proton/_message.py
+++ b/python/proton/_message.py
@@ -37,6 +37,13 @@
from ._data import char, Data, symbol, ulong, AnnotationDict
from ._endpoints import Link
from ._exceptions import EXCEPTIONS, MessageException
+from typing import Dict, Optional, Union, TYPE_CHECKING, overload
+
+if TYPE_CHECKING:
+ from proton._delivery import Delivery
+ from proton._endpoints import Sender, Receiver
+ from uuid import UUID
+ from proton._data import Described
class Message(object):
@@ -69,19 +76,19 @@
getattr(self, k) # Raise exception if it's not a valid attribute.
setattr(self, k, v)
- def __del__(self):
+ def __del__(self) -> None:
if hasattr(self, "_msg"):
pn_message_free(self._msg)
del self._msg
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, MessageException)
raise exc("[%s]: %s" % (err, pn_error_text(pn_message_error(self._msg))))
else:
return err
- def _check_property_keys(self):
+ def _check_property_keys(self) -> None:
"""
AMQP allows only string keys for properties. This function checks that this requirement is met
and raises a MessageException if not. However, in certain cases, conversions to string are
@@ -110,7 +117,7 @@
for old_key, new_key in changed_keys:
self.properties[new_key] = self.properties.pop(old_key)
- def _pre_encode(self):
+ def _pre_encode(self) -> None:
inst = Data(pn_message_instructions(self._msg))
ann = Data(pn_message_annotations(self._msg))
props = Data(pn_message_properties(self._msg))
@@ -130,7 +137,7 @@
if self.body is not None:
body.put_object(self.body)
- def _post_decode(self):
+ def _post_decode(self) -> None:
inst = Data(pn_message_instructions(self._msg))
ann = Data(pn_message_annotations(self._msg))
props = Data(pn_message_properties(self._msg))
@@ -153,7 +160,7 @@
else:
self.body = None
- def clear(self):
+ def clear(self) -> None:
"""
Clears the contents of the :class:`Message`. All fields will be reset to
their default values.
@@ -495,7 +502,7 @@
self._check(err)
return data
- def decode(self, data):
+ def decode(self, data: bytes) -> None:
self._check(pn_message_decode(self._msg, data))
self._post_decode()
@@ -548,7 +555,7 @@
self.decode(dlv.encoded)
return dlv
- def __repr__(self):
+ def __repr__(self) -> str:
props = []
for attr in ("inferred", "address", "reply_to", "durable", "ttl",
"priority", "first_acquirer", "delivery_count", "id",
diff --git a/python/proton/_reactor.py b/python/proton/_reactor.py
index 28d02c9..8e69d4a 100644
--- a/python/proton/_reactor.py
+++ b/python/proton/_reactor.py
@@ -17,14 +17,21 @@
# under the License.
#
-from __future__ import absolute_import
-
import heapq
import json
import logging
import re
import os
import queue
+from typing import Any, Dict, Iterator, Optional, List, Union, Callable
+try:
+ from typing import Literal
+except ImportError:
+ class Literal:
+ @classmethod
+ def __class_getitem__(cls, item):
+ pass
+
import time
import traceback
import uuid
@@ -46,55 +53,62 @@
from ._handlers import OutgoingMessageHandler, IOHandler
from ._io import IO
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+ from ._endpoints import Receiver, Sender
+ from ._handlers import ConnectSelectable, TransactionHandler
+ from ._utils import BlockingConnection
+ from socket import socket
+ from uuid import UUID
_logger = logging.getLogger("proton")
-def _generate_uuid():
+def _generate_uuid() -> 'UUID':
return uuid.uuid4()
-def _now():
+def _now() -> float:
return time.time()
@total_ordering
class Task(object):
- def __init__(self, reactor, deadline, handler):
+ def __init__(self, reactor: 'Container', deadline: float, handler: Handler) -> None:
self._deadline = deadline
self._handler = handler
self._reactor = reactor
self._cancelled = False
- def __lt__(self, rhs):
+ def __lt__(self, rhs: 'Task') -> bool:
return self._deadline < rhs._deadline
- def cancel(self):
+ def cancel(self) -> None:
self._cancelled = True
@property
- def handler(self):
+ def handler(self) -> Handler:
return self._handler
@property
- def container(self):
+ def container(self) -> 'Container':
return self._reactor
class TimerSelectable(Selectable):
- def __init__(self, reactor):
+ def __init__(self, reactor: 'Container') -> None:
super(TimerSelectable, self).__init__(None, reactor)
- def readable(self):
+ def readable(self) -> None:
pass
- def writable(self):
+ def writable(self) -> None:
pass
- def expired(self):
+ def expired(self) -> None:
self._reactor.timer_tick()
self.deadline = self._reactor.timer_deadline
self.update()
@@ -102,7 +116,7 @@
class Reactor(object):
- def __init__(self, *handlers, **kwargs):
+ def __init__(self, *handlers, **kwargs) -> None:
self._previous = PN_EVENT_NONE
self._timeout = 0
self.mark()
@@ -153,13 +167,13 @@
def yield_(self):
self._yield = True
- def mark(self):
+ def mark(self) -> float:
""" This sets the reactor now instant to the current time """
self._now = _now()
return self._now
@property
- def now(self):
+ def now(self) -> float:
return self._now
def _get_handler(self):
@@ -186,12 +200,12 @@
self._handler = None
# Cross thread reactor wakeup
- def wakeup(self):
+ def wakeup(self) -> None:
# TODO: Do this with pipe and write?
# os.write(self._wakeup[1], "x", 1);
pass
- def start(self):
+ def start(self) -> None:
self.push_event(self, Event.REACTOR_INIT)
self._selectable = TimerSelectable(self)
self._selectable.deadline = self.timer_deadline
@@ -201,7 +215,7 @@
self.update(self._selectable)
@property
- def quiesced(self):
+ def quiesced(self) -> bool:
event = self._collector.peek()
if not event:
return True
@@ -209,7 +223,7 @@
return False
return event.type is Event.REACTOR_QUIESCED
- def _check_errors(self):
+ def _check_errors(self) -> None:
""" This """
if self.errors:
for exc, value, tb in self.errors[:-1]:
@@ -222,7 +236,7 @@
else:
raise value.with_traceback(tb)
- def process(self):
+ def process(self) -> bool:
# result = pn_reactor_process(self._impl)
# self._check_errors()
# return result
@@ -262,11 +276,11 @@
_logger.debug('%s Stopping', self)
return False
- def stop(self):
+ def stop(self) -> None:
self._stop = True
self._check_errors()
- def stop_events(self):
+ def stop_events(self) -> None:
self._collector.release()
def schedule(self, delay, handler):
@@ -287,7 +301,7 @@
self.update(self._selectable)
return task
- def timer_tick(self):
+ def timer_tick(self) -> None:
while self._timers > 0:
t = self._timerheap[0]
if t._cancelled:
@@ -301,7 +315,7 @@
self.push_event(t, Event.TIMER_TASK)
@property
- def timer_deadline(self):
+ def timer_deadline(self) -> Optional[float]:
while self._timers > 0:
t = self._timerheap[0]
if t._cancelled:
@@ -387,7 +401,7 @@
needed, to allow the event loop to end if needed.
"""
- def __init__(self):
+ def __init__(self) -> None:
self.queue = queue.Queue()
self.pipe = os.pipe()
self._transport = None
@@ -404,7 +418,7 @@
self.queue.put(event)
os.write(self.pipe[1], b"!")
- def close(self):
+ def close(self) -> None:
"""
Request that this EventInjector be closed. Existing events
will be dispatched on the container's event dispatch thread,
@@ -413,16 +427,16 @@
self._closed = True
os.write(self.pipe[1], b"!")
- def fileno(self):
+ def fileno(self) -> int:
return self.pipe[0]
- def on_selectable_init(self, event):
+ def on_selectable_init(self, event: Event) -> None:
sel = event.context
# sel.fileno(self.fileno())
sel.reading = True
sel.update()
- def on_selectable_readable(self, event):
+ def on_selectable_readable(self, event: Event) -> None:
s = event.context
os.read(self.pipe[0], 512)
while not self.queue.empty():
@@ -440,22 +454,24 @@
extended event types - see :class:`proton.EventType` for details.
:param typename: Event type name
- :type typename: ``str``
:param connection: Associates this event with a connection.
- :type connection: :class:`proton.Connection`
:param session: Associates this event with a session.
- :type session: :class:`proton.Session`
:param link: Associate this event with a link.
- :type link: :class:`proton.Link` or one of its subclasses
:param delivery: Associate this event with a delivery.
- :type delivery: :class:`proton.Delivery`
:param subject: Associate this event with an arbitrary object
- :type subject: any
"""
TYPES = {}
- def __init__(self, typename, connection=None, session=None, link=None, delivery=None, subject=None):
+ def __init__(
+ self,
+ typename: str,
+ connection: Optional[Connection] = None,
+ session: Optional[Session] = None,
+ link: Optional[Link] = None,
+ delivery: Optional[Delivery] = None,
+ subject: Any = None
+ ) -> None:
if isinstance(typename, EventType):
eventtype = typename
else:
@@ -485,7 +501,7 @@
"""
return self
- def __repr__(self):
+ def __repr__(self) -> str:
objects = [self.connection, self.session, self.link, self.delivery, self.subject]
return "%s(%s)" % (self.type, ", ".join([str(o) for o in objects if o is not None]))
@@ -518,13 +534,13 @@
self.settle_before_discharge = settle_before_discharge
self.declare()
- def commit(self):
+ def commit(self) -> None:
"""
Commit this transaction. Closes the transaction as a success.
"""
self.discharge(False)
- def abort(self):
+ def abort(self) -> None:
"""
Abort or roll back this transaction. Closes the transaction as a failure,
and reverses, or rolls back all actions (sent and received messages)
@@ -532,7 +548,7 @@
"""
self.discharge(True)
- def declare(self):
+ def declare(self) -> None:
self._declare = self._send_ctrl(symbol(u'amqp:declare:list'), [None])
def discharge(self, failed):
@@ -562,12 +578,11 @@
dlv.update(0x34)
return dlv
- def accept(self, delivery):
+ def accept(self, delivery: Delivery) -> None:
"""
Accept a received message under this transaction.
:param delivery: Delivery object for the received message.
- :type delivery: :class:`proton.Delivery`
"""
self.update(delivery, PN_ACCEPTED)
if self.settle_before_discharge:
@@ -618,14 +633,14 @@
Abstract interface for link configuration options
"""
- def apply(self, link):
+ def apply(self, link: Link) -> None:
"""
Subclasses will implement any configuration logic in this
method
"""
pass
- def test(self, link):
+ def test(self, link: Link) -> bool:
"""
Subclasses can override this to selectively apply an option
e.g. based on some link criteria
@@ -640,7 +655,7 @@
(ie pre-settled).
"""
- def apply(self, link):
+ def apply(self, link: Link) -> None:
"""
Set the at-most-once delivery semantics on the link.
@@ -658,7 +673,7 @@
forces the receiver to settle all messages once they are successfully received.
"""
- def apply(self, link):
+ def apply(self, link: Link) -> None:
"""
Set the at-least-once delivery semantics on the link.
@@ -674,16 +689,15 @@
Abstract class for sender options.
"""
- def apply(self, sender):
+ def apply(self, sender: 'Sender') -> None:
"""
Set the option on the sender.
:param sender: The sender on which this option is to be applied.
- :type sender: :class:`proton.Sender`
"""
pass
- def test(self, link):
+ def test(self, link: Link) -> bool:
return link.is_sender
@@ -692,16 +706,15 @@
Abstract class for receiver options
"""
- def apply(self, receiver):
+ def apply(self, receiver: 'Receiver') -> None:
"""
Set the option on the receiver.
:param receiver: The receiver on which this option is to be applied.
- :type receiver: :class:`proton.Receiver`
"""
pass
- def test(self, link):
+ def test(self, link: Link) -> bool:
return link.is_receiver
@@ -723,12 +736,11 @@
else:
self.properties[symbol(k)] = props[k]
- def apply(self, link):
+ def apply(self, link: Link) -> None:
"""
Set the map of properties on the specified link.
:param link: The link on which this property map is to be set.
- :type link: :class:`proton.Link`
"""
if link.is_receiver:
link.source.properties.put_dict(self.properties)
@@ -748,12 +760,11 @@
def __init__(self, filter_set={}):
self.filter_set = filter_set
- def apply(self, receiver):
+ def apply(self, receiver: 'Receiver') -> None:
"""
Set the filter on the specified receiver.
:param receiver: The receiver on which this filter is to be applied.
- :type receiver: :class:`proton.Receiver`
"""
receiver.source.filter.put_dict(self.filter_set)
@@ -781,12 +792,11 @@
:const:`proton.Terminus.EXPIRE_NEVER`.
"""
- def apply(self, receiver):
+ def apply(self, receiver: 'Receiver'):
"""
Set durability on the specified receiver.
:param receiver: The receiver on which durability is to be set.
- :type receiver: :class:`proton.Receiver`
"""
receiver.source.durability = Terminus.DELIVERIES
receiver.source.expiry_policy = Terminus.EXPIRE_NEVER
@@ -800,12 +810,11 @@
mode to :const:`proton.Terminus.DIST_MODE_MOVE`.
"""
- def apply(self, receiver):
+ def apply(self, receiver: 'Receiver'):
"""
Set message move semantics on the specified receiver.
:param receiver: The receiver on which message move semantics is to be set.
- :type receiver: :class:`proton.Receiver`
"""
receiver.source.distribution_mode = Terminus.DIST_MODE_MOVE
@@ -818,12 +827,11 @@
:const:`proton.Terminus.DIST_MODE_COPY`.
"""
- def apply(self, receiver):
+ def apply(self, receiver: 'Receiver'):
"""
Set message copy semantics on the specified receiver.
:param receiver: The receiver on which message copy semantics is to be set.
- :type receiver: :class:`proton.Receiver`
"""
receiver.source.distribution_mode = Terminus.DIST_MODE_COPY
@@ -839,13 +847,13 @@
options.apply(link)
-def _create_session(connection, handler=None):
+def _create_session(connection: Connection, handler: Optional[Handler] = None) -> Session:
session = connection.session()
session.open()
return session
-def _get_attr(target, name):
+def _get_attr(target: Any, name: str) -> Optional[Any]:
if hasattr(target, name):
return getattr(target, name)
else:
@@ -853,10 +861,10 @@
class SessionPerConnection(object):
- def __init__(self):
+ def __init__(self) -> None:
self._default_session = None
- def session(self, connection):
+ def session(self, connection: Connection) -> Session:
if not self._default_session:
self._default_session = _create_session(connection)
return self._default_session
@@ -868,21 +876,21 @@
opened connection.
"""
- def __init__(self, base):
+ def __init__(self, base: IOHandler) -> None:
self.base = base
- def on_unhandled(self, name, event):
+ def on_unhandled(self, name: str, event: Event) -> None:
if not self._override(event):
event.dispatch(self.base)
- def _override(self, event):
+ def _override(self, event: Event) -> Optional[bool]:
conn = event.connection
return conn and hasattr(conn, '_overrides') and event.dispatch(conn._overrides)
class Acceptor(Handler):
- def __init__(self, reactor, host, port, handler=None):
+ def __init__(self, reactor: 'Container', host: str, port: int, handler: Optional[Handler] = None) -> None:
self._ssl_domain = None
self._reactor = reactor
self._handler = handler
@@ -893,15 +901,15 @@
self._selectable = s
reactor.update(s)
- def set_ssl_domain(self, ssl_domain):
+ def set_ssl_domain(self, ssl_domain: SSLDomain) -> None:
self._ssl_domain = ssl_domain
- def close(self):
+ def close(self) -> None:
if not self._selectable.is_terminal:
self._selectable.terminate()
self._selectable.update()
- def on_selectable_readable(self, event):
+ def on_selectable_readable(self, event: Event) -> None:
s = event.selectable
sock, name = IO.accept(self._selectable)
@@ -923,7 +931,12 @@
IOHandler.update(t, s, r.now)
-def delay_iter(initial=0.1, factor=2.0, max_delay=10.0, max_tries=None):
+def delay_iter(
+ initial: float = 0.1,
+ factor: float = 2.0,
+ max_delay: float = 10.0,
+ max_tries: Optional[int] = None
+) -> Iterator[float]:
"""
iterator yielding the next delay in the sequence of delays. The first
delay is 0 seconds, the second 0.1 seconds, and each subsequent
@@ -947,11 +960,11 @@
with an initial value of 0 seconds.
"""
- def __init__(self, **kwargs):
+ def __init__(self, **kwargs) -> None:
self.kwargs = kwargs
self.iter = delay_iter(**self.kwargs)
- def __iter__(self):
+ def __iter__(self) -> Iterator[float]:
return self.iter
@@ -984,7 +997,7 @@
def __init__(self, values):
self.values = [Url(v) for v in values]
- def __iter__(self):
+ def __iter__(self) -> Iterator[Url]:
return iter(self.values)
@@ -994,7 +1007,7 @@
opened connection.
"""
- def __init__(self, connection):
+ def __init__(self, connection: Connection) -> None:
self.connection = connection
self.address = None
self.heartbeat = None
@@ -1011,7 +1024,7 @@
self._connect_sequence = None
self._next_url = None
- def _connect(self, connection, url):
+ def _connect(self, connection: Connection, url: Url) -> None:
connection.url = url
# if virtual-host not set, use host from address as default
if self.virtual_host is None:
@@ -1043,7 +1056,7 @@
if self.max_frame_size:
transport.max_frame_size = self.max_frame_size
- def on_connection_local_open(self, event):
+ def on_connection_local_open(self, event: Event) -> None:
if self.reconnect is None:
self._connect_sequence = ((delay, url) for delay in delay_iter() for url in self.address)
elif self.reconnect is False:
@@ -1053,7 +1066,7 @@
_, url = next(self._connect_sequence) # Ignore delay as we assume first delay must be 0
self._connect(event.connection, url)
- def on_connection_remote_open(self, event):
+ def on_connection_remote_open(self, event: Event) -> None:
_logger.info("Connected to %s" % event.connection.hostname)
if self.reconnect is None:
self._connect_sequence = ((delay, url) for delay in delay_iter() for url in self.address)
@@ -1062,7 +1075,7 @@
else:
self._connect_sequence = None # Help take out the garbage
- def on_transport_closed(self, event):
+ def on_transport_closed(self, event: Event) -> None:
if self.connection is None:
return
@@ -1090,14 +1103,14 @@
# See connector.cpp: conn.free()/pn_connection_release() here?
self.connection = None
- def on_timer_task(self, event):
+ def on_timer_task(self, event: Event) -> None:
if self._next_url:
self._connect(self.connection, self._next_url)
self._next_url = None
class SSLConfig(object):
- def __init__(self):
+ def __init__(self) -> None:
self.client = SSLDomain(SSLDomain.MODE_CLIENT)
self.server = SSLDomain(SSLDomain.MODE_SERVER)
@@ -1110,7 +1123,7 @@
self.server.set_trusted_ca_db(certificate_db)
-def _find_config_file():
+def _find_config_file() -> Optional[str]:
confname = 'connect.json'
confpath = ['.', os.path.expanduser('~/.config/messaging'), '/etc/messaging']
for d in confpath:
@@ -1131,7 +1144,7 @@
return {}
-def _strip_json_comments(json_text):
+def _strip_json_comments(json_text: str) -> str:
"""This strips c-style comments from text, taking into account '/*comments*/' and '//comments'
nested inside a string etc."""
def replacer(match):
@@ -1144,7 +1157,7 @@
return re.sub(pattern, replacer, json_text)
-def _get_default_port_for_scheme(scheme):
+def _get_default_port_for_scheme(scheme: str) -> int:
if scheme == 'amqps':
return 5671
else:
@@ -1160,7 +1173,7 @@
for creating connections and sender- or receiver- links.
"""
- def __init__(self, *handlers, **kwargs):
+ def __init__(self, *handlers, **kwargs) -> None:
super(Container, self).__init__(*handlers, **kwargs)
if "impl" not in kwargs:
try:
@@ -1366,7 +1379,7 @@
else:
return "%s-%s" % (container, str(_generate_uuid()))
- def _get_session(self, context):
+ def _get_session(self, context: Connection) -> Session:
if isinstance(context, Url):
return self._get_session(self.connect(url=context))
elif isinstance(context, Session):
@@ -1379,7 +1392,16 @@
else:
return context.session()
- def create_sender(self, context, target=None, source=None, name=None, handler=None, tags=None, options=None):
+ def create_sender(
+ self,
+ context: Union[str, Connection],
+ target: Optional[str] = None,
+ source: Optional[str] = None,
+ name: Optional[str] = None,
+ handler: Optional[Handler] = None,
+ tags: Optional[Callable[[], bytes]] = None,
+ options: Optional[Union['SenderOption', List['SenderOption']]] = None
+ ) -> 'Sender':
"""
Initiates the establishment of a link over which messages can
be sent.
@@ -1404,28 +1426,14 @@
attachment.
:param context: A connection object or a URL.
- :type context: :class:`proton.Connection` or ``str``
-
:param target: Address of target node.
- :type target: ``str``
-
:param source: Address of source node.
- :type source: ``str``
-
:param name: Sender name.
- :type name: ``str``
-
:param handler: Event handler for this sender.
- :type handler: Any child class of :class:`proton.Handler`
-
:param tags: Function to generate tags for this sender of the form ``def simple_tags():`` and returns a ``bytes`` type
- :type tags: function pointer
-
:param options: A single option, or a list of sender options
- :type options: :class:`SenderOption` or [SenderOption, SenderOption, ...]
:return: New sender instance.
- :rtype: :class:`proton.Sender`
"""
if isstring(context):
context = Url(context)
@@ -1445,7 +1453,16 @@
snd.open()
return snd
- def create_receiver(self, context, source=None, target=None, name=None, dynamic=False, handler=None, options=None):
+ def create_receiver(
+ self,
+ context: Union[Connection, str],
+ source: Optional[str] = None,
+ target: Optional[str] = None,
+ name: Optional[str] = None,
+ dynamic: bool = False,
+ handler: Optional[Handler] = None,
+ options: Optional[Union[ReceiverOption, List[ReceiverOption]]] = None
+ ) -> 'Receiver':
"""
Initiates the establishment of a link over which messages can
be received (aka a subscription).
@@ -1470,28 +1487,14 @@
attachment.
:param context: A connection object or a URL.
- :type context: :class:`proton.Connection` or ``str``
-
:param source: Address of source node.
- :type source: ``str``
-
:param target: Address of target node.
- :type target: ``str``
-
:param name: Receiver name.
- :type name: ``str``
-
:param dynamic: If ``True``, indicates dynamic creation of the receiver.
- :type dynamic: ``bool``
-
:param handler: Event handler for this receiver.
- :type handler: Any child class of :class:`proton.Handler`
-
:param options: A single option, or a list of receiver options
- :type options: :class:`ReceiverOption` or [ReceiverOption, ReceiverOption, ...]
:return: New receiver instance.
- :rtype: :class:`proton.Receiver`
"""
if isstring(context):
context = Url(context)
@@ -1511,17 +1514,19 @@
rcv.open()
return rcv
- def declare_transaction(self, context, handler=None, settle_before_discharge=False):
+ def declare_transaction(
+ self,
+ context: Connection,
+ handler: Optional['TransactionHandler'] = None,
+ settle_before_discharge: bool = False
+ ) -> Transaction:
"""
Declare a local transaction.
:param context: Context for the transaction, usually the connection.
- :type context: :class:`proton.Connection`
:param handler: Handler for transactional events.
- :type handler: :class:`proton.handlers.TransactionHandler`
:param settle_before_discharge: Settle all transaction control messages before
the transaction is discharged.
- :type settle_before_discharge: ``bool``
"""
if not _get_attr(context, '_txn_ctrl'):
class InternalTransactionHandler(OutgoingMessageHandler):
@@ -1565,7 +1570,7 @@
acceptor.set_ssl_domain(ssl_config)
return acceptor
- def do_work(self, timeout=None):
+ def do_work(self, timeout: Optional[float] = None) -> bool:
if timeout:
self.timeout = timeout
return self.process()
diff --git a/python/proton/_selectable.py b/python/proton/_selectable.py
index 678d96b..a54d2b4 100644
--- a/python/proton/_selectable.py
+++ b/python/proton/_selectable.py
@@ -17,12 +17,16 @@
# under the License.
#
-from __future__ import absolute_import
-
-
from ._events import Event
from ._io import PN_INVALID_SOCKET
+from typing import Callable, Optional, Union, TYPE_CHECKING
+if TYPE_CHECKING:
+ from ._events import EventType
+ from ._handlers import ConnectSelectable
+ from ._reactor import Container, EventInjector, TimerSelectable
+ from socket import socket
+
class Selectable(object):
@@ -37,11 +41,11 @@
self._reactor = reactor
self.push_event(self, Event.SELECTABLE_INIT)
- def close(self):
+ def close(self) -> None:
if self._delegate and not self._released:
self._delegate.close()
- def fileno(self):
+ def fileno(self) -> int:
if self._delegate:
return self._delegate.fileno()
else:
@@ -68,7 +72,7 @@
def push_event(self, context, etype):
self._reactor.push_event(context, etype)
- def update(self):
+ def update(self) -> None:
if not self._terminated:
if self._terminal:
self._terminated = True
@@ -76,21 +80,21 @@
else:
self.push_event(self, Event.SELECTABLE_UPDATED)
- def readable(self):
+ def readable(self) -> None:
self.push_event(self, Event.SELECTABLE_READABLE)
- def writable(self):
+ def writable(self) -> None:
self.push_event(self, Event.SELECTABLE_WRITABLE)
- def expired(self):
+ def expired(self) -> None:
self.push_event(self, Event.SELECTABLE_EXPIRED)
@property
- def is_terminal(self):
+ def is_terminal(self) -> bool:
return self._terminal
- def terminate(self):
+ def terminate(self) -> None:
self._terminal = True
- def release(self):
+ def release(self) -> None:
self._released = True
diff --git a/python/proton/_transport.py b/python/proton/_transport.py
index 5764384..50f2720 100644
--- a/python/proton/_transport.py
+++ b/python/proton/_transport.py
@@ -17,8 +17,6 @@
# under the License.
#
-from __future__ import absolute_import
-
from cproton import PN_EOS, PN_OK, PN_SASL_AUTH, PN_SASL_NONE, PN_SASL_OK, PN_SASL_PERM, PN_SASL_SYS, PN_SASL_TEMP, \
PN_SSL_ANONYMOUS_PEER, PN_SSL_CERT_SUBJECT_CITY_OR_LOCALITY, PN_SSL_CERT_SUBJECT_COMMON_NAME, \
PN_SSL_CERT_SUBJECT_COUNTRY_NAME, PN_SSL_CERT_SUBJECT_ORGANIZATION_NAME, PN_SSL_CERT_SUBJECT_ORGANIZATION_UNIT, \
@@ -47,6 +45,12 @@
from ._exceptions import EXCEPTIONS, SSLException, SSLUnavailable, SessionException, TransportException
from ._wrapper import Wrapper
+from typing import Callable, Optional, Type, Union, TYPE_CHECKING, List
+
+if TYPE_CHECKING:
+ from ._condition import Condition
+ from ._endpoints import Connection # would produce circular import
+
class TraceAdapter:
@@ -96,13 +100,13 @@
else:
raise TransportException("Cannot initialise Transport from mode: %s" % str(mode))
- def _init(self):
+ def _init(self) -> None:
self._sasl = None
self._ssl = None
self._reactor = None
self._connect_selectable = None
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, TransportException)
raise exc("[%s]: %s" % (err, pn_error_text(pn_transport_error(self._impl))))
@@ -138,7 +142,7 @@
"""
pn_transport_log(self._impl, message)
- def require_auth(self, bool):
+ def require_auth(self, bool: bool) -> None:
"""
Set whether a non-authenticated transport connection is allowed.
@@ -151,12 +155,11 @@
The default if this option is not set is to allow unauthenticated connections.
:param bool: ``True`` when authenticated connections are required.
- :type bool: ``bool``
"""
pn_transport_require_auth(self._impl, bool)
@property
- def authenticated(self):
+ def authenticated(self) -> bool:
"""
Indicate whether the transport connection is authenticated.
@@ -184,19 +187,17 @@
pn_transport_require_encryption(self._impl, bool)
@property
- def encrypted(self):
+ def encrypted(self) -> bool:
"""
Indicate whether the transport connection is encrypted.
.. note:: This property may not be stable until the :const:`Event.CONNECTION_REMOTE_OPEN`
event is received.
-
- :type: ``bool``
"""
return pn_transport_is_encrypted(self._impl)
@property
- def user(self):
+ def user(self) -> Optional[str]:
"""
The authenticated user.
@@ -205,32 +206,28 @@
The returned value is only reliable after the ``PN_TRANSPORT_AUTHENTICATED``
event has been received.
-
- :type: ``str``
"""
return pn_transport_get_user(self._impl)
- def bind(self, connection):
+ def bind(self, connection: 'Connection') -> None:
"""
Assign a connection to the transport.
:param connection: Connection to which to bind.
- :type connection: :class:`Connection`
:raise: :exc:`TransportException` if there is any Proton error.
"""
self._check(pn_transport_bind(self._impl, connection._impl))
- def bind_nothrow(self, connection):
+ def bind_nothrow(self, connection: 'Connection') -> None:
"""
Assign a connection to the transport. Any failure is
ignored rather than thrown.
:param connection: Connection to which to bind.
- :type connection: :class:`Connection`
"""
pn_transport_bind(self._impl, connection._impl)
- def unbind(self):
+ def unbind(self) -> None:
"""
Unbinds a transport from its AMQP connection.
@@ -238,7 +235,7 @@
"""
self._check(pn_transport_unbind(self._impl))
- def trace(self, n):
+ def trace(self, n: int) -> None:
"""
Update a transports trace flags.
@@ -248,11 +245,10 @@
a bitwise or operation.
:param n: Trace flags
- :type n: ``int``
"""
pn_transport_trace(self._impl, n)
- def tick(self, now):
+ def tick(self, now: float) -> float:
"""
Process any pending transport timer events (like heartbeat generation).
@@ -264,22 +260,19 @@
from or written to the transport.
:param now: seconds since epoch.
- :type now: ``float``
:return: If non-zero, then the monotonic expiration time of the next
pending timer event for the transport. The caller must invoke
:meth:`tick` again at least once at or before this deadline
occurs. If ``0.0``, then there are no pending events.
- :rtype: ``float``
"""
return millis2secs(pn_transport_tick(self._impl, secs2millis(now)))
- def capacity(self):
+ def capacity(self) -> int:
"""
Get the amount of free space for input following the transport's
tail pointer.
:return: Available space for input in bytes.
- :rtype: ``int``
:raise: :exc:`TransportException` if there is any Proton error.
"""
c = pn_transport_capacity(self._impl)
@@ -288,7 +281,7 @@
else:
return self._check(c)
- def push(self, binary):
+ def push(self, binary: bytes) -> None:
"""
Pushes the supplied bytes into the tail of the transport.
Only some of the bytes will be copied if there is insufficient
@@ -296,7 +289,6 @@
capacity the transport has.
:param binary: Data to be pushed onto the transport tail.
- :type binary: ``bytes``
:raise: - :exc:`TransportException` if there is any Proton error.
- ``OverflowError`` if the size of the data exceeds the
transport capacity.
@@ -305,7 +297,7 @@
if n != len(binary):
raise OverflowError("unable to process all bytes: %s, %s" % (n, len(binary)))
- def close_tail(self):
+ def close_tail(self) -> None:
"""
Indicate that the input has reached End Of Stream (EOS).
@@ -315,13 +307,12 @@
"""
self._check(pn_transport_close_tail(self._impl))
- def pending(self):
+ def pending(self) -> int:
"""
Get the number of pending output bytes following the transport's
head pointer.
:return: The number of pending output bytes.
- :rtype: ``int``
:raise: :exc:`TransportException` if there is any Proton error.
"""
p = pn_transport_pending(self._impl)
@@ -330,7 +321,7 @@
else:
return self._check(p)
- def peek(self, size):
+ def peek(self, size: int) -> Optional[bytes]:
"""
Returns ``size`` bytes from the head of the transport.
@@ -338,10 +329,8 @@
is greater than the value reported by :meth:`pending`.
:param size: Number of bytes to return.
- :type size: ``int``
:return: ``size`` bytes from the head of the transport, or ``None``
if none are available.
- :rtype: ``bytes``
:raise: :exc:`TransportException` if there is any Proton error.
"""
cd, out = pn_transport_peek(self._impl, size)
@@ -365,7 +354,7 @@
"""
pn_transport_pop(self._impl, size)
- def close_head(self):
+ def close_head(self) -> None:
"""
Indicate that the output has closed.
@@ -376,12 +365,10 @@
self._check(pn_transport_close_head(self._impl))
@property
- def closed(self):
+ def closed(self) -> bool:
"""
``True`` iff both the transport head and transport tail are closed
using :meth:`close_head` and :meth:`close_tail` respectively.
-
- :type: ``bool``
"""
return pn_transport_closed(self._impl)
@@ -459,54 +446,44 @@
""")
@property
- def remote_idle_timeout(self):
+ def remote_idle_timeout(self) -> float:
"""
Get the idle timeout for a transport's remote peer in
seconds. A zero idle timeout means heartbeats are disabled.
-
- :type: ``float``
"""
return millis2secs(pn_transport_get_remote_idle_timeout(self._impl))
@property
- def frames_output(self):
+ def frames_output(self) -> int:
"""
Get the number of frames output by a transport.
-
- :type: ``int``
"""
return pn_transport_get_frames_output(self._impl)
@property
- def frames_input(self):
+ def frames_input(self) -> int:
"""
Get the number of frames input by a transport.
-
- :type: ``int``
"""
return pn_transport_get_frames_input(self._impl)
- def sasl(self):
+ def sasl(self) -> 'SASL':
"""
Get the :class:`SASL` object associated with this transport.
:return: SASL object associated with this transport.
- :rtype: :class:`SASL`
"""
return SASL(self)
- def ssl(self, domain=None, session_details=None):
+ def ssl(self, domain: Optional['SSLDomain'] = None, session_details: Optional['SSLSessionDetails'] = None) -> 'SSL':
"""
Get the :class:`SSL` session associated with this transport. If
not set, then a new session will be created using ``domain`` and
``session_details``.
:param domain: An SSL domain configuration object
- :type domain: :class:`SSLDomain`
:param session_details: A unique identifier for the SSL session.
- :type session_details: :class:`SSLSessionDetails`
:return: SSL session associated with this transport.
- :rtype: :class:`SSL`
"""
# SSL factory (singleton for this transport)
if not self._ssl:
@@ -561,7 +538,7 @@
TEMP = PN_SASL_TEMP
@staticmethod
- def extended():
+ def extended() -> bool:
"""
Check for support of extended SASL negotiation.
@@ -576,7 +553,7 @@
"""
return pn_sasl_extended()
- def __init__(self, transport):
+ def __init__(self, transport: Transport) -> None:
Wrapper.__init__(self, transport._impl, pn_transport_attachments)
self._sasl = pn_sasl(transport._impl)
@@ -631,7 +608,7 @@
return pn_sasl_get_authorization(self._sasl)
@property
- def mech(self):
+ def mech(self) -> str:
"""
Return the selected SASL mechanism.
@@ -643,7 +620,7 @@
return pn_sasl_get_mech(self._sasl)
@property
- def outcome(self):
+ def outcome(self) -> Optional[int]:
"""
Retrieve the outcome of SASL negotiation.
@@ -656,7 +633,7 @@
else:
return outcome
- def allowed_mechs(self, mechs):
+ def allowed_mechs(self, mechs: Union[str, List[str]]) -> None:
"""
SASL mechanisms that are to be considered for authentication.
@@ -678,7 +655,6 @@
either a string containing a space-separated list of mechs
``"mech1 mech2 ..."``, or a Python list of strings
``["mech1", "mech2", ...]``.
- :type mechs: string
"""
if isinstance(mechs, list):
mechs = " ".join(mechs)
@@ -702,7 +678,7 @@
"""
pn_sasl_done(self._sasl, outcome)
- def config_name(self, name):
+ def config_name(self, name: str):
"""
Set the SASL configuration name. This is used to construct the SASL
configuration filename. In the current implementation ``".conf"`` is
@@ -713,11 +689,10 @@
and ``"proton-client"`` for a client.
:param name: The configuration name.
- :type name: string
"""
pn_sasl_config_name(self._sasl, name)
- def config_path(self, path):
+ def config_path(self, path: str):
"""
Set the SASL configuration path. This is used to tell SASL where
to look for the configuration file. In the current implementation
@@ -731,7 +706,6 @@
:param path: The configuration path, may contain colon-separated list
if more than one path is specified.
- :type path: string
"""
pn_sasl_config_path(self._sasl, path)
@@ -757,19 +731,19 @@
ANONYMOUS_PEER = PN_SSL_ANONYMOUS_PEER
"""Do not require a certificate nor cipher authorization."""
- def __init__(self, mode):
+ def __init__(self, mode: int) -> None:
self._domain = pn_ssl_domain(mode)
if self._domain is None:
raise SSLUnavailable()
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, SSLException)
raise exc("SSL failure.")
else:
return err
- def set_credentials(self, cert_file, key_file, password):
+ def set_credentials(self, cert_file: str, key_file: str, password: Optional[str]) -> int:
"""
Set the certificate that identifies the local node to the remote.
@@ -785,25 +759,21 @@
:param cert_file: Specifier for the file/database containing the identifying
certificate. For Openssl users, this is a PEM file. For Windows SChannel
users, this is the PKCS#12 file or system store.
- :type cert_file: ``str``
:param key_file: An optional key to access the identifying certificate. For
Openssl users, this is an optional PEM file containing the private key
used to sign the certificate. For Windows SChannel users, this is the
friendly name of the self-identifying certificate if there are multiple
certificates in the store.
- :type key_file: ``str``
:param password: The password used to sign the key, else ``None`` if key is not
protected.
- :type password: ``str`` or ``None``
:return: 0 on success
- :rtype: ``int``
:raise: :exc:`SSLException` if there is any Proton error
"""
return self._check(pn_ssl_domain_set_credentials(self._domain,
cert_file, key_file,
password))
- def set_trusted_ca_db(self, certificate_db):
+ def set_trusted_ca_db(self, certificate_db: str) -> int:
"""
Configure the set of trusted CA certificates used by this domain to verify peers.
@@ -821,15 +791,13 @@
will be the users default trusted certificate store.
:param certificate_db: Database of trusted CAs, used to authenticate the peer.
- :type certificate_db: ``str``
:return: 0 on success
- :rtype: ``int``
:raise: :exc:`SSLException` if there is any Proton error
"""
return self._check(pn_ssl_domain_set_trusted_ca_db(self._domain,
certificate_db))
- def set_peer_authentication(self, verify_mode, trusted_CAs=None):
+ def set_peer_authentication(self, verify_mode: int, trusted_CAs: Optional[str] = None) -> int:
"""
This method controls how the peer's certificate is validated, if at all. By default,
servers do not attempt to verify their peers (PN_SSL_ANONYMOUS_PEER) but
@@ -848,18 +816,15 @@
:param verify_mode: The level of validation to apply to the peer, one of :const:`VERIFY_PEER`,
:const:`VERIFY_PEER_NAME`, :const:`ANONYMOUS_PEER`,
- :type verify_mode: ``int``
:param trusted_CAs: Path to a database of trusted CAs that the server will advertise.
- :type trusted_CAs: ``str``
:return: 0 on success
- :rtype: ``int``
:raise: :exc:`SSLException` if there is any Proton error
"""
return self._check(pn_ssl_domain_set_peer_authentication(self._domain,
verify_mode,
trusted_CAs))
- def allow_unsecured_client(self):
+ def allow_unsecured_client(self) -> int:
"""
Permit a server to accept connection requests from non-SSL clients.
@@ -871,7 +836,7 @@
"""
return self._check(pn_ssl_domain_allow_unsecured_client(self._domain))
- def __del__(self):
+ def __del__(self) -> None:
pn_ssl_domain_free(self._domain)
@@ -882,16 +847,15 @@
"""
@staticmethod
- def present():
+ def present() -> bool:
"""
Tests for an SSL implementation being present.
:return: ``True`` if we support SSL, ``False`` if not.
- :rtype: ``bool``
"""
return pn_ssl_present()
- def _check(self, err):
+ def _check(self, err: int) -> int:
if err < 0:
exc = EXCEPTIONS.get(err, SSLException)
raise exc("SSL failure.")
@@ -923,7 +887,7 @@
transport._ssl = obj
return transport._ssl
- def cipher_name(self):
+ def cipher_name(self) -> Optional[str]:
"""
Get the name of the Cipher that is currently in use.
@@ -934,14 +898,13 @@
or other changes to the SSL state.
:return: The cypher name, or ``None`` if no cipher in use.
- :rtype: ``str`` or ``None``
"""
rc, name = pn_ssl_get_cipher_name(self._ssl, 128)
if rc:
return name
return None
- def protocol_name(self):
+ def protocol_name(self) -> Optional[str]:
"""
Get the name of the SSL protocol that is currently in use.
@@ -952,7 +915,6 @@
:return: The protocol name if SSL is active, or ``None`` if SSL connection
is not ready or active.
- :rtype: ``str`` or ``None``
"""
rc, name = pn_ssl_get_protocol_name(self._ssl, 128)
if rc:
@@ -989,7 +951,7 @@
CERT_COMMON_NAME = PN_SSL_CERT_SUBJECT_COMMON_NAME
"""Certificate common name or URL"""
- def get_cert_subject_subfield(self, subfield_name):
+ def get_cert_subject_subfield(self, subfield_name: int) -> Optional[str]:
"""
Returns a string that contains the value of the sub field of
the subject field in the ssl certificate. The subject field
@@ -1004,86 +966,77 @@
:param subfield_name: The enumeration representing the required
sub field listed above
- :type subfield_name: ``int``
:return: A string which contains the requested sub field value which
is valid until the ssl object is destroyed.
- :rtype: ``str``
"""
subfield_value = pn_ssl_get_remote_subject_subfield(self._ssl, subfield_name)
return subfield_value
- def get_cert_subject(self):
+ def get_cert_subject(self) -> str:
"""
Get the subject from the peer's certificate.
:return: A string containing the full subject.
- :rtype: ``str``
"""
subject = pn_ssl_get_remote_subject(self._ssl)
return subject
- def _get_cert_subject_unknown_subfield(self):
+ def _get_cert_subject_unknown_subfield(self) -> None:
# Pass in an unhandled enum
return self.get_cert_subject_subfield(10)
# Convenience functions for obtaining the subfields of the subject field.
- def get_cert_common_name(self):
+ def get_cert_common_name(self) -> str:
"""
A convenience method to get a string that contains the :const:`CERT_COMMON_NAME`
sub field of the subject field in the ssl certificate.
:return: A string containing the :const:`CERT_COMMON_NAME` sub field.
- :rtype: ``str``
"""
return self.get_cert_subject_subfield(SSL.CERT_COMMON_NAME)
- def get_cert_organization(self):
+ def get_cert_organization(self) -> str:
"""
A convenience method to get a string that contains the :const:`CERT_ORGANIZATION_NAME`
sub field of the subject field in the ssl certificate.
:return: A string containing the :const:`CERT_ORGANIZATION_NAME` sub field.
- :rtype: ``str``
"""
return self.get_cert_subject_subfield(SSL.CERT_ORGANIZATION_NAME)
- def get_cert_organization_unit(self):
+ def get_cert_organization_unit(self) -> str:
"""
A convenience method to get a string that contains the :const:`CERT_ORGANIZATION_UNIT`
sub field of the subject field in the ssl certificate.
:return: A string containing the :const:`CERT_ORGANIZATION_UNIT` sub field.
- :rtype: ``str``
"""
return self.get_cert_subject_subfield(SSL.CERT_ORGANIZATION_UNIT)
- def get_cert_locality_or_city(self):
+ def get_cert_locality_or_city(self) -> str:
"""
A convenience method to get a string that contains the :const:`CERT_CITY_OR_LOCALITY`
sub field of the subject field in the ssl certificate.
:return: A string containing the :const:`CERT_CITY_OR_LOCALITY` sub field.
- :rtype: ``str``
"""
return self.get_cert_subject_subfield(SSL.CERT_CITY_OR_LOCALITY)
- def get_cert_country(self):
+ def get_cert_country(self) -> str:
"""
A convenience method to get a string that contains the :const:`CERT_COUNTRY_NAME`
sub field of the subject field in the ssl certificate.
:return: A string containing the :const:`CERT_COUNTRY_NAME` sub field.
- :rtype: ``str``
"""
return self.get_cert_subject_subfield(SSL.CERT_COUNTRY_NAME)
- def get_cert_state_or_province(self):
+ def get_cert_state_or_province(self) -> str:
"""
A convenience method to get a string that contains the :const:`CERT_STATE_OR_PROVINCE`
sub field of the subject field in the ssl certificate.
:return: A string containing the :const:`CERT_STATE_OR_PROVINCE` sub field.
- :rtype: ``str``
"""
return self.get_cert_subject_subfield(SSL.CERT_STATE_OR_PROVINCE)
@@ -1112,57 +1065,51 @@
return None
# Convenience functions for obtaining fingerprint for specific hashing algorithms
- def _get_cert_fingerprint_unknown_hash_alg(self):
+ def _get_cert_fingerprint_unknown_hash_alg(self) -> None:
return self.get_cert_fingerprint(41, 10)
- def get_cert_fingerprint_sha1(self):
+ def get_cert_fingerprint_sha1(self) -> Optional[str]:
"""
A convenience method to get the :const:`SHA1` fingerprint of the
certificate.
:return: Hex fingerprint in a string, or ``None`` if an error occurred.
- :rtype: ``str`` or ``None``
"""
return self.get_cert_fingerprint(41, SSL.SHA1)
- def get_cert_fingerprint_sha256(self):
+ def get_cert_fingerprint_sha256(self) -> Optional[str]:
"""
A convenience method to get the :const:`SHA256` fingerprint of the
certificate.
:return: Hex fingerprint in a string, or ``None`` if an error occurred.
- :rtype: ``str`` or ``None``
"""
# sha256 produces a fingerprint that is 64 characters long
return self.get_cert_fingerprint(65, SSL.SHA256)
- def get_cert_fingerprint_sha512(self):
+ def get_cert_fingerprint_sha512(self) -> Optional[str]:
"""
A convenience method to get the :const:`SHA512` fingerprint of the
certificate.
:return: Hex fingerprint in a string, or ``None`` if an error occurred.
- :rtype: ``str`` or ``None``
"""
# sha512 produces a fingerprint that is 128 characters long
return self.get_cert_fingerprint(129, SSL.SHA512)
- def get_cert_fingerprint_md5(self):
+ def get_cert_fingerprint_md5(self) -> Optional[str]:
"""
A convenience method to get the :const:`MD5` fingerprint of the
certificate.
:return: Hex fingerprint in a string, or ``None`` if an error occurred.
- :rtype: ``str`` or ``None``
"""
return self.get_cert_fingerprint(33, SSL.MD5)
@property
- def remote_subject(self):
+ def remote_subject(self) -> str:
"""
The subject from the peers certificate.
-
- :type: ``str``
"""
return pn_ssl_get_remote_subject(self._ssl)
@@ -1231,14 +1178,13 @@
session on a new SSL connection.
"""
- def __init__(self, session_id):
+ def __init__(self, session_id: str) -> None:
self._session_id = session_id
- def get_session_id(self):
+ def get_session_id(self) -> str:
"""
Get the unique identifier for this SSL session
:return: Session identifier
- :rtype: ``str``
"""
return self._session_id
diff --git a/python/proton/_utils.py b/python/proton/_utils.py
index 275f12b..f14db9f 100644
--- a/python/proton/_utils.py
+++ b/python/proton/_utils.py
@@ -17,8 +17,6 @@
# under the License.
#
-from __future__ import absolute_import
-
import collections
import time
import threading
@@ -32,9 +30,24 @@
from ._reactor import Container
from ._handlers import MessagingHandler, IncomingMessageHandler
+from typing import Callable, Optional, Union, TYPE_CHECKING, List
+try:
+ from typing import Literal
+except ImportError:
+ class Literal:
+ def __class_getitem__(cls, item):
+ pass
-class BlockingLink(object):
- def __init__(self, connection, link):
+if TYPE_CHECKING:
+ from ._transport import SSLDomain
+ from ._reactor import Backoff, SenderOption, ReceiverOption, Connection
+ from ._endpoints import Receiver, Sender, Terminus
+ from ._events import Event
+ from ._message import Message
+
+
+class BlockingLink:
+ def __init__(self, connection: 'BlockingConnection', link: Union['Sender', 'Receiver']) -> None:
self.connection = connection
self.link = link
self.connection.wait(lambda: not (self.link.state & Endpoint.REMOTE_UNINIT),
@@ -50,7 +63,7 @@
pass
self._checkClosed()
- def _checkClosed(self):
+ def _checkClosed(self) -> None:
if self.link.state & Endpoint.REMOTE_CLOSED:
self.link.close()
if not self.connection.closing:
@@ -74,14 +87,13 @@
Exception used to indicate an exceptional state/condition on a send request.
:param state: The delivery state which caused the exception.
- :type state: ``int``
"""
- def __init__(self, state):
+ def __init__(self, state: int) -> None:
self.state = state
-def _is_settled(delivery):
+def _is_settled(delivery: Delivery) -> bool:
return delivery.settled or delivery.link.snd_settle_mode == Link.SND_SETTLED
@@ -91,7 +103,7 @@
:meth:`BlockingConnection.create_sender`.
"""
- def __init__(self, connection, sender):
+ def __init__(self, connection: 'BlockingConnection', sender: 'Sender') -> None:
super(BlockingSender, self).__init__(connection, sender)
if self.link.target and self.link.target.address and self.link.target.address != self.link.remote_target.address:
# this may be followed by a detach, which may contain an error condition, so wait a little...
@@ -145,44 +157,40 @@
self.incoming = collections.deque([])
self.unsettled = collections.deque([])
- def on_message(self, event):
+ def on_message(self, event: 'Event') -> None:
self.incoming.append((event.message, event.delivery))
self.connection.container.yield_() # Wake up the wait() loop to handle the message.
- def on_link_error(self, event):
+ def on_link_error(self, event: 'Event') -> None:
if event.link.state & Endpoint.LOCAL_ACTIVE:
event.link.close()
if not self.connection.closing:
raise LinkDetached(event.link)
- def on_connection_error(self, event):
+ def on_connection_error(self, event: 'Event') -> None:
if not self.connection.closing:
raise ConnectionClosed(event.connection)
@property
- def has_message(self):
+ def has_message(self) -> int:
"""
The number of messages that have been received and are waiting to be
retrieved with :meth:`pop`.
-
- :type: ``int``
"""
return len(self.incoming)
- def pop(self):
+ def pop(self) -> 'Message':
"""
Get the next available incoming message. If the message is unsettled, its
delivery object is moved onto the unsettled queue, and can be settled with
a call to :meth:`settle`.
-
- :rtype: :class:`proton.Message`
"""
message, delivery = self.incoming.popleft()
if not delivery.settled:
self.unsettled.append(delivery)
return message
- def settle(self, state=None):
+ def settle(self, state: Optional[int] = None) -> None:
"""
Settle the next message previously taken with :meth:`pop`.
@@ -201,7 +209,13 @@
:meth:`BlockingConnection.create_receiver`.
"""
- def __init__(self, connection, receiver, fetcher, credit=1):
+ def __init__(
+ self,
+ connection: 'BlockingConnection',
+ receiver: 'Receiver',
+ fetcher: Optional[Fetcher],
+ credit: int = 1
+ ) -> None:
super(BlockingReceiver, self).__init__(connection, receiver)
if self.link.source and self.link.source.address and self.link.source.address != self.link.remote_source.address:
# this may be followed by a detach, which may contain an error condition, so wait a little...
@@ -240,21 +254,21 @@
timeout=timeout)
return self.fetcher.pop()
- def accept(self):
+ def accept(self) -> None:
"""
Accept and settle the received message. The delivery is set to
:const:`proton.Delivery.ACCEPTED`.
"""
self.settle(Delivery.ACCEPTED)
- def reject(self):
+ def reject(self) -> None:
"""
Reject the received message. The delivery is set to
:const:`proton.Delivery.REJECTED`.
"""
self.settle(Delivery.REJECTED)
- def release(self, delivered=True):
+ def release(self, delivered: bool = True) -> None:
"""
Release the received message.
@@ -262,7 +276,6 @@
:const:`proton.Delivery.MODIFIED`, ie being returned to the sender
and annotated. If ``False``, the message is returned without
annotations and the delivery set to :const:`proton.Delivery.RELEASED`.
- :type delivered: ``bool``
"""
if delivered:
self.settle(Delivery.MODIFIED)
@@ -289,10 +302,9 @@
context, or an unexpected link error occurs.
:param link: The link which closed unexpectedly.
- :type link: :class:`proton.Link`
"""
- def __init__(self, link):
+ def __init__(self, link: Link) -> None:
self.link = link
if link.is_sender:
txt = "sender %s to %s closed" % (link.name, link.target.address)
@@ -313,10 +325,9 @@
context, or an unexpected connection error occurs.
:param connection: The connection which closed unexpectedly.
- :type connection: :class:`proton.Connection`
"""
- def __init__(self, connection):
+ def __init__(self, connection: 'Connection') -> None:
self.connection = connection
txt = "Connection %s closed" % connection.hostname
if connection.remote_condition:
@@ -377,7 +388,13 @@
if failed and self.conn:
self.close()
- def create_sender(self, address, handler=None, name=None, options=None):
+ def create_sender(
+ self,
+ address: Optional[str],
+ handler: Optional[Handler] = None,
+ name: Optional[str] = None,
+ options: Optional[Union['SenderOption', List['SenderOption']]] = None
+ ) -> BlockingSender:
"""
Create a blocking sender.
@@ -395,24 +412,25 @@
return BlockingSender(self, self.container.create_sender(self.conn, address, name=name, handler=handler,
options=options))
- def create_receiver(self, address, credit=None, dynamic=False, handler=None, name=None, options=None):
+ def create_receiver(
+ self,
+ address: Optional[str] = None,
+ credit: Optional[int] = None,
+ dynamic: bool = False,
+ handler: Optional[Handler] = None,
+ name: Optional[str] = None,
+ options: Optional[Union['ReceiverOption', List['ReceiverOption']]] = None
+ ) -> BlockingReceiver:
"""
Create a blocking receiver.
:param address: Address of source node.
- :type address: ``str``
:param credit: Initial link flow credit. If not set, will default to 1.
- :type credit: ``int``
:param dynamic: If ``True``, indicates dynamic creation of the receiver.
- :type dynamic: ``bool``
:param handler: Event handler for this receiver.
- :type handler: Any child class of :class:`proton.Handler`
:param name: Receiver name.
- :type name: ``str``
:param options: A single option, or a list of receiver options
- :type options: :class:`ReceiverOption` or [ReceiverOption, ReceiverOption, ...]
:return: New blocking receiver instance.
- :rtype: :class:`BlockingReceiver`
"""
prefetch = credit
if handler:
@@ -426,7 +444,7 @@
self.container.create_receiver(self.conn, address, name=name, dynamic=dynamic, handler=handler or fetcher,
options=options), fetcher, credit=prefetch)
- def close(self):
+ def close(self) -> None:
"""
Close the connection.
"""
@@ -454,18 +472,16 @@
self.container = None
@property
- def url(self):
+ def url(self) -> str:
"""
The address for this connection.
-
- :type: ``str``
"""
return self.conn and self.conn.connected_address
- def _is_closed(self):
+ def _is_closed(self) -> int:
return self.conn.state & (Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED)
- def run(self):
+ def run(self) -> None:
"""
Hand control over to the event loop (e.g. if waiting indefinitely for incoming messages)
"""
@@ -510,7 +526,7 @@
raise ConnectionException(
"Connection %s disconnected: %s" % (self.url, self.disconnected))
- def on_link_remote_close(self, event):
+ def on_link_remote_close(self, event: 'Event') -> None:
"""
Event callback for when the remote terminus closes.
"""
@@ -519,7 +535,7 @@
if not self.closing:
raise LinkDetached(event.link)
- def on_connection_remote_close(self, event):
+ def on_connection_remote_close(self, event: 'Event') -> None:
"""
Event callback for when the link peer closes the connection.
"""
@@ -528,24 +544,24 @@
if not self.closing:
raise ConnectionClosed(event.connection)
- def on_transport_tail_closed(self, event):
+ def on_transport_tail_closed(self, event: 'Event') -> None:
self.on_transport_closed(event)
- def on_transport_head_closed(self, event):
+ def on_transport_head_closed(self, event: 'Event') -> None:
self.on_transport_closed(event)
- def on_transport_closed(self, event):
+ def on_transport_closed(self, event: 'Event') -> None:
if not self.closing:
self.disconnected = event.transport.condition or "unknown"
-class AtomicCount(object):
- def __init__(self, start=0, step=1):
+class AtomicCount:
+ def __init__(self, start: int = 0, step: int = 1) -> None:
"""Thread-safe atomic counter. Start at start, increment by step."""
self.count, self.step = start, step
self.lock = threading.Lock()
- def next(self):
+ def next(self) -> int:
"""Get the next value"""
self.lock.acquire()
self.count += self.step
@@ -561,16 +577,14 @@
addresses.
:param connection: Connection for requests and responses.
- :type connection: :class:`BlockingConnection`
:param address: Address for all requests. If not specified, each request
must have the address property set. Successive messages may have
different addresses.
- :type address: ``str`` or ``None``
"""
correlation_id = AtomicCount()
- def __init__(self, connection, address=None):
+ def __init__(self, connection: BlockingConnection, address: Optional[str] = None) -> None:
super(SyncRequestResponse, self).__init__()
self.connection = connection
self.address = address
@@ -580,13 +594,12 @@
self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self)
self.response = None
- def call(self, request):
+ def call(self, request: 'Message') -> 'Message':
"""
Send a request message, wait for and return the response message.
:param request: Request message. If ``self.address`` is not set the
request message address must be set and will be used.
- :type request: :class:`proton.Message`
"""
if not self.address and not request.address:
raise ValueError("Request message has no address: %s" % request)
@@ -604,20 +617,17 @@
return response
@property
- def reply_to(self):
+ def reply_to(self) -> str:
"""
The dynamic address of our receiver.
-
- :type: ``str``
"""
return self.receiver.remote_source.address
- def on_message(self, event):
+ def on_message(self, event: 'Event') -> None:
"""
Called when we receive a message for our receiver.
:param event: The event which occurs when a message is received.
- :type event: :class:`proton.Event`
"""
self.response = event.message
self.connection.container.yield_() # Wake up the wait() loop to handle the message.
diff --git a/python/proton/_wrapper.py b/python/proton/_wrapper.py
index 4703b34..7ad4c87 100644
--- a/python/proton/_wrapper.py
+++ b/python/proton/_wrapper.py
@@ -17,8 +17,6 @@
# under the License.
#
-from __future__ import absolute_import
-
from cproton import pn_incref, pn_decref, \
pn_py2void, pn_void2py, \
pn_record_get, pn_record_def, pn_record_set, \
@@ -26,6 +24,11 @@
from ._exceptions import ProtonException
+from typing import Any, Callable, Optional, Union, TYPE_CHECKING
+if TYPE_CHECKING:
+ from ._delivery import Delivery # circular import
+ from ._transport import SASL, Transport
+
class EmptyAttrs:
@@ -94,26 +97,26 @@
if init:
self._init()
- def __getattr__(self, name):
+ def __getattr__(self, name: str) -> Any:
attrs = self.__dict__["_attrs"]
if name in attrs:
return attrs[name]
else:
raise AttributeError(name + " not in _attrs")
- def __setattr__(self, name, value):
+ def __setattr__(self, name: str, value: Any) -> None:
if hasattr(self.__class__, name):
object.__setattr__(self, name, value)
else:
attrs = self.__dict__["_attrs"]
attrs[name] = value
- def __delattr__(self, name):
+ def __delattr__(self, name: str) -> None:
attrs = self.__dict__["_attrs"]
if attrs:
del attrs[name]
- def __hash__(self):
+ def __hash__(self) -> int:
return hash(addressof(self._impl))
def __eq__(self, other):
@@ -121,15 +124,15 @@
return addressof(self._impl) == addressof(other._impl)
return False
- def __ne__(self, other):
+ def __ne__(self, other: Any) -> bool:
if isinstance(other, Wrapper):
return addressof(self._impl) != addressof(other._impl)
return True
- def __del__(self):
+ def __del__(self) -> None:
pn_decref(self._impl)
- def __repr__(self):
+ def __repr__(self) -> str:
return '<%s.%s 0x%x ~ 0x%x>' % (self.__class__.__module__,
self.__class__.__name__,
id(self), addressof(self._impl))
