Run the server.
()
| 75 | |
| 76 | |
| 77 | def main() -> int: |
| 78 | """Run the server.""" |
| 79 | args = parse_args() |
| 80 | |
| 81 | # Configure a listening socket on args.address and args.port. |
| 82 | with get_listening_socket(args.address, args.port) as listening_sock: |
| 83 | print(f"Listening on {args.address}:{args.port}") |
| 84 | |
| 85 | read_bytes: bytes = b"" |
| 86 | while len(read_bytes) == 0: |
| 87 | with accept_connection(listening_sock) as sock: |
| 88 | read_bytes = wait_for_headers_complete(sock) |
| 89 | if len(read_bytes) == 0: |
| 90 | # This is probably the PortOpenv4 test. Try again. |
| 91 | print("No bytes read on this connection. Trying again.") |
| 92 | sock.close() |
| 93 | continue |
| 94 | |
| 95 | # Send a response now, before the body is read. This implements |
| 96 | # the "quick" attribute of this quick_server. |
| 97 | send_response(sock, args.abort_response_headers) |
| 98 | |
| 99 | if args.abort_response_headers: |
| 100 | # We're done. |
| 101 | break |
| 102 | |
| 103 | if args.drain_request: |
| 104 | num_bytes_to_drain = determine_outstanding_bytes_to_read(read_bytes) |
| 105 | print(f'Read {len(read_bytes)} bytes. ' |
| 106 | f'Draining {num_bytes_to_drain} bytes.') |
| 107 | drain_socket(sock, read_bytes, num_bytes_to_drain) |
| 108 | else: |
| 109 | print(f'Read {len(read_bytes)} bytes. ' |
| 110 | f'Not draining per configuration.') |
| 111 | return 0 |
| 112 | |
| 113 | |
| 114 | if __name__ == "__main__": |
no test coverage detected