(host='::1', maddr=maddr)
| 49 | maddr = ('ff12::4242', 4242) |
| 50 | |
| 51 | def ipv6Multicast(host='::1', maddr=maddr): |
| 52 | haddr = socket.getaddrinfo(host, maddr[1], socket.AF_INET6, socket.SOCK_DGRAM)[0][-1] |
| 53 | maddr = socket.getaddrinfo(maddr[0], maddr[1], socket.AF_INET6, socket.SOCK_DGRAM)[0][-1] |
| 54 | |
| 55 | sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) |
| 56 | |
| 57 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 58 | if hasattr(socket, "SO_REUSEPORT"): |
| 59 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) |
| 60 | |
| 61 | sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_LOOP, 1) |
| 62 | sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 5) |
| 63 | |
| 64 | ifn = haddr[3] or if_nametoindex('lo0') |
| 65 | ifn = struct.pack("I", ifn) |
| 66 | sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, ifn) |
| 67 | |
| 68 | group = socket.inet_pton(socket.AF_INET6, maddr[0]) + ifn |
| 69 | sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, group) |
| 70 | |
| 71 | sock.bind(haddr) |
| 72 | sock.setblocking(False) |
| 73 | |
| 74 | return sock, maddr |
| 75 | |
| 76 | #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ |
| 77 | #~ Main |
no test coverage detected