multicast.recv

Multicast RECV functionality implementation.

This module provides functionality for receiving multicast messages. It contains classes and functions to handle receiving messages from multicast groups, with support for single-message reception modes.

Functions: joinstep: Configure socket for joining multicast groups. tryrecv: Attempt to receive data on a socket. recvstep: Receive messages continuously until interrupted.

Classes: McastRECV: Main tool class for RECV operations.

Caution: See details regarding dynamic imports documented in this module.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

    Testcase 0: Multicast should be importable.

            >>> import multicast
            >>> multicast.recv is not None
            True
            >>> multicast.__doc__ is not None
            True
            >>>

    Testcase 1: Recv should be automatically imported.
            A: Test that the multicast component is initialized.
            B: Test that the recv component is initialized.
            C: Test that the recv component has __doc__

            >>> multicast is not None
            True
            >>> multicast.recv is not None
            True
            >>> multicast.recv.__doc__ is not None
            True
            >>> type(multicast.recv.__doc__) == type(str(''''''))
            True
            >>>

    Testcase 2: Recv should be detailed with some metadata.
            A: Test that the __MAGIC__ variables are initialized.
            B: Test that the __MAGIC__ variables are strings.

            >>> multicast.recv is not None
            True
            >>> multicast.recv.__module__ is not None
            True
            >>> multicast.recv.__package__ is not None
            True
            >>> type(multicast.recv.__doc__) == type(multicast.recv.__module__)
            True
            >>>

    Testcase 3: main should return an int.
            A: Test that the multicast component is initialized.
            B: Test that the recv component is initialized.
            C: Test that the main(RECV) function-flow is initialized.
            D: Test that the main(RECV) function-flow returns an int 0-255.

            >>> multicast.__main__ is not None
            True
            >>> multicast.__main__.main is not None
            True
            >>> tst_fxtr_args = ['''RECV''', '''--port=1234''']
            >>> (test_fixture, ignored_value) = multicast.__main__.main(tst_fxtr_args)
            >>> test_fixture is not None
            True
            >>> type(test_fixture) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <...int...>
            >>> int(test_fixture) >= int(0)
            True
            >>> type(test_fixture) is type(0)
            True
            >>> int(test_fixture) < int(256)
            True
            >>> (int(test_fixture) >= int(0)) and (int(test_fixture) < int(256))
            True
            >>>

Module Contents

Classes

McastRECV

Subclasses the multicast.mtool to provide the RECV functions.

Functions

_validate_join_args	Validates joinstep arguments.
joinstep	Join multicast groups to prepare for receiving messages.
tryrecv	Attempt to receive data on the given socket and decode it into the message buffer.
recvstep	Receive messages continuously until interrupted.

Data

__package__	The package of this program.
__module__	The module of this program.
__file__	The file of this component.
__name__	The name of this component.
module_logger
_w_prefix
_w_example_code
_w_advice
_w_empty_join_warning
_w_unspec_bind
_w_non_multicast

API

multicast.recv.__package__

‘multicast’

The package of this program.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: Recv should be automatically imported.

    >>> multicast.recv.__package__ is not None
    True
    >>>
    >>> multicast.recv.__package__ == multicast.__package__
    True
    >>>

multicast.recv.__module__

‘multicast’

The module of this program.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: Recv should be automatically imported.

    >>> multicast.recv.__module__ is not None
    True
    >>>

multicast.recv.__file__

‘multicast/recv.py’

The file of this component.

multicast.recv.__name__

‘multicast.recv’

The name of this component.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

Testcase 0: Multicast should be importable.

    >>> import multicast
    >>>

Testcase 1: Recv should be automatically imported.

    >>> multicast.recv.__name__ is not None
    True
    >>>

multicast.recv.module_logger

‘getLogger(…)’

multicast.recv._w_prefix: str

‘Unusual call to multicast.joinstep with no groups.’

multicast.recv._w_example_code: str = <Multiline-String>

multicast.recv._w_advice: str

‘join(…)’

multicast.recv._w_empty_join_warning: str

‘join(…)’

multicast.recv._w_unspec_bind: str

‘join(…)’

multicast.recv._w_non_multicast

None

multicast.recv._validate_join_args(groups=None, port=None, iface=None, bind_group=None, isock=None) → tuple

Validates joinstep arguments.

This is a helper function and should NOT be called directly.

Args: groups (list): List of multicast group addresses to join. port (int): Port number to bind the socket to. iface (str, optional): Network interface to use. bind_group (str, optional): Specific group address to bind to. isock (socket.socket, optional): Existing socket to configure.

