3.0.ipv8.lazy_community ======================= .. py:module:: 3.0.ipv8.lazy_community Attributes ---------- .. autoapisummary:: 3.0.ipv8.lazy_community.UT 3.0.ipv8.lazy_community.CTF Exceptions ---------- .. autoapisummary:: 3.0.ipv8.lazy_community.PacketDecodingError Classes ------- .. autoapisummary:: 3.0.ipv8.lazy_community.EZPackOverlay Functions --------- .. autoapisummary:: 3.0.ipv8.lazy_community.cache_retrieval_failed 3.0.ipv8.lazy_community.retrieve_cache 3.0.ipv8.lazy_community.lazy_wrapper 3.0.ipv8.lazy_community.lazy_wrapper_wd 3.0.ipv8.lazy_community.lazy_wrapper_unsigned 3.0.ipv8.lazy_community.lazy_wrapper_unsigned_wd Module Contents --------------- .. py:data:: UT .. py:data:: CTF .. py:function:: cache_retrieval_failed(overlay: 3.0.ipv8.overlay.Overlay, cache_class: type[3.0.ipv8.types.NumberCache]) -> None Handler for messages which failed to match to an existing overlay cache. .. py:function:: retrieve_cache(cache_class: type[3.0.ipv8.types.NumberCache]) -> Callable[[CTF], CTF] This function wrapper match a payload to a registered cache for you. For this wrapper to function, you will need to comply with three standards: - The last specified payload must include an ``identifier`` attribute. - The ``cache_class`` must specify a ``name`` attribute. - The overlay this method belongs to has a ``request_cache`` attribute. You can now message handlers as follows: :: @lazy_wrapper(MyPayload1, MyPayload2) # MyPayload2.identifier must exist! @retrieve_cache(MyCache) # MyCache.name must exist! def on_message(peer: Peer, payload1: MyPayload1, payload2: MyPayload2, cache: NumberCache): pass :param cache_class: the cache to fetch. .. py:function:: lazy_wrapper(*payloads: type[3.0.ipv8.messaging.serialization.Payload]) -> Callable[[3.0.ipv8.types.LazyWrappedHandler], 3.0.ipv8.types.MessageHandlerFunction] This function wrapper will unpack the BinMemberAuthenticationPayload for you. You can now write your authenticated and signed functions as follows: :: @lazy_wrapper(IntroductionRequestPayload, IntroductionResponsePayload) def on_message(peer: Peer, payload1: IntroductionRequestPayload, payload2: IntroductionResponsePayload): pass .. py:function:: lazy_wrapper_wd(*payloads: type[3.0.ipv8.messaging.serialization.Payload]) -> Callable[[3.0.ipv8.types.LazyWrappedWDataHandler], 3.0.ipv8.types.MessageHandlerFunction] This function wrapper will unpack the BinMemberAuthenticationPayload for you, as well as pass the raw data to the decorated function. You can now write your authenticated and signed functions as follows: :: @lazy_wrapper_wd(IntroductionRequestPayload, IntroductionResponsePayload) def on_message(peer: Peer, payload1: IntroductionRequestPayload, payload2: IntroductionResponsePayload, data: bytes): pass .. py:function:: lazy_wrapper_unsigned(*payloads: type[3.0.ipv8.messaging.serialization.Payload]) -> Callable[[3.0.ipv8.types.LazyWrappedUnsignedHandler], 3.0.ipv8.types.MessageHandlerFunction] This function wrapper will unpack just the normal payloads for you. You can now write your non-authenticated and signed functions as follows: :: @lazy_wrapper_unsigned(IntroductionRequestPayload, IntroductionResponsePayload) def on_message(source_address: Address, payload1: IntroductionRequestPayload, payload2: IntroductionResponsePayload): pass .. py:function:: lazy_wrapper_unsigned_wd(*payloads: type[3.0.ipv8.messaging.serialization.Payload]) -> Callable[[3.0.ipv8.types.LazyWrappedWDataUnsignedHandler], 3.0.ipv8.types.MessageHandlerFunction] This function wrapper will unpack just the normal payloads for you, as well as pass the raw data to the decorated function. You can now write your non-authenticated and signed functions as follows: :: @lazy_wrapper_unsigned_wd(IntroductionRequestPayload, IntroductionResponsePayload) def on_message(source_address: Address, payload1: IntroductionRequestPayload, payload2: IntroductionResponsePayload, data: bytes): pass .. py:class:: EZPackOverlay(settings: SettingsClass) Bases: :py:obj:`3.0.ipv8.overlay.Overlay`, :py:obj:`abc.ABC` Base class that provides you with some easy ways to pack and unpack payloads. .. py:method:: get_prefix() -> bytes :abstractmethod: Get the prefix of this overlay. .. py:method:: ez_send(peer: 3.0.ipv8.peer.Peer, *payloads: 3.0.ipv8.messaging.serialization.Payload, **kwargs) -> None Send a Payload instance (with a defined `msg_id` field) to a peer. If you supply more than one Payload instance, the `msg_id` of the LAST instance will be used. :param peer: the peer to send to :param sig: whether or not to sign this message :type sig: bool :param payloads: the list of Payload instances to serialize .. py:method:: _ez_senda(address: 3.0.ipv8.types.Address, *payloads: 3.0.ipv8.messaging.serialization.Payload, **kwargs) -> None Send a Payload instance to an address. You will probably not need this, try to use `ez_send` instead. :param address: the address to send to :param sig: whether or not to sign this message :type sig: bool :param payloads: the list of Payload instances to serialize .. py:method:: ezr_pack(msg_num: int, *payloads: 3.0.ipv8.messaging.serialization.Payload, **kwargs) -> bytes The easier way to pack your messages. Supply with the message number and the Payloads you want to serialize. Optionally you can choose to sign the message. :param msg_num: the message number to claim for this message :param sig: whether or not to sign this message :type sig: bool :param payloads: the list of Payload instances to serialize :return: the serialized message .. py:method:: _ez_pack(prefix: bytes, msg_num: int, payloads: collections.abc.Sequence[3.0.ipv8.messaging.serialization.Payload], sig: bool = True) -> bytes .. py:method:: _verify_signature(auth: 3.0.ipv8.messaging.payload_headers.BinMemberAuthenticationPayload, data: bytes) -> tuple[bool, bytes] .. py:method:: _ez_unpack_auth(payload_class: type[UT], data: bytes) -> tuple[3.0.ipv8.messaging.payload_headers.BinMemberAuthenticationPayload, 3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, UT] .. py:method:: _ez_unpack_noauth(payload_class: type[UT], data: bytes, global_time: bool = True) -> tuple[3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, UT] | UT .. py:exception:: PacketDecodingError Bases: :py:obj:`RuntimeError` Exception for when binary data doesn't match its expected format.