Source code for proxy.core.work.task.local

# -*- coding: utf-8 -*-
"""
    proxy.py
    ~~~~~~~~
    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
    Network monitoring, controls & Application development, testing, debugging.

    :copyright: (c) 2013-present by Abhinav Singh and contributors.
    :license: BSD, see LICENSE for more details.
"""
import time
import uuid
import threading
from typing import Any

from ..local import BaseLocalExecutor
from ....common.backports import NonBlockingQueue


[docs]class LocalTaskExecutor(BaseLocalExecutor): """We'll define a local executor which is capable of receiving log lines over a non blocking queue."""
[docs] def work(self, *args: Any) -> None: task_id = int(time.time()) uid = '%s-%s' % (self.iid, task_id) self.works[task_id] = self.create(uid, *args)
[docs]class ThreadedTaskExecutor(threading.Thread): def __init__(self, **kwargs: Any) -> None: super().__init__() self.daemon = True self.executor = LocalTaskExecutor( iid=uuid.uuid4().hex, work_queue=NonBlockingQueue(), **kwargs, ) def __enter__(self) -> 'ThreadedTaskExecutor': self.start() return self def __exit__(self, *args: Any) -> None: self.executor.running.set() self.join()
[docs] def run(self) -> None: self.executor.run()