sockets

 avatar
unknown
python
2 years ago
2.7 kB
14
Indexable
### server.py
import socket
from config import Config
import threading


class Server:
    def __init__(self, host, port, number_of_connections=500) -> None:
        self.host = host
        self.port = port
        self.address = (host, port)
        self.number_of_connections = number_of_connections

        socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        self.socket_server = socket_server

    def handle_client(self, connection, host):
        print('Connected by', connection, host)
        assert connection is not None and isinstance(
            connection, socket.socket), 'Invalid connection'

        while True:
            try:
                connection.send(b'Hello, client!')

                # receive data from the connected client
                data = connection.recv(1024)

                # print the received data
                print(data.decode())
            except Exception as e:
                print('handle_client Exception', e)
                connection.close()
                break

    def close(self):
        self.socket_server.close()

    def listen(self):
        self.socket_server.listen(self.number_of_connections)

    def bind(self):
        self.socket_server.bind(self.address)


def main():
    running = True

    server = Server(
        Config.server.get('host'),
        Config.server.getint('port')
    )

    server.bind()
    server.listen()

    while running:
        connection = None
        try:
            connection, address = server.socket_server.accept()

            thread = threading.Thread(
                target=server.handle_client,
                args=(connection, address)
            )

            thread.start()
        except KeyboardInterrupt:
            print('keyboard interrupt detected, exiting...')

            if connection:
                connection.close()
            if server:
                server.close()

            threading.Event().set()
            thread.join()
            running = False
            break


if __name__ == "__main__":
    main()
    

### client.py
from server import Server
from config import Config


class Client:
    def __init__(self, host, port) -> None:
        self.host = host
        self.port = port
        self.address = (host, port)


def main():
    server = Server(
        Config.server.get('host'),
        Config.server.getint('port')
    )

    running = True

    while running:
        server.socket_server.connect(server.address)

        data = server.socket_server.recv(1024)

        # print the received data
        print(data.decode())


if __name__ == "__main__":
    main()
Editor is loading...