Module msgs.

Msgs

The Msgs class is intended to be used during testing to send and receive messages between threads.

Example:

send a message to remote thread

>>> import threading
>>> from scottbrian_utils.msgs import Msgs
>>> def f1(msgs) -> None:
...     print('f1 beta entered')
...     my_msg = msgs.get_msg('beta')
...     print(my_msg)
...     print('f1 beta exiting')
>>> def example1() -> None:
...     print('mainline entered')
...     msgs = Msgs()
...     f1_thread = threading.Thread(target=f1, args=(msgs,))
...     f1_thread.start()
...     msgs.queue_msg('beta', 'hello beta')
...     f1_thread.join()
...     print('mainline exiting')
>>> example1()
mainline entered
f1 beta entered
hello beta
f1 beta exiting
mainline exiting
Example:

a command loop using Msgs

>>> import threading
>>> from scottbrian_utils.msgs import Msgs
>>> import time
>>> def main() -> None:
...     def f1() -> None:
...         print('f1 beta entered')
...         while True:
...             my_msg = msgs.get_msg('beta')
...             print(f'beta received msg: {my_msg}')
...             if my_msg == 'exit':
...                 break
...             else:
...                 # handle message
...                 msgs.queue_msg('alpha', f'msg "{my_msg}" completed')
...         print('f1 beta exiting')
...     print('mainline entered')
...     msgs = Msgs()
...     f1_thread = threading.Thread(target=f1)
...     f1_thread.start()
...     msgs.queue_msg('beta', 'do command a')
...     print(f"alpha received response: {msgs.get_msg('alpha')}")
...     msgs.queue_msg('beta', 'do command b')
...     print(f"alpha received response: {msgs.get_msg('alpha')}")
...     msgs.queue_msg('beta', 'exit')
...     f1_thread.join()
...     print('mainline exiting')
>>> main()
mainline entered
f1 beta entered
beta received msg: do command a
alpha received response: msg "do command a" completed
beta received msg: do command b
alpha received response: msg "do command b" completed
beta received msg: exit
f1 beta exiting
mainline exiting

The msgs module contains:

  1. Msgs class with methods:

    1. get_msg

    2. queue_msg

class msgs.Msgs

Msgs class for testing.

The Msgs class is used to assist in the testing of multi-threaded functions. It provides a set of methods that help with test case coordination and verification. The test case setup involves a mainline thread that starts one or more remote threads. The queue_msg and get_msg methods are used for inter-thread communications.

Initialize the object.

get_msg(recipient, timeout=3.0)

Get the next message in the queue.

Parameters:

  • recipient (str) – arbitrary name that designates the target of the message and which will be used with the queue_msg method to identify the intended recipient of the message

  • timeout (Union[int, float, None]) – number of seconds allowed for msg response. A negative value, zero, or None means no timeout will happen. If timeout is not specified, then the default timeout value will be used.

Return type:

Any

Returns:

the received message

Raises:

GetMsgTimedOut – {recipient} timed out waiting for msg

queue_msg(target, msg='go')

Place a msg on the msg queue for the specified target.

Parameters:

  • target (str) – arbitrary name that designates the target of the message and which will be used with the get_msg method to retrieve the message

  • msg (Optional[Any]) – message to place on queue

Return type:

None