Source code for galaxy.util.sockets

import random
import shlex
import socket
import subprocess

[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 = shlex.split("netstat tuln") p1 = subprocess.Popen(cmd_netstat, stdout=subprocess.PIPE) occupied_ports = set() for line in'\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