Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if first_size:
RangeFetch.first_size = first_size
if max_size:
RangeFetch.max_size = max_size
if threads:
RangeFetch.threads = threads
if down_rate:
RangeFetch.down_rate_min = int(down_rate * 2)
RangeFetch.down_rate_max = RangeFetch.down_rate_min + min(max(down_rate, 1024 * 100), 1024 * 200)
if proxy:
RangeFetch.proxy = proxy
if headers:
RangeFetch.headers.update(headers)
new_server = LocalTCPServer((bind, port), RangeFetchHandler)
RangeFetch.bufsize = RangeFetchHandler.bufsize
thread.start_new_thread(new_server.serve_forever, ())
return new_server
def spawn_later(seconds, target, *args, **kwargs):
def wrap(*args, **kwargs):
sleep(seconds)
target(*args, **kwargs)
thread.start_new_thread(wrap, args, kwargs)