ยปCore Development>Code coverage>Tools/demo/mcast.py

Python code coverage for Tools/demo/mcast.py

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