Source code for galaxy.util.sockets

import random
import socket

from galaxy.util import commands


[docs]def get_ip():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        # doesn't even have to be reachable
        s.connect(('10.255.255.255', 1))
        ip = s.getsockname()[0]
    except Exception:
        ip = None
    finally:
        s.close()
    return ip


[docs]def unused_port(range=None):
    if range:
        return __unused_port_on_range(range)
    else:
        return __unused_port_rangeless()


def __unused_port_rangeless():
    # TODO: Allow ranges (though then need to guess and check)...
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('localhost', 0))
    addr, port = s.getsockname()
    s.close()
    return port


def __unused_port_on_range(range):
    assert range[0] and range[1]

    # Find all ports that are already occupied
    cmd_netstat = ["netstat", "tuln"]
    stdout = commands.execute(cmd_netstat)

    occupied_ports = set()
    for line in stdout.split('\n'):
        if line.startswith('tcp') or line.startswith('tcp6'):
            col = line.split()
            local_address = col[3]
            local_port = local_address.split(':')[1]
            occupied_ports.add(int(local_port))

    # Generate random free port number.
    while True:
        port = random.randrange(range[0], range[1])
        if port not in occupied_ports:
            break
    return port