__author__ = 'Alek Ratzloff <[email protected]>; Chris Campell <[email protected]>'
import signal
import socket
import sys
from typing import Optional
from loguru import logger
[docs]
class Client:
def __init__(self, host_address: Optional[str] = "127.0.0.1", host_listening_port: Optional[int] = 65000,
receive_size_bytes: Optional[int] = 1024):
"""
.. todo:: Docstrings.
Args:
host_address:
host_listening_port:
receive_size_bytes:
"""
self._host_address: str = host_address
self._host_listening_port: int = host_listening_port
self._receive_size_bytes: int = receive_size_bytes
self._socket: Optional[socket.socket] = None
signal.signal(signal.SIGINT, self.on_sigint)
signal.signal(signal.SIGTERM, self.on_sigterm)
[docs]
def connect(self):
"""
Establishes a connection from the client to the server, commands will be sent via this channel.
Raises:
ConnectionError: Raises a `ConnectionError` in the event that the client fails to connect to the server.
"""
self._socket = socket.socket()
try:
self._socket.connect((self.host_address, self.host_listening_port))
except Exception as err:
self._socket = None
raise ConnectionError(f"Client {__name__} failed to connect to server: {self._host_address} "
f"on port: {self._host_listening_port}.")
[docs]
def close(self):
"""
Closes a connection from the client to a server.
Raises:
ConnectionError: Raises a `ConnectionError` in the event that the client is not connected to the server.
"""
if not self._socket:
raise ConnectionError(f"Client: {__name__} is not connected to the server.")
self._socket.close()
self._socket = None
[docs]
def send(self, message: str):
"""
Sends a message from the client to the server.
.. note::
This method requires that a pre-existing connection between the client and server has already been
established. In order to establish a connection, call the `connect` method.
Args:
message: .. todo:: Docstring.
Raises:
ConnectionError: Raises a `ConnectionError` in the event that the client is not yet connected to the server.
"""
if not self._socket:
raise ConnectionError(f"Client: {__name__} is not connected to the server.")
# Convert the message to bytes:
encoded_message: bytes = message.encode(encoding='ascii')
self._socket.send(encoded_message)
[docs]
def wait_receive(self) -> bytes:
"""
Blocks program execution until a message is received from the server.
Returns:
bytes: The raw bytes received from the server.
Raises:
ConnectionError: Raises a `ConnectionError` in the event that the client is not connected to the server.
"""
if not self._socket:
raise ConnectionError(f"Client: {__name__} is not connected to the server.")
response: bytes = self._socket.recv(self._receive_size_bytes)
return response
[docs]
def send_receive(self, message: str) -> bytes:
"""
Creates a connection to the server, sends the message, waits on a response, closes the connection, and returns
the server's response.
Args:
message: .. todo:: Docstring.
Returns:
bytes: The raw binary response received from the server.
Raises:
ConnectionError: Raises a `ConnectionError` in the event that the client cannot connect to the server.
"""
if self._socket is None:
self.connect()
self.send(message=message)
response: bytes = self.wait_receive()
self.close()
return response
@staticmethod
def on_sigterm(*args, **kwargs):
logger.info('SIGTERM received... exiting.')
exit(0)
@staticmethod
def on_sigint(*args, **kwargs):
logger.info('SIGINT received... exiting')
exit(0)
@property
def host_address(self) -> str:
return self._host_address
@property
def host_listening_port(self) -> int:
return self._host_listening_port
@property
def receive_size_bytes(self) -> int:
return self._receive_size_bytes
def main():
__version: str = '3.0.0-beta'
print('-------------------------------------------')
print('-------~ Beemon ~-------')
print('-------~ Type help for commands ~-------')
print('-------------------------------------------')
print('(version: %s)' % __version)
beemon_client = Client(host_address='127.0.0.1', host_listening_port=65000, receive_size_bytes=1024)
while True:
try:
arg = input('-> ')
except EOFError:
# sending EOF is the same as saying to quit
arg = 'quit'
print(arg)
if not bool(arg.strip()):
continue
if arg == 'quit':
sys.exit()
try:
response = beemon_client.send_receive(arg)
except ConnectionError as err:
response = "Could not make a connection to the server\n"
response += "reason: %s" % err
# logger.error(response)
if response == 'Bad command':
print('Bad command type "help" for list of commands')
#used for the 'help' command to allow formatting
elif type(response) == bytes:
print(response.decode('utf-8'))
else:
print(response)
if __name__ == '__main__':
__version: str = '3.0.0-beta'
print('-------------------------------------------')
print('-------~ Beemon ~-------')
print('-------~ Type help for commands ~-------')
print('-------------------------------------------')
print('(version: %s)' % __version)
beemon_client = Client(host_address='127.0.0.1', host_listening_port=65000, receive_size_bytes=1024)
while True:
try:
arg = input('-> ')
except EOFError:
# sending EOF is the same as saying to quit
arg = 'quit'
print(arg)
if not bool(arg.strip()):
continue
if arg == 'quit':
sys.exit()
try:
response = beemon_client.send_receive(arg)
except ConnectionError as err:
response = "Could not make a connection to the server\n"
response += "reason: %s" % err
# logger.error(response)
if response == 'Bad command':
print('Bad command type "help" for list of commands')
else:
print(response)