Python code coverage for Tools/demo/mcast.py

#	count	content
1	n/a	#!/usr/bin/env python3
2	n/a
3	n/a	"""
4	n/a	Send/receive UDP multicast packets.
5	n/a	Requires that your OS kernel supports IP multicast.
6	n/a
7	n/a	Usage:
8	n/a	mcast -s (sender, IPv4)
9	n/a	mcast -s -6 (sender, IPv6)
10	n/a	mcast (receivers, IPv4)
11	n/a	mcast -6 (receivers, IPv6)
12	n/a	"""
13	n/a
14	n/a	MYPORT = 8123
15	n/a	MYGROUP_4 = '225.0.0.250'
16	n/a	MYGROUP_6 = 'ff15:7079:7468:6f6e:6465:6d6f:6d63:6173'
17	n/a	MYTTL = 1 # Increase to reach other networks
18	n/a
19	n/a	import time
20	n/a	import struct
21	n/a	import socket
22	n/a	import sys
23	n/a
24	n/a	def main():
25	n/a	group = MYGROUP_6 if "-6" in sys.argv[1:] else MYGROUP_4
26	n/a
27	n/a	if "-s" in sys.argv[1:]:
28	n/a	sender(group)
29	n/a	else:
30	n/a	receiver(group)
31	n/a
32	n/a
33	n/a	def sender(group):
34	n/a	addrinfo = socket.getaddrinfo(group, None)[0]
35	n/a
36	n/a	s = socket.socket(addrinfo[0], socket.SOCK_DGRAM)
37	n/a
38	n/a	# Set Time-to-live (optional)
39	n/a	ttl_bin = struct.pack('@i', MYTTL)
40	n/a	if addrinfo[0] == socket.AF_INET: # IPv4
41	n/a	s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl_bin)
42	n/a	else:
43	n/a	s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, ttl_bin)
44	n/a
45	n/a	while True:
46	n/a	data = repr(time.time()).encode('utf-8') + b'\0'
47	n/a	s.sendto(data, (addrinfo[4][0], MYPORT))
48	n/a	time.sleep(1)
49	n/a
50	n/a
51	n/a	def receiver(group):
52	n/a	# Look up multicast group address in name server and find out IP version
53	n/a	addrinfo = socket.getaddrinfo(group, None)[0]
54	n/a
55	n/a	# Create a socket
56	n/a	s = socket.socket(addrinfo[0], socket.SOCK_DGRAM)
57	n/a
58	n/a	# Allow multiple copies of this program on one machine
59	n/a	# (not strictly needed)
60	n/a	s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
61	n/a
62	n/a	# Bind it to the port
63	n/a	s.bind(('', MYPORT))
64	n/a
65	n/a	group_bin = socket.inet_pton(addrinfo[0], addrinfo[4][0])
66	n/a	# Join group
67	n/a	if addrinfo[0] == socket.AF_INET: # IPv4
68	n/a	mreq = group_bin + struct.pack('=I', socket.INADDR_ANY)
69	n/a	s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
70	n/a	else:
71	n/a	mreq = group_bin + struct.pack('@I', 0)
72	n/a	s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq)
73	n/a
74	n/a	# Loop, printing any data we receive
75	n/a	while True:
76	n/a	data, sender = s.recvfrom(1500)
77	n/a	while data[-1:] == '\0': data = data[:-1] # Strip trailing \0's
78	n/a	print(str(sender) + ' ' + repr(data))
79	n/a
80	n/a
81	n/a	if __name__ == '__main__':
82	n/a	main()