Connect to socket and handle user input
(args, path)
| 103 | |
| 104 | |
| 105 | def handle_socket(args, path): |
| 106 | """ Connect to socket and handle user input """ |
| 107 | prompt = '' # this evaluates to false in conditions |
| 108 | sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) |
| 109 | global CMDS |
| 110 | |
| 111 | if os.isatty(sys.stdin.fileno()): |
| 112 | prompt = '--> ' |
| 113 | print("Connecting to " + path) |
| 114 | try: |
| 115 | sock.connect(path) |
| 116 | except OSError: |
| 117 | print("Error connecting to " + path) |
| 118 | sock.close() |
| 119 | # if socket exists but is bad, or if non-interactive just return |
| 120 | if os.path.exists(path) or not prompt: |
| 121 | return |
| 122 | # if user didn't give a valid socket path, but there are |
| 123 | # some sockets, help the user out by printing how to connect |
| 124 | socks = find_sockets(os.path.dirname(path)) |
| 125 | if socks: |
| 126 | print("\nOther DPDK telemetry sockets found:") |
| 127 | print_socket_options(args.file_prefix, socks) |
| 128 | else: |
| 129 | list_fp() |
| 130 | return |
| 131 | json_reply = read_socket(sock, 1024, prompt, prompt) |
| 132 | output_buf_len = json_reply["max_output_len"] |
| 133 | app_name = get_app_name(json_reply["pid"]) |
| 134 | if app_name and prompt: |
| 135 | print('Connected to application: "%s"' % app_name) |
| 136 | |
| 137 | # get list of commands for readline completion |
| 138 | sock.send("/".encode()) |
| 139 | CMDS = read_socket(sock, output_buf_len, False)["/"] |
| 140 | |
| 141 | # interactive prompt |
| 142 | try: |
| 143 | text = input(prompt).strip() |
| 144 | while text != "quit": |
| 145 | if text.startswith('/'): |
| 146 | sock.send(text.encode()) |
| 147 | read_socket(sock, output_buf_len, pretty=prompt) |
| 148 | text = input(prompt).strip() |
| 149 | except EOFError: |
| 150 | pass |
| 151 | finally: |
| 152 | sock.close() |
| 153 | |
| 154 | |
| 155 | def readline_complete(text, state): |
no test coverage detected