`multicast.recv`

Provides multicast RECV Features.

Use for receiving multicast messages. Contains classes and functions to handle receiving messages from multicast groups.

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

`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.

API

multicast.recv.__package__[source]

‘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__[source]

‘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__[source]

‘multicast/recv.py’

The file of this component.

multicast.recv.__name__[source]

‘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.joinstep(groups, port, iface=None, bind_group=None, isock=None)[source][source]

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.

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

multicast.recv.tryrecv(msgbuffer, chunk, sock)[source][source]

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 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.

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, chunk, sock)[source][source]

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[source][source]

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__[source]: ‘multicast.recv’

__name__[source]: ‘multicast.recv.McastRECV’

__proc__[source]: ‘RECV’

__epilogue__ = <Multiline-String>[source]

__prologue__[source]: ‘Python Multicast Receiver. Primitives for a listener for multicast data.’

classmethod setupArgs(parser)[source][source]

static _hearstep(groups, port, iface=None, bind_group=None)[source][source]

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)[source][source]

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.