Skip to content

Commit

Permalink
Examples code refactor
Browse files Browse the repository at this point in the history
  • Loading branch information
@ccie18643
ccie18643 committed Sep 9, 2024
1 parent 6bf931d commit 0c2cd55
Show file tree
Hide file tree
Showing 18 changed files with 254 additions and 239 deletions.
4 changes: 2 additions & 2 deletions examples/icmp_echo_client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ def start(self) -> None:

click.echo("Starting the ICMP Echo client.")
self._run_thread = True
threading.Thread(target=self.__thread__client).start()
threading.Thread(target=self._thread__client).start()
time.sleep(0.1)

def stop(self) -> None:
Expand All @@ -108,7 +108,7 @@ def stop(self) -> None:
self._run_thread = False
time.sleep(0.1)

def __thread__client(self) -> None:
def _thread__client(self) -> None:
assert self._local_ip_address is not None

flow_id = random.randint(0, 65535)
Expand Down
149 changes: 149 additions & 0 deletions examples/lib/service.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
#!/usr/bin/env python3

################################################################################
## ##
## PyTCP - Python TCP/IP stack ##
## Copyright (C) 2020-present Sebastian Majewski ##
## ##
## This program is free software: you can redistribute it and/or modify ##
## it under the terms of the GNU General Public License as published by ##
## the Free Software Foundation, either version 3 of the License, or ##
## (at your option) any later version. ##
## ##
## This program is distributed in the hope that it will be useful, ##
## but WITHOUT ANY WARRANTY; without even the implied warranty of ##
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ##
## GNU General Public License for more details. ##
## ##
## You should have received a copy of the GNU General Public License ##
## along with this program. If not, see <https://www.gnu.org/licenses/>. ##
## ##
## Author's email: [email protected] ##
## Github repository: https://github.com/ccie18643/PyTCP ##
## ##
################################################################################


"""
The 'user space' generic service base class used in examples.
examples/lib/service.py
ver 3.0.2
"""


from __future__ import annotations

import threading
import time
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING

import click

from pytcp.lib import socket
from pytcp.lib.net_addr.ip_address import IpAddress

if TYPE_CHECKING:
from pytcp.lib.socket import Socket


class Service(ABC):
"""
Generic service support class.
"""

_protocol_name: str

def __init__(
self, *, service_name: str, local_ip_address: IpAddress, local_port: int
) -> None:
"""
Initialize the Service object.
"""

self._service_name = service_name
self._local_ip_address = local_ip_address
self._local_port = local_port
self._run_thread = False

def start(self) -> None:
"""
Start the service thread.
"""

click.echo(
f"Starting the {self._protocol_name} {self._service_name} service."
)
self._run_thread = True
threading.Thread(target=self._thread__service).start()
time.sleep(0.1)

def stop(self) -> None:
"""
Stop the service thread.
"""

click.echo(
f"Stopinging the {self._protocol_name} {self._service_name} service."
)
self._run_thread = False
time.sleep(0.1)

def _get_listening_socket(self) -> Socket | None:
"""
Create and bind the listening socket.
"""

match self._local_ip_address.version, self._protocol_name:
case 6, "TCP":
listening_socket = socket.socket(
family=socket.AF_INET6, type=socket.SOCK_STREAM
)
case 4, "TCP":
listening_socket = socket.socket(
family=socket.AF_INET4, type=socket.SOCK_STREAM
)
case 6, "UDP":
listening_socket = socket.socket(
family=socket.AF_INET6, type=socket.SOCK_DGRAM
)
case 4, "UDP":
listening_socket = socket.socket(
family=socket.AF_INET4, type=socket.SOCK_DGRAM
)

try:
listening_socket.bind(
(str(self._local_ip_address), self._local_port)
)
click.echo(
f"Service {self._protocol_name} {self._service_name}: Socket created, "
f"bound to {self._local_ip_address}, port {self._local_port}."
)

except OSError as error:
click.echo(
f"Service {self._protocol_name} {self._service_name}: bind() call "
f"failed - {error!r}."
)
return None

return listening_socket

@abstractmethod
def _thread__service(self) -> None:
"""
Service thread.
"""

raise NotImplementedError

@abstractmethod
def _service(self, *, socket: Socket) -> None:
"""
Service logic handler.
"""

raise NotImplementedError
110 changes: 24 additions & 86 deletions examples/lib/tcp_service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,114 +36,52 @@
from __future__ import annotations

import threading
import time
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override

import click

from pytcp.lib import socket
from pytcp.lib.net_addr.ip_address import IpAddress
from examples.lib.service import Service

if TYPE_CHECKING:
from pytcp.lib.socket import Socket


class TcpService(ABC):
class TcpService(Service):
"""
TCP service support class.
"""

def __init__(
self, *, service_name: str, local_ip_address: IpAddress, local_port: int
) -> None:
"""
Class constructor.
"""

self._service_name = service_name
self._local_ip_address = local_ip_address
self._local_port = local_port
self._run_thread = False

def start(self) -> None:
"""
Start the service thread.
"""
_protocol_name = "TCP"

click.echo(f"Starting the TCP {self._service_name} service.")
self._run_thread = True
threading.Thread(target=self.__thread__service).start()
time.sleep(0.1)

def stop(self) -> None:
"""
Stop the service thread.
@override
def _thread__service(self) -> None:
"""

click.echo(f"Stopinging the TCP {self._service_name} service.")
self._run_thread = False
time.sleep(0.1)

def __thread__service(self) -> None:
Service thread.
"""
Service initialization.
"""

match self._local_ip_address.version:
case 6:
listening_socket = socket.socket(
family=socket.AF_INET6, type=socket.SOCK_STREAM
)
case 4:
listening_socket = socket.socket(
family=socket.AF_INET4, type=socket.SOCK_STREAM
)

try:
listening_socket.bind(
(str(self._local_ip_address), self._local_port)
)
click.echo(
f"Service TCP {self._service_name}: Socket created, bound to "
f"{self._local_ip_address}, port {self._local_port}."
)

except OSError as error:
click.echo(
f"Service TCP {self._service_name}: bind() call failed - {error!r}."
)
return

listening_socket.listen()
click.echo(
f"Service TCP {self._service_name}: Socket set to listening mode."
)

while True:
connected_socket, _ = listening_socket.accept()
if listening_socket := self._get_listening_socket():
listening_socket.listen()
click.echo(
f"Service TCP {self._service_name}: Inbound connection received from "
f"{connected_socket.remote_ip_address}, port {connected_socket.remote_port}."
f"Service {self._protocol_name} {self._service_name}: Socket "
"set to listening mode."
)
threading.Thread(
target=self.__thread__service__connection_handler,
kwargs={"connected_socket": connected_socket},
).start()
while True:
connected_socket, _ = listening_socket.accept()
click.echo(
f"Service {self._protocol_name} {self._service_name}: Inbound "
f"connection received from {connected_socket.remote_ip_address}, "
f"port {connected_socket.remote_port}."
)
threading.Thread(
target=self._thread__service__connection_handler,
kwargs={"connected_socket": connected_socket},
).start()

def __thread__service__connection_handler(
def _thread__service__connection_handler(
self, *, connected_socket: Socket
) -> None:
"""
Inbound connection handler.
"""

self._service(connected_socket=connected_socket)

@abstractmethod
def _service(self, *, connected_socket: Socket) -> None:
"""
Service method.
"""

raise NotImplementedError
self._service(socket=connected_socket)
Loading

0 comments on commit 0c2cd55

Please sign in to comment.