Welcome, guest | Sign In | My Account | Store | Cart
"""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.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 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)

History