feat(udp): server and client skeleton refactored
- check for UDP connection via ping/pong
This commit is contained in:
parent
73d2ec7a59
commit
fac9988c7e
GPG Key ID:
E13CFA85C53F8062
139
src/myftp/client.py
Normal file
139
src/myftp/client.py
Normal file
@ -0,0 +1,139 @@
|
||||
from socket import socket, AF_INET, SOCK_DGRAM
|
||||
from typing import Tuple
|
||||
from argparse import ArgumentParser
|
||||
|
||||
# custome type to represent the hostname(server name) and the server port
|
||||
Address = Tuple[str, int]
|
||||
|
||||
|
||||
class UDPClient:
|
||||
def __init__(self, server_name: str, server_port: int, debug: bool):
|
||||
self.server_name: str = server_name
|
||||
self.server_port: int = server_port
|
||||
self.mode: str = "UDP"
|
||||
self.pong_received: bool = False
|
||||
self.debug = debug
|
||||
|
||||
def run(self):
|
||||
# server cannot be reached, stop the client immediately
|
||||
if not self.pong_received:
|
||||
return
|
||||
|
||||
client_socket = socket(AF_INET, SOCK_DGRAM)
|
||||
|
||||
client_socket.connect((self.server_name, self.server_port))
|
||||
|
||||
message = input("input lowercase sentence: ")
|
||||
|
||||
client_socket.send(message.encode())
|
||||
|
||||
try:
|
||||
modified_message = client_socket.recv(2048)
|
||||
print(modified_message.decode())
|
||||
|
||||
except ConnectionRefusedError:
|
||||
print(
|
||||
f"myftp> - {self.mode} - ConnectionRefusedError happened. Please restart the client program, make sure the server is running and/or put a different server name and server port."
|
||||
)
|
||||
finally:
|
||||
client_socket.close()
|
||||
|
||||
# ping pong UDP
|
||||
def check_udp_server(self):
|
||||
# Create a UDP socket
|
||||
client_socket = socket(AF_INET, SOCK_DGRAM)
|
||||
# will time out after 5 seconds
|
||||
client_socket.settimeout(5)
|
||||
|
||||
try:
|
||||
# Send a test message to the server
|
||||
message = b"ping"
|
||||
client_socket.sendto(message, (self.server_name, self.server_port))
|
||||
|
||||
# Receive the response
|
||||
data, _ = client_socket.recvfrom(1024)
|
||||
|
||||
# If the server responds, consider the address valid
|
||||
print(
|
||||
f"myftp> - {self.mode} - Server at {self.server_name} is valid. Response received: {data.decode('utf-8')}"
|
||||
)
|
||||
|
||||
# code reached here meaning no problem with the connection
|
||||
self.pong_received = True
|
||||
|
||||
except TimeoutError:
|
||||
# Server did not respond within the specified timeout
|
||||
print(
|
||||
f"myftp> - {self.mode} - Server at {self.server_name} did not respond within 5 seconds. Check the address or server status."
|
||||
)
|
||||
|
||||
finally:
|
||||
# Close the socket
|
||||
client_socket.close()
|
||||
|
||||
|
||||
def get_address_input() -> Address:
|
||||
while True:
|
||||
try:
|
||||
# Get input as a space-separated string
|
||||
input_string = input("myftp>Provide IP address and Port number\n")
|
||||
|
||||
# Split the input into parts
|
||||
input_parts = input_string.split()
|
||||
|
||||
# Ensure there are exactly two parts
|
||||
if len(input_parts) != 2:
|
||||
raise ValueError(
|
||||
"myftp>Invalid input. Please enter a servername/hostname/ip address as a string and the port number as an integer separated by a space."
|
||||
)
|
||||
|
||||
# Extract the values and create the tuple
|
||||
string_part, int_part = input_parts
|
||||
address = (string_part, int(int_part))
|
||||
|
||||
# Valid tuple, return it
|
||||
return address
|
||||
|
||||
except ValueError as e:
|
||||
print(
|
||||
f"Error: {e}. Invalid input. Please enter a servername/hostname/ip address as a string and the port number as an integer separated by a space."
|
||||
)
|
||||
|
||||
|
||||
def init():
|
||||
arg_parser = ArgumentParser(description="A FTP client written in Python")
|
||||
|
||||
arg_parser.add_argument(
|
||||
"--debug",
|
||||
type=int,
|
||||
choices=[0, 1],
|
||||
default=0,
|
||||
required=False,
|
||||
help="Enable or disable the flag (0 or 1)",
|
||||
)
|
||||
|
||||
args = arg_parser.parse_args()
|
||||
|
||||
while (
|
||||
protocol_selection := input("myftp>Press 1 for TCP, Press 2 for UDP\n")
|
||||
) not in {"1", "2"}:
|
||||
print("myftp>Invalid choice. Press 1 for TCP, Press 2 for UDP")
|
||||
|
||||
# UDP client selected here
|
||||
if protocol_selection == "2":
|
||||
user_supplied_address = get_address_input()
|
||||
|
||||
udp_client = UDPClient(
|
||||
user_supplied_address[0], user_supplied_address[1], args.debug
|
||||
)
|
||||
|
||||
udp_client.check_udp_server()
|
||||
|
||||
udp_client.run()
|
||||
else:
|
||||
# tcp client here
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
init()
|
||||
@ -27,8 +27,12 @@ class UDPServer:
|
||||
f"received message from client at {clientAddress}: {message_in_utf8}"
|
||||
) if self.debug else None
|
||||
|
||||
modified_message = message_in_utf8.upper()
|
||||
self.server_socket.sendto(modified_message.encode(), clientAddress)
|
||||
if message_in_utf8 == "ping":
|
||||
response_message = "pong"
|
||||
else:
|
||||
response_message = message_in_utf8.upper()
|
||||
|
||||
self.server_socket.sendto(response_message.encode(), clientAddress)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
shut_down = True
|
||||
@ -40,7 +44,7 @@ class UDPServer:
|
||||
|
||||
|
||||
def init():
|
||||
parser = ArgumentParser(description="A FTP server written in Python. UDP version")
|
||||
parser = ArgumentParser(description="A FTP server written in Python")
|
||||
|
||||
parser.add_argument(
|
||||
"port_number", type=int, help="Port number for the server. Default = 12000"
|
||||
@ -56,9 +60,20 @@ def init():
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
udp_server = UDPServer("127.0.0.1", args.port_number, args.debug)
|
||||
while (
|
||||
protocol_selection := input("myftp>Press 1 for TCP, Press 2 for UDP\n")
|
||||
) not in {"1", "2"}:
|
||||
print("myftp>Invalid choice. Press 1 for TCP, Press 2 for UDP")
|
||||
|
||||
udp_server.run()
|
||||
# UDP client selected here
|
||||
if protocol_selection == "2":
|
||||
udp_server = UDPServer("127.0.0.1", args.port_number, args.debug)
|
||||
|
||||
udp_server.run()
|
||||
|
||||
else:
|
||||
# tcp client here
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
@ -1,64 +0,0 @@
|
||||
from socket import socket, AF_INET, SOCK_DGRAM
|
||||
from argparse import ArgumentParser
|
||||
|
||||
|
||||
class UDPClient:
|
||||
def __init__(self, server_name: str, server_port: int, debug: bool):
|
||||
self.server_name = server_name
|
||||
self.server_port = server_port
|
||||
self.debug = debug
|
||||
|
||||
print(f"New UDP connection created to server at {server_name}:{server_port}")
|
||||
|
||||
def run(self):
|
||||
client_socket = socket(AF_INET, SOCK_DGRAM)
|
||||
|
||||
try:
|
||||
client_socket.connect((self.server_name, self.server_port))
|
||||
except Exception:
|
||||
print(
|
||||
f"Error with the server IP address {self.server_name} or with the server port number {self.server_port}"
|
||||
)
|
||||
|
||||
message = input("input lowercase sentence: ")
|
||||
|
||||
client_socket.send(message.encode())
|
||||
|
||||
modified_message = client_socket.recv(2048)
|
||||
|
||||
print(modified_message.decode())
|
||||
|
||||
client_socket.close()
|
||||
|
||||
|
||||
def init():
|
||||
parser = ArgumentParser(description="A FTP server written in Python. UDP version")
|
||||
|
||||
parser.add_argument(
|
||||
"server_name",
|
||||
type=str,
|
||||
default="127.0.0.1",
|
||||
help="IP address or hostname for the server. Default = localhost",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"port_number", type=int, help="Port number for the server. Default = 12000"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--debug",
|
||||
type=int,
|
||||
choices=[0, 1],
|
||||
default=0,
|
||||
help="Enable or disable the flag (0 or 1)",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
udp_server = UDPClient(args.server_name, args.port_number, args.debug)
|
||||
|
||||
udp_server.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
init()
|
||||
Loading…
x
Reference in New Issue
Block a user