lustre/tests/socketclient

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: GPL-2.0
   3
   4 #
   5 # This file is part of Lustre, http://www.lustre.org/
   6 #
   7 # lustre/tests/socketclient
   8 #
   9 # Domain socket client, used to test that domain sockets
  10 # work on lustre.
  11 #
  12 # Rewrite in Python: Timothy Day <timday@amazon.com>
  13 #
  14
  15 import os
  16 import socket
  17 import sys
  18
  19
  20 # Define a function for logging messages
  21 def logmsg(msg):
  22         print(f"{sys.argv[0]} {os.getpid()}: {msg} at {os.popen('date').read().strip()}")
  23
  24
  25 # Get the socket path from the command-line argument,
  26 # or ask the user to input one
  27 if len(sys.argv) > 1:
  28         socket_path = sys.argv[1]
  29 else:
  30         socket_path = input("Enter socket path: ")
  31
  32 try:
  33         # Create a Unix domain socket
  34         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  35
  36         # Connect to the specified rendezvous point
  37         sock.connect(socket_path)
  38
  39         logmsg(f"connection on {socket_path}")
  40
  41         while True:
  42                 # Receive data from the socket in chunks of 1024 bytes
  43                 data = sock.recv(1024)
  44
  45                 # If there's no more data to read, exit the loop
  46                 if not data:
  47                         break
  48
  49                 # Decode the data as UTF-8 and print it to the standard output
  50                 print("Message:", data.decode('utf-8'), end='')
  51
  52 except socket.error as e:
  53         logmsg(f"Socket error: {e}")
  54
  55 finally:
  56         sock.close()