Note: All warning messages are only emitted when debug is True (i.e., when Python is not running with -O or -OO).

Returns: A tuple of (groups, port, iface, bind_group, isock) after normalization.

multicast.recv.joinstep(groups, port, iface=None, bind_group=None, isock=None) → multicast.socket.socket

Join multicast groups to prepare for receiving messages.

Configures the socket to join specified multicast groups on a given port.

The JOIN function. Will start to listen on the given port of an interface for multicast messages to the given group(s).

Args: groups (list): List of multicast group addresses to join. port (int): Port number to bind the socket to. iface (str, optional): Network interface to use. bind_group (str, optional): Specific group address to bind to. isock (socket.socket, optional): Existing socket to configure.

Returns: socket.socket: Configured socket ready to receive multicast messages.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

    >>> import multicast
    >>> multicast.recv is not None
    True
    >>>

Testcase 1: Stability testing. A: Verify the multicast.recv module is properly initialized. B: Verify the joinstep function exists and has the expected type. C: Test socket creation with no groups (default behavior). D: Test socket creation with a specified multicast group. E: Test socket creation with a multicast group and binding to that group. F: Test socket creation using an existing socket handle.

    >>> import multicast
    >>>
    >>> multicast.recv is None
    False
    >>>
    >>> multicast.recv.joinstep is None
    False
    >>> type(multicast.recv.joinstep)
    <class 'function'>
    >>> tst_sk = multicast.recv.joinstep(None, 59991)
    >>> tst_sk #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket...laddr=...>
    >>> multicast.endSocket(tst_sk)
    >>> tst_sk #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket [closed]...>
    >>> tst_fxtr = multicast._MCAST_DEFAULT_GROUP
    >>> tst_sk_2 = multicast.recv.joinstep([tst_fxtr], 59991)
    >>> tst_sk_2 #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket...>
    >>> multicast.endSocket(tst_sk_2)
    >>> tst_sk_2 #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket [closed]...>
    >>> tst_sk_3 = multicast.recv.joinstep(
    ...             [tst_fxtr], 59991, None, tst_fxtr
    ... ) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    >>> tst_sk_3 #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket...>
    >>> multicast.endSocket(tst_sk_3)
    >>> tst_sk_3 #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket [closed]...>
    >>> sk_fxtr = multicast.genSocket()
    >>> tst_sk_4 = multicast.recv.joinstep(
    ...             [tst_fxtr], 59991, None, tst_fxtr, sk_fxtr
    ... )
    >>> tst_sk_4 #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket...>
    >>> multicast.endSocket(tst_sk_4)
    >>> tst_sk_4 #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <socket.socket [closed]...>
    >>> sk_fxtr.close()
    >>>

multicast.recv.tryrecv(msgbuffer: list, chunk: bytes, sock: multicast.socket.socket) → str

Attempt to receive data on the given socket and decode it into the message buffer.

Will try to listen on the given socket directly into the given chunk for decoding. If the read into the chunk results in content, the chunk will be decoded and appended to the caller-instantiated msgbuffer, which is a collection of utf8 strings (or None). After decoding, chunk is zeroed for memory efficiency and security. Either way the message buffer will be returned.

Tries to receive data without blocking and appends it to the message buffer.

Individual chunk sizes are controlled by the module attribute _MCAST_DEFAULT_BUFFER_SIZE set at module’s load-time. It is possible to override the buffer size via the environment variable “MULTICAST_BUFFER_SIZE” if available at load-time. However changing the value is not recommended unless absolutely needed, and can be done on the sender side too.

Args: msgbuffer (list or None): Caller-instantiated collection to store received messages. chunk (variable or None): Caller-instantiated variable for raw received data. sock (socket.socket): The socket to receive data from.

Returns: list: The message buffer possibly updated with any newly received data.

