"""Module for simple UDP broadcast support.
The classes in this module are stepping stones for building discoverable
services on a network. Server replies are to be handled by the importer."""
################################################################################
__author__ = 'Stephen "Zero" Chappell <Noctis.Skytower@gmail.com>'
__date__ = '18 November 2011'
__version__ = '$Revision: 1 $'
################################################################################
import socket
import _thread
import time
################################################################################
class Client:
"Client(port) -> Client instance"
__slots__ = '__socket', '__address'
def __init__(self, port):
"Initialize the client with a sending socket."
self.__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, True)
self.__address = '255.255.255.255', port
def connect(self, data):
"Send a broadcast through the underlying socket."
self.__socket.sendto(data, self.__address)
################################################################################
class Server:
"Server(port) -> Server instance"
__slots__ = '__socket'
def __init__(self, port):
"Initialize the server with a receiving socket."
self.__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
self.__socket.bind(('0.0.0.0', port))
def accept(self, size):
"Receive a broadcast through the underlying socket."
return self.__socket.recvfrom(size)
################################################################################
def test(port):
"Test the client and server broadcasting classes."
_thread.start_new_thread(test_client, (port,))
test_server(port)
def test_client(port):
"Test the client broadcasting class."
c = Client(port)
while True:
c.connect(time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()).encode())
time.sleep(1)
def test_server(port):
"Test the server broadcasting class."
s = Server(port)
while True:
data, address = s.accept(1 << 10)
print('From: {}\n{}\n'.format(address, data.decode()))
################################################################################
if __name__ == '__main__':
test(50000)
Diff to Previous Revision
--- revision 1 2011-11-19 05:48:25
+++ revision 2 2011-11-19 05:50:53
@@ -45,7 +45,6 @@
"Initialize the server with a receiving socket."
self.__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
- self.__socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, True)
self.__socket.bind(('0.0.0.0', port))
def accept(self, size):