client

  1"""Starts a Client to communicate with the server. For more information, please run `python client.py --help`"""
  2
  3import argparse
  4import atexit
  5import json
  6import os
  7import sys
  8import time
  9
 10import socketio
 11from contextlib import redirect_stderr
 12
 13from transport_message import TransportMessage
 14
 15
 16class Client:
 17    """
 18    Client object for publisher server
 19    """
 20
 21    def __init__(self, server_id) -> None:
 22        """Constructor, init socket and variables
 23
 24        :param server_id: server address
 25        :type server_id: string
 26        """
 27        self.socket = socketio.Client()
 28        try:
 29            self.socket.connect(server_id)
 30        except socketio.exceptions.ConnectionError:
 31            print(f"No connection to {server_id}, please make sure the server is available.")
 32            sys.exit(0)
 33
 34        self.socket.on("PRINT_MESSAGE", self._handleResponse)
 35        self.socket.on("PRINT_MESSAGE_AND_EXIT", self._handleExitResponse)
 36
 37        self.subscribed_topics = []
 38
 39    def disconnect(self) -> None:
 40        """
 41        Disconnect socket
 42        """
 43        self.socket.disconnect()
 44
 45    def subscribe(self, topics) -> None:
 46        """
 47        Request to subscribe to topics, wait for server messages
 48
 49        :param topics: topics
 50        :type topics: string or list of strings
 51        """
 52        self.subscribed_topics = topics
 53        # string to list
 54        if not isinstance(topics, list):
 55            self.subscribed_topics = [topics]
 56
 57        print(f"======= SUBSCRIBED TO {', '.join(self.subscribed_topics)} =======\n")
 58
 59        for topic in self.subscribed_topics:
 60            tMessage = TransportMessage(timestamp=time.time(), topic=topic)
 61            self.socket.emit("SUBSCRIBE_TOPIC", tMessage.json())
 62
 63    def unsubscibe(self) -> None:
 64        """
 65        Request to unsubscribe from topics in self.subscribed_topics and disconnect
 66        """
 67        print(f"\n======= UNSUBSCRIBED FROM {', '.join(self.subscribed_topics)} =======")
 68        for topic in self.subscribed_topics:
 69            tMessage = TransportMessage(timestamp=time.time(), topic=topic)
 70            self.socket.emit("UNSUBSCRIBE_TOPIC", tMessage.json())
 71
 72        # Disconnect
 73        self.socket.disconnect()
 74
 75    def publish(self, topic, message):
 76        """
 77        Request to publish a message to the topic
 78
 79        :param topic: topic
 80        :type topic: string
 81        :param message: message for topic
 82        :type message: string
 83        """
 84        print(f"======= PUBLISH MESSAGE =======")
 85        print(f"Message: {message}")
 86        tMessage = TransportMessage(timestamp=time.time(), topic=topic, payload=message)
 87        self.socket.emit("PUBLISH_TOPIC", tMessage.json())
 88
 89    def listTopics(self):
 90        """
 91        Request to list all topics avaliable
 92        """
 93        tMessage = TransportMessage(timestamp=time.time())
 94        self.socket.emit("LIST_TOPICS", tMessage.json())
 95
 96    def getTopicStatus(self, topic):
 97        """
 98        Request to get topic status
 99
100        :param topic: topic
101        :type topic: string
102        """
103        tMessage = TransportMessage(timestamp=time.time(), topic=topic)
104        self.socket.emit("GET_TOPIC_STATUS", tMessage.json())
105
106    def _handleResponse(self, response):
107        """
108        Receive PRINT_MESSAGE response from server
109
110        :param response: response from server
111        :type response: string
112        """
113        response_dict = json.loads(response)
114        print(f"{response_dict['payload']}")
115
116    def _handleExitResponse(self, response):
117        """
118        Receive PRINT_MESSAGE_AND_EXIT response from server
119
120        :param response: response from server
121        :type response: string
122        """
123        self._handleResponse(response)
124
125        # Exit
126        self.disconnect()
127        sys.exit(0)
128
129
130if __name__ == "__main__":
131    # init parser
132    parser = argparse.ArgumentParser(prog="Client", description="Client for Publisher")
133
134    parser.add_argument("-s", "--server", required=True, help="server address as String", metavar="ADDRESS:PORT")
135    parser.add_argument("-sub", "--subscribe", nargs="+", help="list of topics to subscribe as Strings", metavar="STRING")
136    parser.add_argument("-p", "--publish", help="published topic as String", metavar="STRING")
137    parser.add_argument("-m", "--message", help="message to be published to topic as String", metavar="STRING")
138    parser.add_argument("-st", "--status", help="get topic status from server", metavar="STRING")
139    parser.add_argument("-l", "--list", action="store_true", help="get all list topics from server")
140
141    args = parser.parse_args()
142
143    # init client
144    cli = Client(args.server)
145
146    # workaround for
147    # KeyboardInterrupt: "Exception ignored in: <module 'threading' from '/usr/lib/python3.10/threading.py'>"
148    # To see error messages, comment those lines
149    fnull = open(os.devnull, "w")
150    r = redirect_stderr(fnull)
151    r.__enter__()
152
153    # call client functions
154    if args.subscribe:
155        cli.subscribe(args.subscribe)
156        atexit.register(cli.unsubscibe)
157    elif args.publish and args.message:
158        cli.publish(args.publish, args.message)
159    elif args.list:
160        cli.listTopics()
161    elif args.status:
162        cli.getTopicStatus(args.status)
163    else:
164        print("No action, please check your parameters")