Minimal Acceptance Testing:

    First set up test fixtures by importing multicast.

            >>> import multicast
            >>> multicast.recv is not None
            True
            >>> multicast.recv.tryrecv is not None
            True
            >>>

    Testcase 1: Stability testing.

            >>> import multicast
            >>>
            >>> multicast.recv is None
            False
            >>> multicast.recv.tryrecv is None
            False
            >>> type(multicast.recv.tryrecv) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <class 'function'>
            >>> sk_fxtr = multicast.genSocket()
            >>> tst_args = ("test pass-through", None, sk_fxtr)
            >>> multicast.recv.recvstep(*tst_args) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            'test pass-through'
            >>> sk_fxtr.close()
            >>>

    Testcase 2: Mock overflow testing.

            >>> import multicast
            >>>
            >>> multicast.recv is None
            False
            >>> multicast.recv.tryrecv is None
            False
            >>> type(multicast.recv.tryrecv) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            <class 'function'>
            >>> class mockSocket():
            ...     def recv(self, *args, **kwargs):
            ...         return b'it worked'
            ...
            ...     def close(self):
            ...         pass
            ...
            ...     def shutdown(self, *args, **kwargs):
            ...         pass
            ...
            >>>
            >>> sk_fxtr = mockSocket()
            >>> tst_args = ("test added: ", None, sk_fxtr)
            >>> multicast.recv.recvstep(*tst_args) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
            'test added: it worked'
            >>> sk_fxtr.close()
            >>>

multicast.recv.recvstep(msgbuffer: list, chunk: bytes, sock: multicast.socket.socket) → str

Receive messages continuously until interrupted.

Listens on the socket and accumulates messages into the buffer.

Args: msgbuffer (list): Buffer to store received messages. chunk (int): Maximum number of bytes to read per message. sock (socket.socket): The socket to receive data from.

Returns: list: Updated message buffer with received messages.

class multicast.recv.McastRECV

Bases: multicast.multicast.mtool

Subclasses the multicast.mtool to provide the RECV functions.

    Testing:

    Testcase 0: First set up test fixtures by importing multicast.

            >>> import multicast
            >>> multicast.recv is not None
            True
            >>> multicast._MCAST_DEFAULT_PORT is not None
            True
            >>> multicast._MCAST_DEFAULT_GROUP is not None
            True
            >>> multicast._MCAST_DEFAULT_TTL is not None
            True
            >>> multicast.recv.McastRECV is not None
            True
            >>>

    Testcase 2: Recv should be detailed with some metadata.
            A: Test that the __MAGIC__ variables are initialized.
            B: Test that the __MAGIC__ variables are strings.

            >>> multicast.recv is not None
            True
            >>> multicast.recv.McastRECV is not None
            True
            >>> multicast.recv.McastRECV.__module__ is not None
            True
            >>> multicast.recv.McastRECV.__proc__ is not None
            True
            >>> multicast.recv.McastRECV.__epilogue__ is not None
            True
            >>> multicast.recv.McastRECV.__prologue__ is not None
            True
            >>>

__module__

‘multicast.recv’

__name__

‘multicast.recv.McastRECV’

__proc__

‘RECV’

__epilogue__ = <Multiline-String>

__prologue__

‘Python Multicast Receiver. Primitives for a listener for multicast data.’

classmethod setupArgs(parser)

static _hearstep(groups, port, iface=None, bind_group=None)

Will listen on the given port of an interface for multicast messages to the given group(s).

The work-horse function.

Internal method to set up receiving multicast messages.

Args: groups (list): Multicast groups to join. port (int): Port number for receiving messages. iface (str, optional): Network interface to use. bind_group (str, optional): Specific group address to bind to.

Returns: str: Any received message buffer as a string. May be empty. Raises: NotImplementedError: if joining the multicast group is unsupported on the current system.

Minimal Acceptance Testing:

First set up test fixtures by importing multicast.

    >>> import multicast
    >>> multicast.recv is not None
    True
    >>> multicast.recv.McastRECV is not None
    True
    >>>

Testcase 1: Stability testing.

    >>> import multicast
    >>>
    >>> multicast.recv is None
    False
    >>> multicast.recv.McastRECV is None
    False
    >>> test_RCEV = multicast.recv.McastRECV()
    >>> type(test_RCEV) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    <class ...McastRECV...>
    >>> type(test_RCEV._hearstep)
    <class 'function'>
    >>> test_RCEV._hearstep(None, 59991) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    '...'
    >>> tst_fxtr = multicast._MCAST_DEFAULT_GROUP
    >>> test_RCEV._hearstep([tst_fxtr], 59991) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    '...'
    >>> test_RCEV._hearstep(
    ...     [tst_fxtr], 59991, None, tst_fxtr
    ... ) #doctest: -DONT_ACCEPT_BLANKLINE, +ELLIPSIS
    '...'
    >>>

doStep(*args, **kwargs)

Execute the RECV operation to receive multicast messages.

Overrides the doStep method from mtool to start receiving messages based on provided arguments.

Args: *args: Variable length argument list containing command-line arguments. **kwargs: Arbitrary keyword arguments.

Returns: tuple: A tuple containing received data and a status indicator.