llvm-project/lldb/scripts/swig_bot_lib/server.py

#!/usr/bin/env python

"""
SWIG generation server.  Listens for connections from swig generation clients
and runs swig in the requested fashion, sending back the results.
"""

# Future imports
from __future__ import absolute_import
from __future__ import print_function

# Python modules
import argparse
import logging
import os
import socket
import struct
import sys
import traceback

# LLDB modules
import use_lldb_suite
from lldbsuite.support import sockutil

# package imports
from . import local

default_port = 8537

def process_args(args):
    # Setup the parser arguments that are accepted.
    parser = argparse.ArgumentParser(description='SWIG generation server.')

    parser.add_argument(
        "--port",
        action="store",
        default=default_port,
        help=("The local port to bind to"))

    # Process args.
    return parser.parse_args(args)

def initialize_listening_socket(options):
    logging.debug("Creating socket...")
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    logging.info("Binding to ip address '', port {}".format(options.port))
    s.bind(('', options.port))

    logging.debug("Putting socket in listen mode...")
    s.listen()
    return s

def accept_once(sock, options):
    logging.debug("Waiting for connection...")
    client, addr = sock.accept()
    logging.info("Received connection from {}".format(addr))
    data_size = struct.unpack("!I", sockutil.recvall(client, 4))[0]
    logging.debug("Expecting {} bytes of data from client".format(data_size))
    data = sockutil.recvall(client, data_size)
    logging.info("Received {} bytes of data from client".format(len(data)))

    logging.info("Sending {} byte response".format(len(data)))
    client.sendall(struct.pack("!I", len(data)))
    client.sendall(data)

def accept_loop(sock, options):
    while True:
        try:
            accept_once(sock, options)
        except Exception as e:
            error = traceback.format_exc()
            logging.error("An error occurred while processing the connection.")
            logging.error(error)

def run(args):
    options = process_args(args)
    sock = initialize_listening_socket(options)
    accept_loop(sock, options)
    return options