Open
Description
context
summary
The Python example for basic API usage includes direct process creation and management, which could be improved by explaining error handling, resource cleanup, and signal handling more explicitly. While the example shows the basic mechanics, it doesn't sufficiently emphasize proper exception handling and resource management that would be required in production code.
# imports
import multicast
from multiprocessing import Process
import signal
import random # for random port
# Multicast group address and port
MCAST_GRP = "224.0.0.1" # Replace with your multicast group address (use IPv4 dotted notation)
MCAST_PORT = int(random.SystemRandom().randint(49152, 65535)) # Replace with your multicast port
# Options for multicast listener
listener_options = {
"is_daemon": True, # bool: enable daemon mode
"port": MCAST_PORT, # int: UDP port for multicast
"group": MCAST_GRP # str: multicast group address (use IPv4 dotted notation)
}
# Create a multicast listener
listener = multicast.hear.McastHEAR()
# Register signal handlers for clean shutdown
def handle_signal(sig, frame):
if p and p.is_alive():
print("Shutting down multicast listener...")
p.terminate()
p.join()
exit(0)
signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)
# create a separate process for the listener
p = None
try:
p = Process(
target=listener,
name="HEAR", kwargs=listener_options
)
p.daemon = listener_options["is_daemon"]
p.start()
print(f"Listening on multicast group {MCAST_GRP}, port {MCAST_PORT}")
# Wait for the process to complete or be interrupted
p.join()
except Exception as e:
print(f"Error: {e}")
finally:
# Ensure cleanup happens even if an exception occurs
if p and p.is_alive():
p.terminate()
p.join()
# Process will be terminated by signal handler when user presses Ctrl+C
Originally posted by @codecov-ai[bot] in #429 (comment)
Additional details
(This satisfies AI usage policy required human review criteria about suggestions)
- full examples like this should go in the
docs/USAGE.md
documentation- the FAQ should only mention where full examples can be found
- should refactor exception handling to focus on multicast exceptions and keyboard interruption exceptions
- should use helper function:
def example_kill_switch(target: Process) -> None:
'''helper to cleanup processes'''
if target and target.is_alive():
target.terminate()
target.join()
- need to verify if this is portable code or not
- need to check signal module license
- need new unit-test
Metadata
Metadata
Assignees
Type
Projects
Status
To do