core.run_cycle() - Code Metrics - Inspection of "dispatching client and adding ui from archive" - Sigmanificient/PyTchat - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( e5d7db...d12e93 )

by Yohann

created 2021-07-28 20:02 UTC

core.run_cycle() B

↳ Parent: core

Complexity

Conditions

Size

Total Lines	53
Code Lines	34

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	34
dl	0
loc	53
rs	7.1973
c	0
b	0
f	0
cc	8
nop	3

How to fix Long Method

import select
import socket

HEADER_LENGTH = 10


def receive_message(client_socket_):
    try:
        message_header = client_socket_.recv(HEADER_LENGTH)
        if not len(message_header):
            return False

        message_length = int(message_header.decode('utf-8').strip())
        return {
            "header": message_header,
            "data": client_socket_.recv(message_length)
        }

    except Exception as e:
        print(e)
        return False


def start(ip, port):
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    server_socket.bind((ip, port))
    server_socket.listen()

    print("server started")
    socket_list = [server_socket]
    clients = {}

    while True:
        run_cycle(server_socket, socket_list, clients)


def run_cycle(server_socket, socket_list, clients):
    read_sockets, _, exception_sockets = select.select(
        socket_list, [], socket_list
    )

    for notified_socket in read_sockets:
        if notified_socket == server_socket:
            client_socket, client_address = server_socket.accept()

            user = receive_message(client_socket)

            if not user:
                continue

            socket_list.append(client_socket)
            clients[client_socket] = user

            print(
                f"Accepted new connection from {client_address[0]}:"
                f"{client_address[1]} username: {user['data'].decode('utf-8')}"
            )

        else:
            message = receive_message(notified_socket)

            if not message:
                print(
                    "Closed connection from",
                    clients[notified_socket]['data'].decode('utf-8')
                )

                socket_list.remove(notified_socket)
                del clients[notified_socket]
                continue

            user = clients[notified_socket]
            print(
                f"receive_message from {user['data'].decode('utf-8')}:",
                message['data'].decode('utf-8')
            )

            for client_socket in clients:
                if client_socket != notified_socket:
                    client_socket.send(
                        (
                            user["header"] + user["data"]
                            + message['header'] + message["data"]
                        )
                    )

    for notified_socket in exception_sockets:
        socket_list.remove(notified_socket)
        del clients[notified_socket]


1			import select
2			import socket
3
4			HEADER_LENGTH = 10
5
6
7			def receive_message(client_socket_):
8			try:
9			message_header = client_socket_.recv(HEADER_LENGTH)
10			if not len(message_header):
11			return False
12
13			message_length = int(message_header.decode('utf-8').strip())
14			return {
15			"header": message_header,
16			"data": client_socket_.recv(message_length)
17			}
18
19			except Exception as e:
20			print(e)
21			return False
22
23
24			def start(ip, port):
25			server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
26			server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
27
28			server_socket.bind((ip, port))
29			server_socket.listen()
30
31			print("server started")
32			socket_list = [server_socket]
33			clients = {}
34
35			while True:
36			run_cycle(server_socket, socket_list, clients)
37
38
39			def run_cycle(server_socket, socket_list, clients):
40			read_sockets, _, exception_sockets = select.select(
41			socket_list, [], socket_list
42			)
43
44			for notified_socket in read_sockets:
45			if notified_socket == server_socket:
46			client_socket, client_address = server_socket.accept()
47
48			user = receive_message(client_socket)
49
50			if not user:
51			continue
52
53			socket_list.append(client_socket)
54			clients[client_socket] = user
55
56			print(
57			f"Accepted new connection from {client_address[0]}:"
58			f"{client_address[1]} username: {user['data'].decode('utf-8')}"
59			)
60
61			else:
62			message = receive_message(notified_socket)
63
64			if not message:
65			print(
66			"Closed connection from",
67			clients[notified_socket]['data'].decode('utf-8')
68			)
69
70			socket_list.remove(notified_socket)
71			del clients[notified_socket]
72			continue
73
74			user = clients[notified_socket]
75			print(
76			f"receive_message from {user['data'].decode('utf-8')}:",
77			message['data'].decode('utf-8')
78			)
79
80			for client_socket in clients:
81			if client_socket != notified_socket:
82			client_socket.send(
83			(
84			user["header"] + user["data"]
85			+ message['header'] + message["data"]
86			)
87			)
88
89			for notified_socket in exception_sockets:
90			socket_list.remove(notified_socket)
91			del clients[notified_socket]
92

Sigmanificient / PyTchat

Push — main ( e5d7db...d12e93 )

core.run_cycle() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like