Repository: abhinavsingh/proxy.py
Pull request #62 (2019-08-20 04:17:14)

Proxy Plugins
Targeted towards feature requests in #28 and #30 .  First draft of Proxy plugins.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 277a11b..3cc21e5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,28 +1,21 @@
 language: python
-python:
-  # Disable 2.7 since proxy.py will not support this in future
-  # - "2.7"
-  # Disabled due to test failures like:
-  # TypeError: unsupported operand type(s) for %: 'bytes' and 'bytes'
-  # Needs explicit conversion to bytes after string formatting.
-  # Example, see https://travis-ci.org/abhinavsingh/proxy.py/jobs/490855017
-  #- "3.3"
-  #- "3.4"
-  - "3.5"
-  - "3.6"
-# Enable 3.7 without globally enabling sudo and dist: xenial for other build jobs
-# This is official workaround for Python 3.7 support by Travis
-# See https://github.com/travis-ci/travis-ci/issues/9815
+env:
+  - TESTING_ON_TRAVIS=1
 matrix:
   include:
-    - python: 3.7
-      dist: xenial
-      sudo: true
-os:
-  - linux
-  # No support for OSX Python runtime
-  #- osx
-  # Windows Python support is currently in early access
-  #- windows
-script:
-  - make
+    - name: "Python 3.7 on Xenial Linux"
+      python: 3.7
+    - name: "Python 3.7 on macOS"
+      os: osx
+      osx_image: xcode11
+      language: shell
+    - name: "Python 3.7 on Windows"
+      os: windows
+      language: shell
+      before_install:
+        - choco install python
+        - python -m pip install --upgrade pip
+      env:
+        - PATH=/c/Python37:/c/Python37/Scripts:$PATH
+        - TESTING_ON_TRAVIS=1
+script: python3 tests.py || python tests.py
diff --git a/Dockerfile b/Dockerfile
index 9cb64e3..ef08291 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,10 +1,8 @@
 FROM python:3-alpine
 LABEL com.abhinavsingh.name="abhinavsingh/proxy.py" \
       com.abhinavsingh.description="Lightweight HTTP, HTTPS, WebSockets Proxy Server in a single Python file" \
-#      com.abhinavsingh.build-date="" \
       com.abhinavsingh.url="https://github.com/abhinavsingh/proxy.py" \
       com.abhinavsingh.vcs-url="https://github.com/abhinavsingh/proxy.py" \
-#      com.abhinavsingh.vcs-ref="" \
       com.abhinavsingh.docker.cmd="docker run -it --rm -p 8899:8899 abhinavsingh/proxy.py"
 
 COPY proxy.py /app/
@@ -13,9 +11,4 @@ EXPOSE 8899/tcp
 WORKDIR /app
 ENTRYPOINT [ "./proxy.py" ]
 CMD [ "--host=0.0.0.0", \
-      "--port=8899", \
-      "--backlog=100", \
-      "--server-recvbuf-size=8192", \
-      "--client-recvbuf-size=8192", \
-      "--open-file-limit=1024", \
-      "--log-level=INFO" ]
+      "--ipv4" ]
diff --git a/README.md b/README.md
index 52fe1f0..6e1efcf 100644
--- a/README.md
+++ b/README.md
@@ -6,9 +6,11 @@ Features
 --------
 
 - Distributed as a single file module
+- Optionally enable builtin Web Server
+- Customize proxy and http routing via plugins
 - No external dependency other than standard Python library
 - Support for `http`, `https`, `http2` and `websockets` request proxy
-- Optimize for large file uploads and downloads
+- Optimized for large file uploads and downloads
 - IPv4 and IPv6 support
 - Basic authentication support
 - Can serve a [PAC (Proxy Auto-configuration)](https://en.wikipedia.org/wiki/Proxy_auto-config) file
@@ -31,15 +33,18 @@ Usage
 $ proxy.py -h
 usage: proxy.py [-h] [--backlog BACKLOG] [--basic-auth BASIC_AUTH]
                 [--client-recvbuf-size CLIENT_RECVBUF_SIZE]
-                [--hostname HOSTNAME] [--ipv4 IPV4] [--log-level LOG_LEVEL]
-                [--open-file-limit OPEN_FILE_LIMIT] [--port PORT]
-                [--pac-file PAC_FILE]
-                [--server-recvbuf-size SERVER_RECVBUF_SIZE]
+                [--hostname HOSTNAME] [--ipv4]
+                [--enable-http-proxy ENABLE_HTTP_PROXY]
+                [--enable-web-server ENABLE_WEB_SERVER]
+                [--log-level LOG_LEVEL] [--log-format LOG_FORMAT]
+                [--open-file-limit OPEN_FILE_LIMIT] [--pac-file PAC_FILE]
+                [--pac-file-url-path PAC_FILE_URL_PATH] [--plugins PLUGINS]
+                [--port PORT] [--server-recvbuf-size SERVER_RECVBUF_SIZE]
                 [--num-workers NUM_WORKERS]
 
 proxy.py v0.4
 
-optional arguments ordered alphabetically:
+optional arguments:
   -h, --help            show this help message and exit
   --backlog BACKLOG     Default: 100. Maximum number of pending connections to
                         proxy server
@@ -52,19 +57,30 @@ optional arguments ordered alphabetically:
                         value for faster uploads at the expense of increased
                         RAM.
   --hostname HOSTNAME   Default: 127.0.0.1. Server IP address.
-  --ipv4 IPV4           Whether to listen on IPv4 address. By default server
+  --ipv4                Whether to listen on IPv4 address. By default server
                         only listens on IPv6.
+  --enable-http-proxy ENABLE_HTTP_PROXY
+                        Default: True. Whether to enable
+                        proxy.HttpProxyPlugin.
+  --enable-web-server ENABLE_WEB_SERVER
+                        Default: False. Whether to enable
+                        proxy.HttpWebServerPlugin.
   --log-level LOG_LEVEL
                         Valid options: DEBUG, INFO (default), WARNING, ERROR,
                         CRITICAL. Both upper and lowercase values are
                         allowed.You may also simply use the leading character
                         e.g. --log-level d
+  --log-format LOG_FORMAT
+                        Log format for Python logger.
   --open-file-limit OPEN_FILE_LIMIT
                         Default: 1024. Maximum number of files (TCP
                         connections) that proxy.py can open concurrently.
-  --port PORT           Default: 8899. Server port.
   --pac-file PAC_FILE   A file (Proxy Auto Configuration) or string to serve
                         when the server receives a direct file request.
+  --pac-file-url-path PAC_FILE_URL_PATH
+                        Web server path to serve the PAC file.
+  --plugins PLUGINS     Comma separated plugins
+  --port PORT           Default: 8899. Server port.
   --server-recvbuf-size SERVER_RECVBUF_SIZE
                         Default: 8 KB. Maximum amount of data received from
                         the server in a single recv() operation. Bump this
diff --git a/plugin_examples.py b/plugin_examples.py
new file mode 100644
index 0000000..de22a3d
--- /dev/null
+++ b/plugin_examples.py
@@ -0,0 +1,38 @@
+from urllib import parse as urlparse
+from proxy import HttpProtocolBasePlugin, HttpRequestRejected
+
+
+class RedirectToCustomServerPlugin(HttpProtocolBasePlugin):
+    """Modifies client request to redirect all incoming requests to a fixed server address."""
+
+    def __init__(self, config, client, request):
+        super(RedirectToCustomServerPlugin, self).__init__(config, client, request)
+
+    def on_request_complete(self):
+        if self.request.method != 'CONNECT':
+            self.request.url = urlparse.urlsplit(b'http://localhost:9999')
+
+
+class FilterByTargetDomainPlugin(HttpProtocolBasePlugin):
+    """Only accepts specific requests dropping all other requests."""
+
+    def __init__(self, config, client, request):
+        super(FilterByTargetDomainPlugin, self).__init__(config, client, request)
+        self.allowed_domains = [b'google.com', b'www.google.com', b'google.com:443', b'www.google.com:443']
+
+    def on_request_complete(self):
+        # TODO: Refactor internals to cleanup mess below, due to how urlparse works, hostname/path attributes
+        # are not consistent between CONNECT and non-CONNECT requests.
+        if (self.request.method != b'CONNECT' and self.request.url.hostname not in self.allowed_domains) or \
+                (self.request.method == b'CONNECT' and self.request.url.path not in self.allowed_domains):
+            raise HttpRequestRejected(status_code=418, body='I\'m a tea pot')
+
+
+class SaveHttpResponses(HttpProtocolBasePlugin):
+    """Saves Http Responses locally on disk."""
+
+    def __init__(self, config, client, request):
+        super(SaveHttpResponses, self).__init__(config, client, request)
+
+    def handle_response_chunk(self, chunk):
+        return chunk
diff --git a/proxy.py b/proxy.py
index 2afd450..78bfbb7 100755
--- a/proxy.py
+++ b/proxy.py
@@ -4,7 +4,7 @@
     proxy.py
     ~~~~~~~~
 
-    HTTP Proxy Server in Python.
+    Lightweight Programmable HTTP, HTTPS, WebSockets Proxy Server in a single Python file.
 
     :copyright: (c) 2013-2020 by Abhinav Singh.
     :license: BSD, see LICENSE for more details.
@@ -13,21 +13,26 @@ import argparse
 import base64
 import datetime
 import errno
+import importlib
 import logging
 import multiprocessing
 import os
+import queue
 import select
 import socket
 import sys
 import threading
 from collections import namedtuple
+from typing import Dict, List, Tuple
+from urllib import parse as urlparse
 
 if os.name != 'nt':
     import resource
 
+PY3 = sys.version_info[0] == 3
 VERSION = (0, 4)
 __version__ = '.'.join(map(str, VERSION[0:2]))
-__description__ = 'Lightweight HTTP, HTTPS, WebSockets Proxy Server in a single Python file'
+__description__ = 'Lightweight Programmable HTTP, HTTPS, WebSockets Proxy Server in a single Python file'
 __author__ = 'Abhinav Singh'
 __author_email__ = 'mailsforabhinav@gmail.com'
 __homepage__ = 'https://github.com/abhinavsingh/proxy.py'
@@ -36,87 +41,267 @@ __license__ = 'BSD'
 
 logger = logging.getLogger(__name__)
 
-PY3 = sys.version_info[0] == 3
-
-if PY3:  # pragma: no cover
-    text_type = str
-    binary_type = bytes
-    from urllib import parse as urlparse
-    import queue
-else:  # pragma: no cover
-    text_type = unicode
-    binary_type = str
-    import urlparse
-    import Queue as queue
-
 # Defaults
 DEFAULT_BACKLOG = 100
 DEFAULT_BASIC_AUTH = None
-DEFAULT_BUFFER_SIZE = 8192
+DEFAULT_BUFFER_SIZE = 1024 * 1024
 DEFAULT_CLIENT_RECVBUF_SIZE = DEFAULT_BUFFER_SIZE
 DEFAULT_SERVER_RECVBUF_SIZE = DEFAULT_BUFFER_SIZE
-DEFAULT_HOSTNAME = '127.0.0.1'
+DEFAULT_IPV4_HOSTNAME = '127.0.0.1'
+DEFAULT_IPV6_HOSTNAME = '::'
 DEFAULT_PORT = 8899
 DEFAULT_IPV4 = False
+DEFAULT_ENABLE_HTTP_PROXY = True
+DEFAULT_ENABLE_WEB_SERVER = False
 DEFAULT_LOG_LEVEL = 'INFO'
 DEFAULT_OPEN_FILE_LIMIT = 1024
 DEFAULT_PAC_FILE = None
+DEFAULT_PAC_FILE_URL_PATH = '/'
 DEFAULT_NUM_WORKERS = 0
+DEFAULT_PLUGINS = {}
+DEFAULT_LOG_FORMAT = '%(asctime)s - %(levelname)s - pid:%(process)d - %(funcName)s:%(lineno)d - %(message)s'
+
+# Set to True if under test
+UNDER_TEST = False
 
 
-def text_(s, encoding='utf-8', errors='strict'):  # pragma: no cover
+def text_(s, encoding='utf-8', errors='strict') -> str:
     """Utility to ensure text-like usability.
 
     If ``s`` is an instance of ``binary_type``, return
     ``s.decode(encoding, errors)``, otherwise return ``s``"""
-    if isinstance(s, binary_type):
+    if isinstance(s, bytes):
         return s.decode(encoding, errors)
     return s
 
 
-def bytes_(s, encoding='utf-8', errors='strict'):  # pragma: no cover
+def bytes_(s, encoding='utf-8', errors='strict') -> bytes:
     """Utility to ensure binary-like usability.
 
     If ``s`` is an instance of ``text_type``, return
     ``s.encode(encoding, errors)``, otherwise return ``s``"""
-    if isinstance(s, text_type):
+    if isinstance(s, str):
         return s.encode(encoding, errors)
     return s
 
 
 version = bytes_(__version__)
-CRLF, COLON, SP = b'\r\n', b':', b' '
+CRLF, COLON, WHITESPACE = b'\r\n', b':', b' '
 PROXY_AGENT_HEADER = b'Proxy-agent: proxy.py v' + version
 
-PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT = CRLF.join([
-    b'HTTP/1.1 200 Connection established',
-    PROXY_AGENT_HEADER,
-    CRLF
-])
-
-BAD_GATEWAY_RESPONSE_PKT = CRLF.join([
-    b'HTTP/1.1 502 Bad Gateway',
-    PROXY_AGENT_HEADER,
-    b'Content-Length: 11',
-    b'Connection: close',
-    CRLF
-]) + b'Bad Gateway'
-
-PROXY_AUTHENTICATION_REQUIRED_RESPONSE_PKT = CRLF.join([
-    b'HTTP/1.1 407 Proxy Authentication Required',
-    PROXY_AGENT_HEADER,
-    b'Content-Length: 29',
-    b'Connection: close',
-    b'Proxy-Authenticate: Basic',
-    CRLF
-]) + b'Proxy Authentication Required'
-
-PAC_FILE_RESPONSE_PREFIX = CRLF.join([
-    b'HTTP/1.1 200 OK',
-    b'Content-Type: application/x-ns-proxy-autoconfig',
-    b'Connection: close',
-    CRLF
-])
+
+class TcpConnection(object):
+    """TCP server/client connection abstraction."""
+
+    types = namedtuple('TcpConnectionTypes', (
+        'SERVER',
+        'CLIENT',
+    ))(1, 2)
+
+    def __init__(self, what: types):
+        # Cannot for socket.socket type because initialized value needs to be None?
+        self.conn = None
+        self.buffer: bytes = b''
+        self.closed: bool = False
+        self.what: TcpConnection.types = what
+
+    def send(self, data: bytes) -> int:
+        """Users must handle BrokenPipeError exceptions"""
+        return self.conn.send(data)
+
+    def recv(self, buffer_size: int = DEFAULT_BUFFER_SIZE) -> bytes:
+        try:
+            data = self.conn.recv(buffer_size)
+            if len(data) > 0:
+                logger.debug('received %d bytes from %s' % (len(data), self.what))
+                return data
+        except Exception as e:
+            if e.errno == errno.ECONNRESET:
+                logger.debug('%r' % e)
+            else:
+                logger.exception(
+                    'Exception while receiving from connection %s %r with reason %r' % (self.what, self.conn, e))
+
+    def close(self) -> bool:
+        self.conn.close()
+        self.closed = True
+        return self.closed
+
+    def buffer_size(self) -> int:
+        return len(self.buffer)
+
+    def has_buffer(self) -> bool:
+        return self.buffer_size() > 0
+
+    def queue(self, data) -> int:
+        self.buffer += data
+        return len(data)
+
+    def flush(self) -> int:
+        sent = self.send(self.buffer)
+        self.buffer = self.buffer[sent:]
+        logger.debug('flushed %d bytes to %s' % (sent, self.what))
+        return sent
+
+
+class TcpServerConnection(TcpConnection):
+    """Establishes connection to destination server."""
+
+    def __init__(self, host: str, port: int):
+        super(TcpServerConnection, self).__init__(b'server')
+        self.addr: Tuple[str, int] = (host, int(port))
+
+    def __del__(self):
+        if self.conn:
+            self.close()
+
+    def connect(self):
+        self.conn = socket.create_connection((self.addr[0], self.addr[1]))
+
+
+class TcpClientConnection(TcpConnection):
+    """Accepted client connection."""
+
+    def __init__(self, conn: socket.socket, addr: Tuple[str, int]):
+        super(TcpClientConnection, self).__init__(b'client')
+        self.conn: socket.socket = conn
+        self.addr: Tuple[str, int] = addr
+
+
+class TcpServer(object):
+    """TcpServer server implementation.
+
+    Inheritor MUST implement `handle` method. It accepts an instance of `TcpClientConnection`.
+    Optionally, can also implement `setup` and `shutdown` methods for custom bootstrapping and tearing
+    down internal state.
+    """
+
+    def __init__(self, hostname=DEFAULT_IPV4_HOSTNAME, port=DEFAULT_PORT, backlog=DEFAULT_BACKLOG, ipv4=DEFAULT_IPV4):
+        self.hostname: str = hostname
+        self.port: int = port
+        self.backlog: int = backlog
+        self.ipv4: bool = ipv4
+        self.socket: socket.socket = None
+        self.running: bool = False
+
+    def setup(self):
+        pass
+
+    def handle(self, client: TcpClientConnection):
+        raise NotImplementedError()
+
+    def shutdown(self):
+        pass
+
+    def stop(self):
+        self.running = False
+
+    def run(self):
+        self.running = True
+        self.setup()
+        try:
+            self.socket = socket.socket(socket.AF_INET if self.ipv4 is True else socket.AF_INET6, socket.SOCK_STREAM)
+            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            self.socket.bind((self.hostname, self.port))
+            self.socket.listen(self.backlog)
+            logger.info('Started server on %s:%d' % (self.hostname, self.port))
+            while self.running:
+                r, w, x = select.select([self.socket], [], [], 1)
+                if self.socket in r:
+                    conn, addr = self.socket.accept()
+                    client = TcpClientConnection(conn, addr)
+                    self.handle(client)
+        except Exception as e:
+            logger.exception('Exception while running the server %r' % e)
+        finally:
+            self.shutdown()
+            logger.info('Closing server socket')
+            self.socket.close()
+
+
+class MultiCoreRequestDispatcher(TcpServer):
+    """MultiCoreRequestDispatcher.
+
+    Pre-spawns worker process to utilize all cores available on the system.  Accepted `TcpClientConnection` is
+    dispatched over a queue to workers.  One of the worker picks up the work and starts a new thread to handle the
+    client request.
+    """
+
+    def __init__(self, hostname=DEFAULT_IPV4_HOSTNAME, port=DEFAULT_PORT, backlog=DEFAULT_BACKLOG,
+                 num_workers=DEFAULT_NUM_WORKERS, ipv4=DEFAULT_IPV4, config=None):
+        super(MultiCoreRequestDispatcher, self).__init__(hostname, port, backlog, ipv4)
+
+        self.num_workers: int = multiprocessing.cpu_count()
+        if num_workers > 0:
+            self.num_workers = num_workers
+        self.workers: List[Worker] = []
+        self.work_queues: List[multiprocessing.Queue] = []
+        self.current_worker_id = 0
+
+        self.config: HttpProtocolConfig = config
+
+    def setup(self):
+        logger.info('Starting %d workers' % self.num_workers)
+        for worker_id in range(self.num_workers):
+            work_queue = multiprocessing.Queue()
+
+            worker = Worker(work_queue, self.config)
+            worker.daemon = True
+            worker.start()
+
+            self.workers.append(worker)
+            self.work_queues.append(work_queue)
+
+    def handle(self, client: TcpClientConnection):
+        # Dispatch in round robin fashion
+        work_queue = self.work_queues[self.current_worker_id]
+        logging.debug('Dispatched client request to worker id %d', self.current_worker_id)
+        self.current_worker_id += 1
+        self.current_worker_id %= self.num_workers
+        work_queue.put((Worker.operations.HTTP_PROTOCOL, client))
+
+    def shutdown(self):
+        logger.info('Shutting down %d workers' % self.num_workers)
+        for work_queue in self.work_queues:
+            work_queue.put((Worker.operations.SHUTDOWN, None))
+        for worker in self.workers:
+            worker.join()
+
+
+class Worker(multiprocessing.Process):
+    """Generic worker class implementation.
+
+    Worker instance accepts (operation, payload) over work queue and
+    depending upon requested operation starts a new thread to handle the work.
+    """
+
+    operations = namedtuple('WorkerOperations', (
+        'HTTP_PROTOCOL',
+        'SHUTDOWN',
+    ))(1, 2)
+
+    def __init__(self, work_queue, config=None):
+        super(Worker, self).__init__()
+        self.work_queue: multiprocessing.Queue = work_queue
+        self.config: HttpProtocolConfig = config
+
+    def run(self):
+        while True:
+            try:
+                op, payload = self.work_queue.get(True, 1)
+                if op == Worker.operations.HTTP_PROTOCOL:
+                    proxy = HttpProtocolHandler(payload, config=self.config)
+                    proxy.setDaemon(True)
+                    proxy.start()
+                elif op == Worker.operations.SHUTDOWN:
+                    break
+            except queue.Empty:
+                pass
+            # Safeguard against https://gist.github.com/abhinavsingh/b8d4266ff4f38b6057f9c50075e8cd75
+            except ConnectionRefusedError:
+                pass
+            except KeyboardInterrupt:
+                break
 
 
 class ChunkParser(object):
@@ -130,35 +315,35 @@ class ChunkParser(object):
 
     def __init__(self):
         self.state = ChunkParser.states.WAITING_FOR_SIZE
-        self.body = b''  # Parsed chunks
-        self.chunk = b''  # Partial chunk received
-        self.size = None  # Expected size of next following chunk
+        self.body: bytes = b''  # Parsed chunks
+        self.chunk: bytes = b''  # Partial chunk received
+        self.size: int = None  # Expected size of next following chunk
 
-    def parse(self, data):
-        more = True if len(data) > 0 else False
+    def parse(self, raw: bytes):
+        more = True if len(raw) > 0 else False
         while more:
-            more, data = self.process(data)
+            more, raw = self.process(raw)
 
-    def process(self, data):
+    def process(self, raw: bytes):
         if self.state == ChunkParser.states.WAITING_FOR_SIZE:
             # Consume prior chunk in buffer
             # in case chunk size without CRLF was received
-            data = self.chunk + data
+            raw = self.chunk + raw
             self.chunk = b''
             # Extract following chunk data size
-            line, data = HttpParser.split(data)
+            line, raw = HttpParser.split(raw)
             if not line:  # CRLF not received
-                self.chunk = data
-                data = b''
+                self.chunk = raw
+                raw = b''
             else:
                 self.size = int(line, 16)
                 self.state = ChunkParser.states.WAITING_FOR_DATA
         elif self.state == ChunkParser.states.WAITING_FOR_DATA:
             remaining = self.size - len(self.chunk)
-            self.chunk += data[:remaining]
-            data = data[remaining:]
+            self.chunk += raw[:remaining]
+            raw = raw[remaining:]
             if len(self.chunk) == self.size:
-                data = data[len(CRLF):]
+                raw = raw[len(CRLF):]
                 self.body += self.chunk
                 if self.size == 0:
                     self.state = ChunkParser.states.COMPLETE
@@ -166,7 +351,7 @@ class ChunkParser(object):
                     self.state = ChunkParser.states.WAITING_FOR_SIZE
                 self.chunk = b''
                 self.size = None
-        return len(data) > 0, data
+        return len(raw) > 0, raw
 
 
 class HttpParser(object):
@@ -187,13 +372,14 @@ class HttpParser(object):
 
     def __init__(self, parser_type):
         assert parser_type in (HttpParser.types.REQUEST_PARSER, HttpParser.types.RESPONSE_PARSER)
-        self.type = parser_type
-        self.state = HttpParser.states.INITIALIZED
+        self.type: HttpParser.types = parser_type
+        self.state: HttpParser.states = HttpParser.states.INITIALIZED
 
-        self.raw = b''
-        self.buffer = b''
+        self.bytes: bytes = b''
+        self.buffer: bytes = b''
 
-        self.headers = dict()
+        self.headers: Dict = dict()
+        # Can simply be b'', then set type as bytes?
         self.body = None
 
         self.method = None
@@ -204,22 +390,38 @@ class HttpParser(object):
 
         self.chunk_parser = None
 
+        # This cleans up developer APIs as Python urlparse.urlsplit behaves differently
+        # for incoming proxy request and incoming web request.  Web request is the one
+        # which is broken.
+        self.host = None
+        self.port = None
+
+    def set_host_port(self):
+        if self.type == HttpParser.types.REQUEST_PARSER:
+            if self.method == b'CONNECT':
+                self.host, self.port = self.url.path.split(COLON)
+            elif self.url:
+                self.host, self.port = self.url.hostname, self.url.port \
+                    if self.url.port else 80
+            else:
+                raise Exception('Invalid request\n%s' % self.bytes)
+
     def is_chunked_encoded_response(self):
         return self.type == HttpParser.types.RESPONSE_PARSER and \
                b'transfer-encoding' in self.headers and \
                self.headers[b'transfer-encoding'][1].lower() == b'chunked'
 
-    def parse(self, data):
-        self.raw += data
-        data = self.buffer + data
+    def parse(self, raw):
+        self.bytes += raw
+        raw = self.buffer + raw
         self.buffer = b''
 
-        more = True if len(data) > 0 else False
+        more = True if len(raw) > 0 else False
         while more:
-            more, data = self.process(data)
-        self.buffer = data
+            more, raw = self.process(raw)
+        self.buffer = raw
 
-    def process(self, data):
+    def process(self, raw):
         if self.state in (HttpParser.states.HEADERS_COMPLETE,
                           HttpParser.states.RCVING_BODY,
                           HttpParser.states.COMPLETE) and \
@@ -229,22 +431,22 @@ class HttpParser(object):
 
             if b'content-length' in self.headers:
                 self.state = HttpParser.states.RCVING_BODY
-                self.body += data
+                self.body += raw
                 if len(self.body) >= int(self.headers[b'content-length'][1]):
                     self.state = HttpParser.states.COMPLETE
             elif self.is_chunked_encoded_response():
                 if not self.chunk_parser:
                     self.chunk_parser = ChunkParser()
-                self.chunk_parser.parse(data)
+                self.chunk_parser.parse(raw)
                 if self.chunk_parser.state == ChunkParser.states.COMPLETE:
                     self.body = self.chunk_parser.body
                     self.state = HttpParser.states.COMPLETE
 
             return False, b''
 
-        line, data = HttpParser.split(data)
+        line, raw = HttpParser.split(raw)
         if line is False:
-            return line, data
+            return line, raw
 
         if self.state == HttpParser.states.INITIALIZED:
             self.process_line(line)
@@ -256,7 +458,7 @@ class HttpParser(object):
         if self.state == HttpParser.states.LINE_RCVD and \
                 self.type == HttpParser.types.REQUEST_PARSER and \
                 self.method == b'CONNECT' and \
-                data == CRLF:
+                raw == CRLF:
             self.state = HttpParser.states.COMPLETE
 
         # When raw request has ended with \r\n\r\n and no more http headers are expected
@@ -265,7 +467,7 @@ class HttpParser(object):
         elif self.state == HttpParser.states.HEADERS_COMPLETE and \
                 self.type == HttpParser.types.REQUEST_PARSER and \
                 self.method != b'POST' and \
-                self.raw.endswith(CRLF * 2):
+                self.bytes.endswith(CRLF * 2):
             self.state = HttpParser.states.COMPLETE
         elif self.state == HttpParser.states.HEADERS_COMPLETE and \
                 self.type == HttpParser.types.REQUEST_PARSER and \
@@ -273,13 +475,13 @@ class HttpParser(object):
                 (b'content-length' not in self.headers or
                  (b'content-length' in self.headers and
                   int(self.headers[b'content-length'][1]) == 0)) and \
-                self.raw.endswith(CRLF * 2):
+                self.bytes.endswith(CRLF * 2):
             self.state = HttpParser.states.COMPLETE
 
-        return len(data) > 0, data
+        return len(raw) > 0, raw
 
-    def process_line(self, data):
-        line = data.split(SP)
+    def process_line(self, raw):
+        line = raw.split(WHITESPACE)
         if self.type == HttpParser.types.REQUEST_PARSER:
             self.method = line[0].upper()
             self.url = urlparse.urlsplit(line[1])
@@ -288,17 +490,18 @@ class HttpParser(object):
             self.version = line[0]
             self.code = line[1]
             self.reason = b' '.join(line[2:])
+        self.set_host_port()
         self.state = HttpParser.states.LINE_RCVD
 
-    def process_header(self, data):
-        if len(data) == 0:
+    def process_header(self, raw):
+        if len(raw) == 0:
             if self.state == HttpParser.states.RCVING_HEADERS:
                 self.state = HttpParser.states.HEADERS_COMPLETE
             elif self.state == HttpParser.states.LINE_RCVD:
                 self.state = HttpParser.states.RCVING_HEADERS
         else:
             self.state = HttpParser.states.RCVING_HEADERS
-            parts = data.split(COLON)
+            parts = raw.split(COLON)
             key = parts[0].strip()
             value = COLON.join(parts[1:]).strip()
             self.headers[key.lower()] = (key, value)
@@ -342,452 +545,474 @@ class HttpParser(object):
         return k + b': ' + v
 
     @staticmethod
-    def split(data):
-        pos = data.find(CRLF)
+    def split(raw):
+        pos = raw.find(CRLF)
         if pos == -1:
-            return False, data
-        line = data[:pos]
-        data = data[pos + len(CRLF):]
-        return line, data
+            return False, raw
+        line = raw[:pos]
+        raw = raw[pos + len(CRLF):]
+        return line, raw
 
+    ###################################################################################
+    # HttpParser was originally written to parse the incoming raw Http requests.
+    # Since request / response objects passed to HttpProtocolBasePlugin methods
+    # are also HttpParser objects, methods below were added to simplify developer API.
+    ####################################################################################
 
-class TCPConnection(object):
-    """TCP server/client connection abstraction."""
+    def has_upstream_server(self):
+        """Host field SHOULD be None for incoming local WebServer requests."""
+        return True if self.host is not None else False
 
-    def __init__(self, what):
-        self.conn = None
-        self.buffer = b''
-        self.closed = False
-        self.what = what  # server or client
 
-    def send(self, data):
-        # TODO: Gracefully handle BrokenPipeError exceptions
-        return self.conn.send(data)
+class HttpProtocolException(Exception):
+    """Top level HttpProtocolException exception class.
 
-    def recv(self, bufsiz=DEFAULT_BUFFER_SIZE):
-        try:
-            data = self.conn.recv(bufsiz)
-            if len(data) == 0:
-                logger.debug('rcvd 0 bytes from %s' % self.what)
-                return None
-            logger.debug('rcvd %d bytes from %s' % (len(data), self.what))
-            return data
-        except Exception as e:
-            if e.errno == errno.ECONNRESET:
-                logger.debug('%r' % e)
-            else:
-                logger.exception(
-                    'Exception while receiving from connection %s %r with reason %r' % (self.what, self.conn, e))
-            return None
+    All exceptions raised during execution of Http request lifecycle MUST
+    inherit HttpProtocolException base class. Implement response() method
+    to optionally return custom response to client."""
 
-    def close(self):
-        self.conn.close()
-        self.closed = True
+    def __init__(self):
+        pass
 
-    def buffer_size(self):
-        return len(self.buffer)
+    def response(self, request):
+        pass
 
-    def has_buffer(self):
-        return self.buffer_size() > 0
 
-    def queue(self, data):
-        self.buffer += data
+class HttpRequestRejected(HttpProtocolException):
+    """Generic exception that can be used to reject the client requests.
 
-    def flush(self):
-        sent = self.send(self.buffer)
-        self.buffer = self.buffer[sent:]
-        logger.debug('flushed %d bytes to %s' % (sent, self.what))
+    Connections can either be dropped/closed or optionally an
+    HTTP status code can be returned."""
 
+    def __init__(self, status_code=None, body=None):
+        super(HttpRequestRejected, self).__init__()
 
-class TCPServerConnection(TCPConnection):
-    """Establish connection to destination server."""
+        self.status_code = status_code
+        self.body = body
 
-    def __init__(self, host, port):
-        super(TCPServerConnection, self).__init__(b'server')
-        self.addr = (host, int(port))
+    def response(self, _request):
+        pkt = []
+        if self.status_code:
+            pkt.append(b'HTTP/1.1 ' + self.status_code)
+            pkt.append(PROXY_AGENT_HEADER)
+        if self.body:
+            pkt.append(b'Content-Length: ' + bytes(len(self.body)))
+            pkt.append(CRLF)
+            pkt.append(self.body)
+        return CRLF.join(pkt) if len(pkt) > 0 else None
 
-    def __del__(self):
-        if self.conn:
-            self.close()
 
-    def connect(self):
-        self.conn = socket.create_connection((self.addr[0], self.addr[1]))
+class HttpProtocolConfig(object):
+    """Holds various configuration values applicable to HttpProtocolHandler.
 
+    This config class helps us avoid passing around bunch of key/value pairs across methods.
+    """
 
-class TCPClientConnection(TCPConnection):
-    """Accepted client connection."""
+    def __init__(self, auth_code=DEFAULT_BASIC_AUTH, server_recvbuf_size=DEFAULT_SERVER_RECVBUF_SIZE,
+                 client_recvbuf_size=DEFAULT_CLIENT_RECVBUF_SIZE, pac_file=DEFAULT_PAC_FILE,
+                 pac_file_url_path=DEFAULT_PAC_FILE_URL_PATH, plugins=None):
+        self.auth_code = auth_code
+        self.server_recvbuf_size = server_recvbuf_size
+        self.client_recvbuf_size = client_recvbuf_size
+        self.pac_file = pac_file
+        self.pac_file_url_path = pac_file_url_path
+        if plugins is None:
+            plugins = DEFAULT_PLUGINS
+        self.plugins: Dict = plugins
 
-    def __init__(self, conn, addr):
-        super(TCPClientConnection, self).__init__(b'client')
-        self.conn = conn
-        self.addr = addr
 
+class HttpProtocolBasePlugin(object):
+    """Base HttpProtocolHandler Plugin class.
 
-class ProxyError(Exception):
-    pass
+    Implement various lifecycle event methods to customize behavior."""
 
+    def __init__(self, config: HttpProtocolConfig, client: TcpClientConnection, request: HttpParser):
+        self.config = config
+        self.client = client
+        self.request = request
 
-class ProxyConnectionFailed(ProxyError):
+    def name(self) -> str:
+        """A unique name for your plugin.
 
-    def __init__(self, host, port, reason):
-        self.host = host
-        self.port = port
-        self.reason = reason
+        Defaults to name of the class. This helps plugin developers to directly
+        access a specific plugin by its name."""
+        return self.__class__.__name__
 
-    def __str__(self):
-        return '<ProxyConnectionFailed - %s:%s - %s>' % (self.host, self.port, self.reason)
+    def get_descriptors(self):
+        return [], [], []
 
+    def flush_to_descriptors(self, w):
+        pass
 
-class ProxyAuthenticationFailed(ProxyError):
-    pass
+    def read_from_descriptors(self, r):
+        pass
 
+    def on_client_data(self, raw: bytes):
+        return raw
 
-class HTTPProxy(threading.Thread):
-    """HTTP proxy implementation.
+    def on_request_complete(self):
+        """Called right after client request parser has reached COMPLETE state."""
+        pass
 
-    Accepts `Client` connection object and act as a proxy between client and server.
-    """
+    def handle_response_chunk(self, chunk: bytes):
+        """Handle data chunks as received from the server.
 
-    def __init__(self, client, auth_code=DEFAULT_BASIC_AUTH, server_recvbuf_size=DEFAULT_SERVER_RECVBUF_SIZE,
-                 client_recvbuf_size=DEFAULT_CLIENT_RECVBUF_SIZE, pac_file=DEFAULT_PAC_FILE):
-        super(HTTPProxy, self).__init__()
+        Return optionally modified chunk to return back to client."""
+        return chunk
 
-        self.start_time = self._now()
-        self.last_activity = self.start_time
+    def access_log(self):
+        pass
 
-        self.auth_code = auth_code
-        self.client = client
-        self.client_recvbuf_size = client_recvbuf_size
-        self.server = None
-        self.server_recvbuf_size = server_recvbuf_size
+    def on_client_connection_close(self):
+        pass
 
-        self.request = HttpParser(HttpParser.types.REQUEST_PARSER)
-        self.response = HttpParser(HttpParser.types.RESPONSE_PARSER)
 
-        self.pac_file = pac_file
+class ProxyConnectionFailed(HttpProtocolException):
+    """Exception raised when HttpProxyPlugin is unable to establish connection to upstream server."""
 
-    @staticmethod
-    def _now():
-        return datetime.datetime.utcnow()
+    RESPONSE_PKT = CRLF.join([
+        b'HTTP/1.1 502 Bad Gateway',
+        PROXY_AGENT_HEADER,
+        b'Content-Length: 11',
+        b'Connection: close',
+        CRLF
+    ]) + b'Bad Gateway'
 
-    def _inactive_for(self):
-        return (self._now() - self.last_activity).seconds
+    def __init__(self, host, port, reason):
+        self.host = host
+        self.port = port
+        self.reason = reason
 
-    def _is_inactive(self):
-        return self._inactive_for() > 30
+    def response(self, _request):
+        return self.RESPONSE_PKT
 
-    def _process_request(self, data):
-        # once we have connection to the server
-        # we don't parse the http request packets
-        # any further, instead just pipe incoming
-        # data from client to server
-        if self.server and not self.server.closed:
-            self.server.queue(data)
-            return
+    def __str__(self):
+        return '<ProxyConnectionFailed - %s:%s - %s>' % (self.host, self.port, self.reason)
 
-        # parse http request
-        self.request.parse(data)
 
-        # once http request parser has reached the state complete
-        # we attempt to establish connection to destination server
-        if self.request.state == HttpParser.states.COMPLETE:
-            logger.debug('request parser is in state complete')
+class ProxyAuthenticationFailed(HttpProtocolException):
+    """Exception raised when Http Proxy auth is enabled and
+    incoming request doesn't present necessary credentials."""
 
-            if self.auth_code:
-                if b'proxy-authorization' not in self.request.headers or \
-                        self.request.headers[b'proxy-authorization'][1] != self.auth_code:
-                    raise ProxyAuthenticationFailed()
+    RESPONSE_PKT = CRLF.join([
+        b'HTTP/1.1 407 Proxy Authentication Required',
+        PROXY_AGENT_HEADER,
+        b'Content-Length: 29',
+        b'Connection: close',
+        b'Proxy-Authenticate: Basic',
+        CRLF
+    ]) + b'Proxy Authentication Required'
 
-            if self.request.method == b'CONNECT':
-                host, port = self.request.url.path.split(COLON)
-            elif self.request.url:
-                host, port = self.request.url.hostname, self.request.url.port if self.request.url.port else 80
-            else:
-                raise Exception('Invalid request\n%s' % self.request.raw)
+    def response(self, _request):
+        return self.RESPONSE_PKT
 
-            if host is None and self.pac_file:
-                self._serve_pac_file()
-                return True
 
-            self.server = TCPServerConnection(host, port)
-            try:
-                logger.debug('connecting to server %s:%s' % (host, port))
-                self.server.connect()
-                logger.debug('connected to server %s:%s' % (host, port))
-            except Exception as e:  # TimeoutError, socket.gaierror
-                self.server.closed = True
-                raise ProxyConnectionFailed(host, port, repr(e))
-
-            # for http connect methods (https requests)
-            # queue appropriate response for client
-            # notifying about established connection
-            if self.request.method == b'CONNECT':
-                self.client.queue(PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
-            # for usual http requests, re-build request packet
-            # and queue for the server with appropriate headers
-            else:
-                self.server.queue(self.request.build(
-                    del_headers=[b'proxy-authorization', b'proxy-connection', b'connection', b'keep-alive'],
-                    add_headers=[(b'Via', b'1.1 proxy.py v%s' % version), (b'Connection', b'Close')]
-                ))
+class HttpProxyPlugin(HttpProtocolBasePlugin):
+    """HttpProtocolHandler plugin which implements HttpProxy specifications."""
 
-    def _process_response(self, data):
-        # parse incoming response packet
-        # only for non-https requests
-        if not self.request.method == b'CONNECT':
-            self.response.parse(data)
+    PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT = CRLF.join([
+        b'HTTP/1.1 200 Connection established',
+        PROXY_AGENT_HEADER,
+        CRLF
+    ])
 
-        # queue data for client
-        self.client.queue(data)
+    def __init__(self, config: HttpProtocolConfig, client: TcpClientConnection, request: HttpParser):
+        super(HttpProxyPlugin, self).__init__(config, client, request)
+        self.server = None
+        self.response = HttpParser(HttpParser.types.RESPONSE_PARSER)
 
-    def _access_log(self):
-        host, port = self.server.addr if self.server else (None, None)
-        if self.request.method == b'CONNECT':
-            logger.info(
-                '%s:%s - %s %s:%s' % (self.client.addr[0], self.client.addr[1], self.request.method, host, port))
-        elif self.request.method:
-            logger.info('%s:%s - %s %s:%s%s - %s %s - %s bytes' % (
-                self.client.addr[0], self.client.addr[1], self.request.method, host, port, self.request.build_url(),
-                self.response.code, self.response.reason, len(self.response.raw)))
+    def get_descriptors(self):
+        if not self.request.has_upstream_server():
+            return [], [], []
 
-    def _get_waitable_lists(self):
-        rlist, wlist, xlist = [self.client.conn], [], []
-        if self.client.has_buffer():
-            wlist.append(self.client.conn)
+        r, w, x = [], [], []
         if self.server and not self.server.closed:
-            rlist.append(self.server.conn)
+            r.append(self.server.conn)
         if self.server and not self.server.closed and self.server.has_buffer():
-            wlist.append(self.server.conn)
-        return rlist, wlist, xlist
+            w.append(self.server.conn)
+        return r, w, x
 
-    def _process_wlist(self, w):
-        if self.client.conn in w:
-            logger.debug('client is ready for writes, flushing client buffer')
-            self.client.flush()
+    def flush_to_descriptors(self, w):
+        if not self.request.has_upstream_server():
+            return
 
         if self.server and not self.server.closed and self.server.conn in w:
-            logger.debug('server is ready for writes, flushing server buffer')
+            logger.debug('Server is ready for writes, flushing server buffer')
             self.server.flush()
 
-    def _process_rlist(self, r):
-        """Returns True if connection to client must be closed."""
-        if self.client.conn in r:
-            logger.debug('client is ready for reads, reading')
-            data = self.client.recv(self.client_recvbuf_size)
-            self.last_activity = self._now()
+    def read_from_descriptors(self, r):
+        if not self.request.has_upstream_server():
+            return
 
-            if not data:
-                logger.debug('client closed connection, breaking')
+        if self.server and not self.server.closed and self.server.conn in r:
+            logger.debug('Server is ready for reads, reading')
+            raw = self.server.recv(self.config.server_recvbuf_size)
+            # self.last_activity = HttpProtocolHandler.now()
+            if not raw:
+                logger.debug('Server closed connection, tearing down...')
                 return True
+            # parse incoming response packet
+            # only for non-https requests
+            if not self.request.method == b'CONNECT':
+                self.response.parse(raw)
+            else:
+                # Only purpose of increasing memory footprint is to print response length in access log
+                # Not worth it? Optimize to only persist lengths?
+                self.response.bytes += raw
+            # queue raw data for client
+            self.client.queue(raw)
+
+    def on_client_connection_close(self):
+        if not self.request.has_upstream_server():
+            return
 
-            try:
-                return self._process_request(data)
-            except (ProxyAuthenticationFailed, ProxyConnectionFailed) as e:
-                logger.exception(e)
-                self.client.queue(HTTPProxy._get_response_pkt_by_exception(e))
-                self.client.flush()
-                return True
+        if self.server:
+            logger.debug(
+                'Closed server connection with pending server buffer size %d bytes' % self.server.buffer_size())
+            if not self.server.closed:
+                self.server.close()
 
-        if self.server and not self.server.closed and self.server.conn in r:
-            logger.debug('server is ready for reads, reading')
-            data = self.server.recv(self.server_recvbuf_size)
-            self.last_activity = self._now()
+    def on_client_data(self, raw):
+        if not self.request.has_upstream_server():
+            return raw
 
-            if not data:
-                logger.debug('server closed connection')
-                self.server.close()
-            else:
-                self._process_response(data)
+        if self.server and not self.server.closed:
+            self.server.queue(raw)
+            return None
+        else:
+            return raw
+
+    def on_request_complete(self):
+        if not self.request.has_upstream_server():
+            return
 
-        return False
+        self.authenticate(self.request.headers)
+        self.connect_upstream(self.request.host, self.request.port)
+        # for http connect methods (https requests)
+        # queue appropriate response for client
+        # notifying about established connection
+        if self.request.method == b'CONNECT':
+            self.client.queue(HttpProxyPlugin.PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
+        # for general http requests, re-build request packet
+        # and queue for the server with appropriate headers
+        else:
+            self.server.queue(self.request.build(
+                del_headers=[b'proxy-authorization', b'proxy-connection', b'connection',
+                             b'keep-alive'],
+                add_headers=[(b'Via', b'1.1 proxy.py v%s' % version), (b'Connection', b'Close')]
+            ))
+
+    def access_log(self):
+        if not self.request.has_upstream_server():
+            return
 
-    def _serve_pac_file(self):
-        self.client.queue(PAC_FILE_RESPONSE_PREFIX)
+        host, port = self.server.addr if self.server else (None, None)
+        if self.request.method == b'CONNECT':
+            logger.info(
+                '%s:%s - %s %s:%s - %s bytes' % (self.client.addr[0], self.client.addr[1],
+                                                 text_(self.request.method), text_(host),
+                                                 text_(port), len(self.response.bytes)))
+        elif self.request.method:
+            logger.info('%s:%s - %s %s:%s%s - %s %s - %s bytes' % (
+                self.client.addr[0], self.client.addr[1], text_(self.request.method), text_(host), port,
+                text_(self.request.build_url()), text_(self.response.code), text_(self.response.reason),
+                len(self.response.bytes)))
+
+    def authenticate(self, headers):
+        if self.config.auth_code:
+            if b'proxy-authorization' not in headers or \
+                    headers[b'proxy-authorization'][1] != self.config.auth_code:
+                raise ProxyAuthenticationFailed()
+
+    def connect_upstream(self, host, port):
+        self.server = TcpServerConnection(host, port)
         try:
-            with open(self.pac_file, 'r') as f:
-                logger.debug('serving pac file from disk')
-                self.client.queue(f.read())
-        except IOError:
-            logger.debug('serving pac file content from buffer')
-            self.client.queue(self.pac_file)
-        self.client.flush()
+            logger.debug('Connecting to upstream %s:%s' % (host, port))
+            self.server.connect()
+            logger.debug('Connected to upstream %s:%s' % (host, port))
+        except Exception as e:  # TimeoutError, socket.gaierror
+            self.server.closed = True
+            raise ProxyConnectionFailed(host, port, repr(e))
 
-    def _process(self):
-        while True:
-            rlist, wlist, xlist = self._get_waitable_lists()
-            r, w, x = select.select(rlist, wlist, xlist, 1)
 
-            self._process_wlist(w)
-            if self._process_rlist(r):
-                break
+class HttpWebServerPlugin(HttpProtocolBasePlugin):
+    """HttpProtocolHandler plugin which handles incoming requests to local webserver."""
 
-            if self.client.buffer_size() == 0:
-                # Client may use same connection for multiple request cycles,
-                # hence not appropriate to close the connection upon parser states.
-                #
-                # if self.response.state == HttpParser.states.COMPLETE:
-                #     logger.debug('client buffer is empty and response state is complete, breaking')
-                #     break
+    PAC_FILE_RESPONSE_PREFIX = CRLF.join([
+        b'HTTP/1.1 200 OK',
+        b'Content-Type: application/x-ns-proxy-autoconfig',
+        b'Connection: close',
+        CRLF
+    ])
 
-                if self._is_inactive():
-                    logger.debug('client buffer is empty and maximum inactivity has reached, breaking')
-                    break
+    def __init__(self, config: HttpProtocolConfig, client: TcpClientConnection, request: HttpParser):
+        super(HttpWebServerPlugin, self).__init__(config, client, request)
 
-    @staticmethod
-    def _get_response_pkt_by_exception(e):
-        if e.__class__.__name__ == 'ProxyAuthenticationFailed':
-            return PROXY_AUTHENTICATION_REQUIRED_RESPONSE_PKT
-        if e.__class__.__name__ == 'ProxyConnectionFailed':
-            return BAD_GATEWAY_RESPONSE_PKT
+    def on_request_complete(self):
+        if self.request.has_upstream_server():
+            return
 
-    def run(self):
-        logger.debug('Proxying connection %r' % self.client.conn)
-        try:
-            self._process()
-        except KeyboardInterrupt:
-            pass
-        except Exception as e:
-            logger.exception('Exception while handling connection %r with reason %r' % (self.client.conn, e))
-        finally:
-            logger.debug(
-                'closing client connection with pending client buffer size %d bytes' % self.client.buffer_size())
-            self.client.close()
-            if self.server:
-                logger.debug(
-                    'closed client connection with pending server buffer size %d bytes' % self.server.buffer_size())
-            self._access_log()
-            logger.debug('Closing proxy for connection %r at address %r' % (self.client.conn, self.client.addr))
+        if self.config.pac_file and \
+                self.request.url.path == self.config.pac_file_url_path:
+            self.serve_pac_file()
+        else:
+            self.client.queue(CRLF.join([
+                b'HTTP/1.1 200 OK',
+                b'Server: proxy.py v%s' % version,
+                b'Content-Length: 42',
+                b'Connection: Close',
+                CRLF,
+                b'https://github.com/abhinavsingh/proxy.py'
+            ]))
+            self.client.flush()
 
+        return True
 
-class TCPServer(object):
-    """TCPServer server implementation.
+    def access_log(self):
+        if self.request.has_upstream_server():
+            return
+        logger.info('%s:%s - %s %s' % (self.client.addr[0], self.client.addr[1],
+                                       text_(self.request.method), text_(self.request.build_url())))
 
-    Inheritor MUST implement `handle` method. It accepts an instance of `TCPClientConnection`.
-    Optionally, can also implement `setup` and `shutdown` methods for custom bootstrapping and teardown.
-    """
+    def serve_pac_file(self):
+        self.client.queue(self.PAC_FILE_RESPONSE_PREFIX)
+        try:
+            with open(self.config.pac_file, 'r') as f:
+                logger.debug('Serving pac file from disk')
+                self.client.queue(f.read())
+        except IOError:
+            logger.debug('Serving pac file content from buffer')
+            self.client.queue(self.config.pac_file)
+        self.client.flush()
 
-    def __init__(self, hostname=DEFAULT_HOSTNAME, port=DEFAULT_PORT, backlog=DEFAULT_BACKLOG, ipv4=DEFAULT_IPV4):
-        self.hostname = hostname
-        self.port = port
-        self.backlog = backlog
-        self.ipv4 = ipv4
-        self.socket = None
 
-    def setup(self):
-        pass
+class HttpProtocolHandler(threading.Thread):
+    """HTTP, HTTPS, HTTP2, WebSockets protocol handler.
 
-    def handle(self, client):
-        raise NotImplementedError()
+    Accepts `Client` connection object, manages plugin invocations.
+    """
 
-    def shutdown(self):
-        pass
+    def __init__(self, client, config=None):
+        super(HttpProtocolHandler, self).__init__()
 
-    def run(self):
-        self.setup()
-        try:
-            self.socket = socket.socket(socket.AF_INET if self.ipv4 is True else socket.AF_INET6, socket.SOCK_STREAM)
-            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            self.socket.bind((self.hostname, self.port))
-            self.socket.listen(self.backlog)
-            logger.info('Started server on port %d' % self.port)
-            while True:
-                conn, addr = self.socket.accept()
-                client = TCPClientConnection(conn, addr)
-                self.handle(client)
-        except Exception as e:
-            logger.exception('Exception while running the server %r' % e)
-        finally:
-            self.shutdown()
-            logger.info('Closing server socket')
-            self.socket.close()
+        self.start_time = self.now()
+        self.last_activity = self.start_time
 
+        self.client = client
+        self.config = config if config else HttpProtocolConfig()
+        self.request = HttpParser(HttpParser.types.REQUEST_PARSER)
 
-class HTTPServer(TCPServer):
-    """HTTP server implementation.
+        self.plugins: Dict[str, HttpProtocolBasePlugin] = {}
+        for klass in self.config.plugins:
+            instance = klass(self.config, self.client, self.request)
+            self.plugins[instance.name()] = instance
 
-    Pre-spawns worker process to utilize all cores available on the system.  Accepted `TCPClientConnection` is
-    dispatched over a queue to workers.  One of the worker picks up the work and starts a new thread to handle the
-    client request.
-    """
+    @staticmethod
+    def now():
+        return datetime.datetime.utcnow()
 
-    def __init__(self, hostname=DEFAULT_HOSTNAME, port=DEFAULT_PORT, backlog=DEFAULT_BACKLOG,
-                 num_workers=DEFAULT_NUM_WORKERS,
-                 auth_code=DEFAULT_BASIC_AUTH, server_recvbuf_size=DEFAULT_SERVER_RECVBUF_SIZE,
-                 client_recvbuf_size=DEFAULT_CLIENT_RECVBUF_SIZE, pac_file=DEFAULT_PAC_FILE, ipv4=DEFAULT_IPV4):
-        super(HTTPServer, self).__init__(hostname, port, backlog, ipv4)
-        self.auth_code = auth_code
-        self.client_recvbuf_size = client_recvbuf_size
-        self.server_recvbuf_size = server_recvbuf_size
-        self.pac_file = pac_file
+    def connection_inactive_for(self):
+        return (self.now() - self.last_activity).seconds
 
-        self.worker_queue = multiprocessing.Queue()
-        self.num_workers = multiprocessing.cpu_count()
-        if num_workers > 0:
-            self.num_workers = num_workers
-        self.workers = []
+    def is_connection_inactive(self):
+        return self.connection_inactive_for() > 30
 
-    def setup(self):
-        logger.info('Starting %d workers' % self.num_workers)
-        for worker_id in range(self.num_workers):
-            worker = Worker(self.worker_queue, auth_code=self.auth_code, server_recvbuf_size=self.server_recvbuf_size,
-                            client_recvbuf_size=self.client_recvbuf_size, pac_file=self.pac_file)
-            worker.daemon = True
-            worker.start()
-            self.workers.append(worker)
+    def run_once(self):
+        """Returns True if proxy must teardown."""
+        # Prepare list of descriptors
+        read_desc, write_desc, err_desc = [self.client.conn], [], []
+        if self.client.has_buffer():
+            write_desc.append(self.client.conn)
 
-    def handle(self, client):
-        self.worker_queue.put((Worker.operations.DEFAULT, client))
+        for plugin in self.plugins.values():
+            plugin_read_desc, plugin_write_desc, plugin_err_desc = plugin.get_descriptors()
+            read_desc += plugin_read_desc
+            write_desc += plugin_write_desc
+            err_desc += plugin_err_desc
 
-    def shutdown(self):
-        logger.info('Shutting down %d workers' % self.num_workers)
-        for worker_id in range(self.num_workers):
-            self.worker_queue.put((Worker.operations.SHUTDOWN, None))
-        for worker_id in range(self.num_workers):
-            self.workers[worker_id].join()
+        readable, writable, errored = select.select(read_desc, write_desc, err_desc, 1)
 
+        # Flush buffer for ready to write sockets
+        if self.client.conn in writable:
+            logger.debug('Client is ready for writes, flushing client buffer')
+            try:
+                self.client.flush()
+            except BrokenPipeError:
+                logging.error('BrokenPipeError when flushing buffer for client')
+                return True
 
-class Worker(multiprocessing.Process):
-    """Generic worker class implementation.
+        for plugin in self.plugins.values():
+            plugin.flush_to_descriptors(writable)
 
-    Worker instance accepts (operation, payload) over work queue and
-    starts a new thread to complete the work.
-    """
+        # Read from ready to read sockets
+        if self.client.conn in readable:
+            logger.debug('Client is ready for reads, reading')
+            client_data = self.client.recv(self.config.client_recvbuf_size)
+            self.last_activity = self.now()
+            if not client_data:
+                logger.debug('Client closed connection, tearing down...')
+                return True
 
-    operations = namedtuple('WorkerOperations', (
-        'DEFAULT',  # Default worker action
-        'SHUTDOWN',
-    ))(1, 2)
+            plugin_index = 0
+            plugins = list(self.plugins.values())
+            while plugin_index < len(plugins) and client_data:
+                client_data = plugins[plugin_index].on_client_data(client_data)
+                plugin_index += 1
+
+            if client_data:
+                try:
+                    # Parse http request
+                    self.request.parse(client_data)
+                    if self.request.state == HttpParser.states.COMPLETE:
+                        for plugin in self.plugins.values():
+                            # TODO: Cleanup by not returning True for teardown cases
+                            plugin_response = plugin.on_request_complete()
+                            if type(plugin_response) is bool:
+                                return True
+                except HttpProtocolException as e:  # ProxyAuthenticationFailed, ProxyConnectionFailed, HttpRequestRejected
+                    # logger.exception(e)
+                    response = e.response(self.request)
+                    if response:
+                        self.client.queue(response)
+                        # But is client also ready for writes?
+                        self.client.flush()
+                    raise e
+
+        for plugin in self.plugins.values():
+            teardown = plugin.read_from_descriptors(readable)
+            if teardown:
+                return True
 
-    def __init__(self, work_queue, auth_code=DEFAULT_BASIC_AUTH, server_recvbuf_size=DEFAULT_SERVER_RECVBUF_SIZE,
-                 client_recvbuf_size=DEFAULT_CLIENT_RECVBUF_SIZE, pac_file=DEFAULT_PAC_FILE):
-        super(Worker, self).__init__()
-        self.work_queue = work_queue
-        self.auth_code = auth_code
-        self.server_recvbuf_size = server_recvbuf_size
-        self.client_recvbuf_size = client_recvbuf_size
-        self.pac_file = pac_file
+        # Teardown if client buffer is empty and connection is inactive
+        if self.client.buffer_size() == 0:
+            if self.is_connection_inactive():
+                logger.debug('Client buffer is empty and maximum inactivity has reached '
+                             'between client and server connection, tearing down...')
+                return True
 
     def run(self):
-        while True:
-            try:
-                op, payload = self.work_queue.get(True, 1)
-                if op == Worker.operations.DEFAULT:
-                    proxy = HTTPProxy(payload,
-                                      auth_code=self.auth_code,
-                                      server_recvbuf_size=self.server_recvbuf_size,
-                                      client_recvbuf_size=self.client_recvbuf_size,
-                                      pac_file=self.pac_file)
-                    proxy.daemon = True
-                    proxy.start()
-                elif op == Worker.operations.SHUTDOWN:
+        logger.debug('Proxying connection %r' % self.client.conn)
+        try:
+            while True:
+                teardown = self.run_once()
+                if teardown:
                     break
-            except queue.Empty:
-                pass
-            # Safeguard against https://gist.github.com/abhinavsingh/b8d4266ff4f38b6057f9c50075e8cd75
-            except ConnectionRefusedError:
-                pass
-            except KeyboardInterrupt:
-                break
+        except KeyboardInterrupt:
+            pass
+        except Exception as e:
+            logger.exception('Exception while handling connection %r with reason %r' % (self.client.conn, e))
+        finally:
+            for plugin in self.plugins.values():
+                plugin.access_log()
+
+            self.client.close()
+            logger.debug('Closed client connection with pending '
+                         'client buffer size %d bytes' % self.client.buffer_size())
+            for plugin in self.plugins.values():
+                plugin.on_client_connection_close()
+
+            logger.debug('Closed proxy for connection %r '
+                         'at address %r' % (self.client.conn, self.client.addr))
 
 
 def set_open_file_limit(soft_limit):
@@ -799,7 +1024,23 @@ def set_open_file_limit(soft_limit):
             logger.debug('Open file descriptor soft limit set to %d' % soft_limit)
 
 
-def parse_args(args):
+def load_plugins(plugins) -> List:
+    p = []
+    plugins = plugins.split(',')
+    for plugin in plugins:
+        plugin = plugin.strip()
+        if plugin == '':
+            continue
+        logging.debug('Loading plugin %s', plugin)
+        module_name, klass_name = plugin.rsplit('.', 1)
+        module = importlib.import_module(module_name)
+        klass = getattr(module, klass_name)
+        logging.info('%s initialized' % klass.__name__)
+        p.append(klass)
+    return p
+
+
+def init_parser():
     parser = argparse.ArgumentParser(
         description='proxy.py v%s' % __version__,
         epilog='Proxy.py not working? Report at: %s/issues/new' % __homepage__
@@ -815,23 +1056,32 @@ def parse_args(args):
                              'client in a single recv() operation. Bump this '
                              'value for faster uploads at the expense of '
                              'increased RAM.')
-    parser.add_argument('--hostname', type=str, default=DEFAULT_HOSTNAME,
+    parser.add_argument('--hostname', type=str, default=DEFAULT_IPV4_HOSTNAME,
                         help='Default: 127.0.0.1. Server IP address.')
     parser.add_argument('--ipv4', action='store_true', default=DEFAULT_IPV4,
                         help='Whether to listen on IPv4 address. '
                              'By default server only listens on IPv6.')
+    parser.add_argument('--enable-http-proxy', type=bool, default=DEFAULT_ENABLE_HTTP_PROXY,
+                        help='Default: True.  Whether to enable proxy.HttpProxyPlugin.')
+    parser.add_argument('--enable-web-server', type=bool, default=DEFAULT_ENABLE_WEB_SERVER,
+                        help='Default: False.  Whether to enable proxy.HttpWebServerPlugin.')
     parser.add_argument('--log-level', type=str, default=DEFAULT_LOG_LEVEL,
                         help='Valid options: DEBUG, INFO (default), WARNING, ERROR, CRITICAL. '
                              'Both upper and lowercase values are allowed.'
                              'You may also simply use the leading character e.g. --log-level d')
+    parser.add_argument('--log-format', type=str, default=DEFAULT_LOG_FORMAT,
+                        help='Log format for Python logger.')
     parser.add_argument('--open-file-limit', type=int, default=DEFAULT_OPEN_FILE_LIMIT,
                         help='Default: 1024. Maximum number of files (TCP connections) '
                              'that proxy.py can open concurrently.')
-    parser.add_argument('--port', type=int, default=DEFAULT_PORT,
-                        help='Default: 8899. Server port.')
     parser.add_argument('--pac-file', type=str, default=DEFAULT_PAC_FILE,
                         help='A file (Proxy Auto Configuration) or string to serve when '
                              'the server receives a direct file request.')
+    parser.add_argument('--pac-file-url-path', type=str, default=DEFAULT_PAC_FILE_URL_PATH,
+                        help='Web server path to serve the PAC file.')
+    parser.add_argument('--plugins', type=str, default='', help='Comma separated plugins')
+    parser.add_argument('--port', type=int, default=DEFAULT_PORT,
+                        help='Default: 8899. Server port.')
     parser.add_argument('--server-recvbuf-size', type=int, default=DEFAULT_SERVER_RECVBUF_SIZE,
                         help='Default: 8 KB. Maximum amount of data received from the '
                              'server in a single recv() operation. Bump this '
@@ -839,11 +1089,23 @@ def parse_args(args):
                              'increased RAM.')
     parser.add_argument('--num-workers', type=int, default=DEFAULT_NUM_WORKERS,
                         help='Defaults to number of CPU cores.')
-    return parser.parse_args(args)
+    return parser
 
 
 def main():
-    args = parse_args(sys.argv[1:])
+    if not PY3 and not UNDER_TEST:
+        print(
+            'DEPRECATION: "develop" branch no longer supports Python 2.7.  Kindly upgrade to Python 3+. '
+            'If for some reasons you cannot upgrade, consider using "master" branch or simply '
+            '"pip install proxy.py".'
+            '\n\n'
+            'DEPRECATION: Python 2.7 will reach the end of its life on January 1st, 2020. '
+            'Please upgrade your Python as Python 2.7 won\'t be maintained after that date. '
+            'A future version of pip will drop support for Python 2.7.')
+        sys.exit(0)
+
+    parser = init_parser()
+    args = parser.parse_args(sys.argv[1:])
     logging.basicConfig(level=getattr(logging,
                                       {
                                           'D': 'DEBUG',
@@ -852,13 +1114,7 @@ def main():
                                           'E': 'ERROR',
                                           'C': 'CRITICAL'
                                       }[args.log_level.upper()[0]]),
-                        format='%(asctime)s - %(levelname)s - pid:%(process)d - %(funcName)s:%(lineno)d - %(message)s')
-
-    if not PY3:
-        logging.error('"develop" branch no longer supports Python 2.7.  Kindly upgrade to Python 3+. '
-                      'If for some reasons you cannot upgrade, consider using "master" branch or simply '
-                      '"pip install proxy.py".')
-        sys.exit(0)
+                        format=args.log_format)
 
     try:
         set_open_file_limit(args.open_file_limit)
@@ -867,15 +1123,25 @@ def main():
         if args.basic_auth:
             auth_code = b'Basic %s' % base64.b64encode(bytes_(args.basic_auth))
 
-        server = HTTPServer(hostname=args.hostname,
-                            port=args.port,
-                            backlog=args.backlog,
-                            auth_code=auth_code,
-                            server_recvbuf_size=args.server_recvbuf_size,
-                            client_recvbuf_size=args.client_recvbuf_size,
-                            pac_file=args.pac_file,
-                            ipv4=args.ipv4,
-                            num_workers=args.num_workers)
+        config = HttpProtocolConfig(auth_code=auth_code,
+                                    server_recvbuf_size=args.server_recvbuf_size,
+                                    client_recvbuf_size=args.client_recvbuf_size,
+                                    pac_file=args.pac_file,
+                                    pac_file_url_path=args.pac_file_url_path)
+
+        default_plugins = ''
+        if args.enable_http_proxy:
+            default_plugins += 'proxy.HttpProxyPlugin,'
+        if args.enable_web_server:
+            default_plugins += 'proxy.HttpWebServerPlugin,'
+        config.plugins = load_plugins('%s%s' % (default_plugins, args.plugins))
+
+        server = MultiCoreRequestDispatcher(hostname=args.hostname,
+                                            port=args.port,
+                                            backlog=args.backlog,
+                                            ipv4=args.ipv4,
+                                            num_workers=args.num_workers,
+                                            config=config)
         server.run()
     except KeyboardInterrupt:
         pass


Test patch:
diff --git a/tests.py b/tests.py
index e0ce3d9..8859dc2 100644
--- a/tests.py
+++ b/tests.py
@@ -3,38 +3,155 @@
     proxy.py
     ~~~~~~~~
 
-    HTTP Proxy Server in Python.
+    HTTP, HTTPS, HTTP2 and WebSockets Proxy Server in Python.
 
-    :copyright: (c) 2013-2018 by Abhinav Singh.
+    :copyright: (c) 2013-2020 by Abhinav Singh.
     :license: BSD, see LICENSE for more details.
 """
-import multiprocessing
-import sys
 import base64
-import socket
 import logging
+import multiprocessing
+import os
+import socket
+import time
 import unittest
-from unittest.mock import patch, Mock
-from threading import Thread
 from contextlib import closing
-from proxy import HTTPProxy, ChunkParser, HttpParser, TCPClientConnection, HTTPServer, Worker, main
-from proxy import ProxyAuthenticationFailed, ProxyConnectionFailed
-from proxy import CRLF, version, PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from threading import Thread
+from unittest import mock
+
+import proxy
+
+logging.basicConfig(level=logging.DEBUG,
+                    format='%(asctime)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s')
 
-# logging.basicConfig(level=logging.DEBUG,
-#                     format='%(asctime)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s')
 
-# True if we are running on Python 3.
-if sys.version_info[0] == 3:
-    from http.server import HTTPServer, BaseHTTPRequestHandler
-else:
-    from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
+def get_available_port():
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
+        sock.bind(('', 0))
+        _, port = sock.getsockname()
+        return port
+
+
+@unittest.skipIf(os.getenv('TESTING_ON_TRAVIS', 0), 'Opening sockets not allowed on Travis')
+class TestTcpServer(unittest.TestCase):
+    ipv4_port = None
+    ipv6_port = None
+    ipv4_server = None
+    ipv6_server = None
+    ipv4_thread = None
+    ipv6_thread = None
+
+    class _TestTcpServer(proxy.TcpServer):
+
+        def handle(self, client):
+            data = client.recv(proxy.DEFAULT_BUFFER_SIZE)
+            assert data == b'HELLO'
+            client.conn.sendall(b'WORLD')
+            client.close()
+
+    @classmethod
+    def setUpClass(cls):
+        cls.ipv4_port = get_available_port()
+        cls.ipv6_port = get_available_port()
+        cls.ipv4_server = TestTcpServer._TestTcpServer(port=cls.ipv4_port, ipv4=True)
+        cls.ipv6_server = TestTcpServer._TestTcpServer(hostname=proxy.DEFAULT_IPV6_HOSTNAME, port=cls.ipv6_port,
+                                                       ipv4=False)
+        cls.ipv4_thread = Thread(target=cls.ipv4_server.run)
+        cls.ipv6_thread = Thread(target=cls.ipv6_server.run)
+        cls.ipv4_thread.setDaemon(True)
+        cls.ipv6_thread.setDaemon(True)
+        cls.ipv4_thread.start()
+        cls.ipv6_thread.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.ipv4_server.stop()
+        cls.ipv4_thread.join()
+
+    def baseTestCase(self, ipv4=True):
+        while True:
+            sock = None
+            try:
+                sock = socket.socket(socket.AF_INET if ipv4 else socket.AF_INET6, socket.SOCK_STREAM, 0)
+                sock.connect((proxy.DEFAULT_IPV4_HOSTNAME if ipv4 else proxy.DEFAULT_IPV6_HOSTNAME,
+                              self.ipv4_port if ipv4 else self.ipv6_port))
+                sock.sendall(b'HELLO')
+                data = sock.recv(proxy.DEFAULT_BUFFER_SIZE)
+                self.assertEqual(data, b'WORLD')
+                break
+            except ConnectionRefusedError:
+                time.sleep(0.1)
+            finally:
+                sock.close()
+
+    def testIpv4ClientConnection(self):
+        self.baseTestCase()
+
+    def testIpv6ClientConnection(self):
+        self.baseTestCase(ipv4=False)
+
+
+class MockHttpProxy(object):
+
+    def __init__(self, client, **kwargs):
+        self.client = client
+        self.kwargs = kwargs
+
+    def setDaemon(self, _val):
+        pass
+
+    def start(self):
+        self.client.conn.sendall(proxy.CRLF.join([b'HTTP/1.1 200 OK', proxy.CRLF]))
+        self.client.conn.close()
+
+
+def mock_tcp_proxy_side_effect(client, **kwargs):
+    return MockHttpProxy(client, **kwargs)
+
+
+@unittest.skipIf(os.getenv('TESTING_ON_TRAVIS', 0), 'Opening sockets not allowed on Travis')
+class TestMultiCoreRequestDispatcher(unittest.TestCase):
+    tcp_port = None
+    tcp_server = None
+    tcp_thread = None
+
+    @mock.patch.object(proxy, 'HttpProtocolHandler', side_effect=mock_tcp_proxy_side_effect)
+    def testHttpProxyConnection(self, mock_tcp_proxy):
+        try:
+            self.tcp_port = get_available_port()
+            self.tcp_server = proxy.MultiCoreRequestDispatcher(hostname=proxy.DEFAULT_IPV4_HOSTNAME, port=self.tcp_port,
+                                                               ipv4=True, num_workers=1)
+            self.tcp_thread = Thread(target=self.tcp_server.run)
+            self.tcp_thread.setDaemon(True)
+            self.tcp_thread.start()
+
+            while True:
+                try:
+                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
+                    sock.connect((proxy.DEFAULT_IPV4_HOSTNAME, self.tcp_port))
+                    sock.send(proxy.CRLF.join([
+                        b'GET http://httpbin.org/get HTTP/1.1',
+                        b'Host: httpbin.org',
+                        proxy.CRLF
+                    ]))
+                    data = sock.recv(proxy.DEFAULT_BUFFER_SIZE)
+                    self.assertEqual(data, proxy.CRLF.join([b'HTTP/1.1 200 OK', proxy.CRLF]))
+                    self.tcp_server.shutdown()  # explicit early call worker shutdown to avoid resource leak warnings
+                    break
+                except ConnectionRefusedError:
+                    time.sleep(0.1)
+                finally:
+                    sock.close()
+        finally:
+            self.tcp_server.stop()
+            self.tcp_thread.join()
 
 
 class TestChunkParser(unittest.TestCase):
 
     def setUp(self):
-        self.parser = ChunkParser()
+        self.parser = proxy.ChunkParser()
 
     def test_chunk_parse_basic(self):
         self.parser.parse(b''.join([
@@ -50,68 +167,68 @@ class TestChunkParser(unittest.TestCase):
         self.assertEqual(self.parser.chunk, b'')
         self.assertEqual(self.parser.size, None)
         self.assertEqual(self.parser.body, b'Wikipedia in\r\n\r\nchunks.')
-        self.assertEqual(self.parser.state, ChunkParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.COMPLETE)
 
     def test_chunk_parse_issue_27(self):
-        """Case when data ends with the chunk size but without CRLF."""
+        """Case when data ends with the chunk size but without ending CRLF."""
         self.parser.parse(b'3')
         self.assertEqual(self.parser.chunk, b'3')
         self.assertEqual(self.parser.size, None)
         self.assertEqual(self.parser.body, b'')
-        self.assertEqual(self.parser.state, ChunkParser.states.WAITING_FOR_SIZE)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.WAITING_FOR_SIZE)
         self.parser.parse(b'\r\n')
         self.assertEqual(self.parser.chunk, b'')
         self.assertEqual(self.parser.size, 3)
         self.assertEqual(self.parser.body, b'')
-        self.assertEqual(self.parser.state, ChunkParser.states.WAITING_FOR_DATA)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.WAITING_FOR_DATA)
         self.parser.parse(b'abc')
         self.assertEqual(self.parser.chunk, b'')
         self.assertEqual(self.parser.size, None)
         self.assertEqual(self.parser.body, b'abc')
-        self.assertEqual(self.parser.state, ChunkParser.states.WAITING_FOR_SIZE)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.WAITING_FOR_SIZE)
         self.parser.parse(b'\r\n')
         self.assertEqual(self.parser.chunk, b'')
         self.assertEqual(self.parser.size, None)
         self.assertEqual(self.parser.body, b'abc')
-        self.assertEqual(self.parser.state, ChunkParser.states.WAITING_FOR_SIZE)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.WAITING_FOR_SIZE)
         self.parser.parse(b'4\r\n')
         self.assertEqual(self.parser.chunk, b'')
         self.assertEqual(self.parser.size, 4)
         self.assertEqual(self.parser.body, b'abc')
-        self.assertEqual(self.parser.state, ChunkParser.states.WAITING_FOR_DATA)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.WAITING_FOR_DATA)
         self.parser.parse(b'defg\r\n0')
         self.assertEqual(self.parser.chunk, b'0')
         self.assertEqual(self.parser.size, None)
         self.assertEqual(self.parser.body, b'abcdefg')
-        self.assertEqual(self.parser.state, ChunkParser.states.WAITING_FOR_SIZE)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.WAITING_FOR_SIZE)
         self.parser.parse(b'\r\n\r\n')
         self.assertEqual(self.parser.chunk, b'')
         self.assertEqual(self.parser.size, None)
         self.assertEqual(self.parser.body, b'abcdefg')
-        self.assertEqual(self.parser.state, ChunkParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.ChunkParser.states.COMPLETE)
 
 
 class TestHttpParser(unittest.TestCase):
 
     def setUp(self):
-        self.parser = HttpParser(HttpParser.types.REQUEST_PARSER)
+        self.parser = proxy.HttpParser(proxy.HttpParser.types.REQUEST_PARSER)
 
     def test_build_header(self):
-        self.assertEqual(HttpParser.build_header(b'key', b'value'), b'key: value')
+        self.assertEqual(proxy.HttpParser.build_header(b'key', b'value'), b'key: value')
 
     def test_split(self):
-        self.assertEqual(HttpParser.split(b'CONNECT python.org:443 HTTP/1.0\r\n\r\n'),
+        self.assertEqual(proxy.HttpParser.split(b'CONNECT python.org:443 HTTP/1.0\r\n\r\n'),
                          (b'CONNECT python.org:443 HTTP/1.0', b'\r\n'))
 
     def test_split_false_line(self):
-        self.assertEqual(HttpParser.split(b'CONNECT python.org:443 HTTP/1.0'),
+        self.assertEqual(proxy.HttpParser.split(b'CONNECT python.org:443 HTTP/1.0'),
                          (False, b'CONNECT python.org:443 HTTP/1.0'))
 
     def test_get_full_parse(self):
-        raw = CRLF.join([
+        raw = proxy.CRLF.join([
             b'GET %s HTTP/1.1',
             b'Host: %s',
-            CRLF
+            proxy.CRLF
         ])
         self.parser.parse(raw % (b'https://example.com/path/dir/?a=b&c=d#p=q', b'example.com'))
         self.assertEqual(self.parser.build_url(), b'/path/dir/?a=b&c=d#p=q')
@@ -119,7 +236,7 @@ class TestHttpParser(unittest.TestCase):
         self.assertEqual(self.parser.url.hostname, b'example.com')
         self.assertEqual(self.parser.url.port, None)
         self.assertEqual(self.parser.version, b'HTTP/1.1')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
         self.assertDictContainsSubset({b'host': (b'Host', b'example.com')}, self.parser.headers)
         self.assertEqual(raw % (b'/path/dir/?a=b&c=d#p=q', b'example.com'),
                          self.parser.build(del_headers=[b'host'], add_headers=[(b'Host', b'example.com')]))
@@ -129,39 +246,39 @@ class TestHttpParser(unittest.TestCase):
 
     def test_line_rcvd_to_rcving_headers_state_change(self):
         self.parser.parse(b'GET http://localhost HTTP/1.1')
-        self.assertEqual(self.parser.state, HttpParser.states.INITIALIZED)
-        self.parser.parse(CRLF)
-        self.assertEqual(self.parser.state, HttpParser.states.LINE_RCVD)
-        self.parser.parse(CRLF)
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_HEADERS)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.INITIALIZED)
+        self.parser.parse(proxy.CRLF)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.LINE_RCVD)
+        self.parser.parse(proxy.CRLF)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_HEADERS)
 
     def test_get_partial_parse1(self):
-        self.parser.parse(CRLF.join([
+        self.parser.parse(proxy.CRLF.join([
             b'GET http://localhost:8080 HTTP/1.1'
         ]))
         self.assertEqual(self.parser.method, None)
         self.assertEqual(self.parser.url, None)
         self.assertEqual(self.parser.version, None)
-        self.assertEqual(self.parser.state, HttpParser.states.INITIALIZED)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.INITIALIZED)
 
-        self.parser.parse(CRLF)
+        self.parser.parse(proxy.CRLF)
         self.assertEqual(self.parser.method, b'GET')
         self.assertEqual(self.parser.url.hostname, b'localhost')
         self.assertEqual(self.parser.url.port, 8080)
         self.assertEqual(self.parser.version, b'HTTP/1.1')
-        self.assertEqual(self.parser.state, HttpParser.states.LINE_RCVD)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.LINE_RCVD)
 
         self.parser.parse(b'Host: localhost:8080')
         self.assertDictEqual(self.parser.headers, dict())
         self.assertEqual(self.parser.buffer, b'Host: localhost:8080')
-        self.assertEqual(self.parser.state, HttpParser.states.LINE_RCVD)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.LINE_RCVD)
 
-        self.parser.parse(CRLF * 2)
+        self.parser.parse(proxy.CRLF * 2)
         self.assertDictContainsSubset({b'host': (b'Host', b'localhost:8080')}, self.parser.headers)
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
     def test_get_partial_parse2(self):
-        self.parser.parse(CRLF.join([
+        self.parser.parse(proxy.CRLF.join([
             b'GET http://localhost:8080 HTTP/1.1',
             b'Host: '
         ]))
@@ -170,27 +287,27 @@ class TestHttpParser(unittest.TestCase):
         self.assertEqual(self.parser.url.port, 8080)
         self.assertEqual(self.parser.version, b'HTTP/1.1')
         self.assertEqual(self.parser.buffer, b'Host: ')
-        self.assertEqual(self.parser.state, HttpParser.states.LINE_RCVD)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.LINE_RCVD)
 
-        self.parser.parse(b'localhost:8080' + CRLF)
+        self.parser.parse(b'localhost:8080' + proxy.CRLF)
         self.assertDictContainsSubset({b'host': (b'Host', b'localhost:8080')}, self.parser.headers)
         self.assertEqual(self.parser.buffer, b'')
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_HEADERS)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_HEADERS)
 
-        self.parser.parse(b'Content-Type: text/plain' + CRLF)
+        self.parser.parse(b'Content-Type: text/plain' + proxy.CRLF)
         self.assertEqual(self.parser.buffer, b'')
         self.assertDictContainsSubset({b'content-type': (b'Content-Type', b'text/plain')}, self.parser.headers)
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_HEADERS)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_HEADERS)
 
-        self.parser.parse(CRLF)
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.parser.parse(proxy.CRLF)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
     def test_post_full_parse(self):
-        raw = CRLF.join([
+        raw = proxy.CRLF.join([
             b'POST %s HTTP/1.1',
             b'Host: localhost',
             b'Content-Length: 7',
-            b'Content-Type: application/x-www-form-urlencoded' + CRLF,
+            b'Content-Type: application/x-www-form-urlencoded' + proxy.CRLF,
             b'a=b&c=d'
         ])
         self.parser.parse(raw % b'http://localhost')
@@ -203,11 +320,11 @@ class TestHttpParser(unittest.TestCase):
         self.assertDictContainsSubset({b'content-length': (b'Content-Length', b'7')}, self.parser.headers)
         self.assertEqual(self.parser.body, b'a=b&c=d')
         self.assertEqual(self.parser.buffer, b'')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
         self.assertEqual(len(self.parser.build()), len(raw % b'/'))
 
     def test_post_partial_parse(self):
-        self.parser.parse(CRLF.join([
+        self.parser.parse(proxy.CRLF.join([
             b'POST http://localhost HTTP/1.1',
             b'Host: localhost',
             b'Content-Length: 7',
@@ -217,21 +334,21 @@ class TestHttpParser(unittest.TestCase):
         self.assertEqual(self.parser.url.hostname, b'localhost')
         self.assertEqual(self.parser.url.port, None)
         self.assertEqual(self.parser.version, b'HTTP/1.1')
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_HEADERS)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_HEADERS)
 
-        self.parser.parse(CRLF)
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_HEADERS)
+        self.parser.parse(proxy.CRLF)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_HEADERS)
 
-        self.parser.parse(CRLF)
-        self.assertEqual(self.parser.state, HttpParser.states.HEADERS_COMPLETE)
+        self.parser.parse(proxy.CRLF)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.HEADERS_COMPLETE)
 
         self.parser.parse(b'a=b')
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_BODY)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_BODY)
         self.assertEqual(self.parser.body, b'a=b')
         self.assertEqual(self.parser.buffer, b'')
 
         self.parser.parse(b'&c=d')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
         self.assertEqual(self.parser.body, b'a=b&c=d')
         self.assertEqual(self.parser.buffer, b'')
 
@@ -249,7 +366,7 @@ class TestHttpParser(unittest.TestCase):
         self.parser.parse(b'CONNECT pypi.org:443 HTTP/1.0\r\n\r\n')
         self.assertEqual(self.parser.method, b'CONNECT')
         self.assertEqual(self.parser.version, b'HTTP/1.0')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
     def test_request_parse_without_content_length(self):
         """Case when incoming request doesn't contain a content-length header.
@@ -260,14 +377,14 @@ class TestHttpParser(unittest.TestCase):
 
         See https://github.com/abhinavsingh/proxy.py/issues/20 for details.
         """
-        self.parser.parse(CRLF.join([
+        self.parser.parse(proxy.CRLF.join([
             b'POST http://localhost HTTP/1.1',
             b'Host: localhost',
             b'Content-Type: application/x-www-form-urlencoded',
-            CRLF
+            proxy.CRLF
         ]))
         self.assertEqual(self.parser.method, b'POST')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
     def test_response_parse_without_content_length(self):
         """Case when server response doesn't contain a content-length header for non-chunk response types.
@@ -279,20 +396,20 @@ class TestHttpParser(unittest.TestCase):
 
         See https://github.com/abhinavsingh/proxy.py/issues/20 for details.
         """
-        self.parser.type = HttpParser.types.RESPONSE_PARSER
-        self.parser.parse(b'HTTP/1.0 200 OK' + CRLF)
+        self.parser.type = proxy.HttpParser.types.RESPONSE_PARSER
+        self.parser.parse(b'HTTP/1.0 200 OK' + proxy.CRLF)
         self.assertEqual(self.parser.code, b'200')
         self.assertEqual(self.parser.version, b'HTTP/1.0')
-        self.assertEqual(self.parser.state, HttpParser.states.LINE_RCVD)
-        self.parser.parse(CRLF.join([
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.LINE_RCVD)
+        self.parser.parse(proxy.CRLF.join([
             b'Server: BaseHTTP/0.3 Python/2.7.10',
             b'Date: Thu, 13 Dec 2018 16:24:09 GMT',
-            CRLF
+            proxy.CRLF
         ]))
-        self.assertEqual(self.parser.state, HttpParser.states.HEADERS_COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.HEADERS_COMPLETE)
 
     def test_response_parse(self):
-        self.parser.type = HttpParser.types.RESPONSE_PARSER
+        self.parser.type = proxy.HttpParser.types.RESPONSE_PARSER
         self.parser.parse(b''.join([
             b'HTTP/1.1 301 Moved Permanently\r\n',
             b'Location: http://www.google.com/\r\n',
@@ -317,10 +434,10 @@ class TestHttpParser(unittest.TestCase):
                          b'<TITLE>301 Moved</TITLE></HEAD><BODY>\n<H1>301 Moved</H1>\nThe document has moved\n' +
                          b'<A HREF="http://www.google.com/">here</A>.\r\n</BODY></HTML>\r\n')
         self.assertDictContainsSubset({b'content-length': (b'Content-Length', b'219')}, self.parser.headers)
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
     def test_response_partial_parse(self):
-        self.parser.type = HttpParser.types.RESPONSE_PARSER
+        self.parser.type = proxy.HttpParser.types.RESPONSE_PARSER
         self.parser.parse(b''.join([
             b'HTTP/1.1 301 Moved Permanently\r\n',
             b'Location: http://www.google.com/\r\n',
@@ -334,20 +451,20 @@ class TestHttpParser(unittest.TestCase):
             b'X-Frame-Options: SAMEORIGIN\r\n'
         ]))
         self.assertDictContainsSubset({b'x-frame-options': (b'X-Frame-Options', b'SAMEORIGIN')}, self.parser.headers)
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_HEADERS)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_HEADERS)
         self.parser.parse(b'\r\n')
-        self.assertEqual(self.parser.state, HttpParser.states.HEADERS_COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.HEADERS_COMPLETE)
         self.parser.parse(
             b'<HTML><HEAD><meta http-equiv="content-type" content="text/html;charset=utf-8">\n' +
             b'<TITLE>301 Moved</TITLE></HEAD>')
-        self.assertEqual(self.parser.state, HttpParser.states.RCVING_BODY)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.RCVING_BODY)
         self.parser.parse(
             b'<BODY>\n<H1>301 Moved</H1>\nThe document has moved\n' +
             b'<A HREF="http://www.google.com/">here</A>.\r\n</BODY></HTML>\r\n')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
     def test_chunked_response_parse(self):
-        self.parser.type = HttpParser.types.RESPONSE_PARSER
+        self.parser.type = proxy.HttpParser.types.RESPONSE_PARSER
         self.parser.parse(b''.join([
             b'HTTP/1.1 200 OK\r\n',
             b'Content-Type: application/json\r\n',
@@ -365,10 +482,10 @@ class TestHttpParser(unittest.TestCase):
             b'\r\n'
         ]))
         self.assertEqual(self.parser.body, b'Wikipedia in\r\n\r\nchunks.')
-        self.assertEqual(self.parser.state, HttpParser.states.COMPLETE)
+        self.assertEqual(self.parser.state, proxy.HttpParser.states.COMPLETE)
 
 
-class MockConnection(object):
+class MockTcpConnection(object):
 
     def __init__(self, b=b''):
         self.buffer = b
@@ -396,25 +513,20 @@ class HTTPRequestHandler(BaseHTTPRequestHandler):
 
 
 class TestProxy(unittest.TestCase):
-
     http_server = None
     http_server_port = None
     http_server_thread = None
-
-    @staticmethod
-    def get_available_port():
-        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
-            sock.bind(('', 0))
-            _, port = sock.getsockname()
-            return port
+    config = None
 
     @classmethod
     def setUpClass(cls):
-        cls.http_server_port = cls.get_available_port()
+        cls.http_server_port = get_available_port()
         cls.http_server = HTTPServer(('127.0.0.1', cls.http_server_port), HTTPRequestHandler)
         cls.http_server_thread = Thread(target=cls.http_server.serve_forever)
         cls.http_server_thread.setDaemon(True)
         cls.http_server_thread.start()
+        cls.config = proxy.HttpProtocolConfig()
+        cls.config.plugins = proxy.load_plugins('proxy.HttpProxyPlugin,proxy.HttpWebServerPlugin')
 
     @classmethod
     def tearDownClass(cls):
@@ -423,207 +535,258 @@ class TestProxy(unittest.TestCase):
         cls.http_server_thread.join()
 
     def setUp(self):
-        self._conn = MockConnection()
+        self._conn = MockTcpConnection()
         self._addr = ('127.0.0.1', 54382)
-        self.proxy = HTTPProxy(TCPClientConnection(self._conn, self._addr))
+        self.proxy = proxy.HttpProtocolHandler(proxy.TcpClientConnection(self._conn, self._addr), config=self.config)
+
+    @mock.patch('select.select')
+    @mock.patch('proxy.TcpServerConnection')
+    def test_http_get(self, mock_server_connection, mock_select):
+        server = mock_server_connection.return_value
+        server.connect.return_value = True
+        mock_select.side_effect = [([self._conn], [], []), ([self._conn], [], []), ([], [server.conn], [])]
 
-    def test_http_get(self):
         # Send request line
-        self.proxy.client.conn.queue((b'GET http://localhost:%d HTTP/1.1' % self.http_server_port) + CRLF)
-        self.proxy._process_request(self.proxy.client.recv())
-        self.assertNotEqual(self.proxy.request.state, HttpParser.states.COMPLETE)
+        self.proxy.client.conn.queue((b'GET http://localhost:%d HTTP/1.1' % self.http_server_port) + proxy.CRLF)
+        self.proxy.run_once()
+        self.assertEqual(self.proxy.request.state, proxy.HttpParser.states.LINE_RCVD)
+        self.assertNotEqual(self.proxy.request.state, proxy.HttpParser.states.COMPLETE)
+
         # Send headers and blank line, thus completing HTTP request
-        self.proxy.client.conn.queue(CRLF.join([
-            b'User-Agent: proxy.py/%s' % version,
+        self.proxy.client.conn.queue(proxy.CRLF.join([
+            b'User-Agent: proxy.py/%s' % proxy.version,
             b'Host: localhost:%d' % self.http_server_port,
             b'Accept: */*',
             b'Proxy-Connection: Keep-Alive',
-            CRLF
+            proxy.CRLF
         ]))
-        self.proxy._process_request(self.proxy.client.recv())
-        self.assertEqual(self.proxy.request.state, HttpParser.states.COMPLETE)
-        self.assertEqual(self.proxy.server.addr, (b'localhost', self.http_server_port))
-        # Flush data queued for server
-        self.proxy.server.flush()
-        self.assertEqual(self.proxy.server.buffer_size(), 0)
-        # Receive full response from server
-        data = self.proxy.server.recv()
-        while data:
-            self.proxy._process_response(data)
-            logging.info(self.proxy.response.state)
-            if self.proxy.response.state == HttpParser.states.COMPLETE:
-                break
-            data = self.proxy.server.recv()
-        # Verify 200 success response code
-        self.assertEqual(self.proxy.response.state, HttpParser.states.COMPLETE)
-        self.assertEqual(int(self.proxy.response.code), 200)
+        self.proxy.run_once()
+        self.assertEqual(self.proxy.request.state, proxy.HttpParser.states.COMPLETE)
+        mock_server_connection.assert_called_once()
+        server.connect.assert_called_once()
+        server.closed = False
+        server.queue.assert_called_once_with(proxy.CRLF.join([
+            b'GET / HTTP/1.1',
+            b'User-Agent: proxy.py/%s' % proxy.version,
+            b'Host: localhost:%d' % self.http_server_port,
+            b'Accept: */*',
+            b'Via: %s' % b'1.1 proxy.py v%s' % proxy.version,
+            b'Connection: Close',
+            proxy.CRLF
+        ]))
+
+        self.proxy.run_once()
+        server.flush.assert_called_once()
+
+    @mock.patch('select.select')
+    @mock.patch('proxy.TcpServerConnection')
+    def test_http_tunnel(self, mock_server_connection, mock_select):
+        server = mock_server_connection.return_value
+        server.connect.return_value = True
+        server.has_buffer.side_effect = [False, False, False, True]
+        mock_select.side_effect = [([self._conn], [], []), ([], [self._conn], []),
+                                   ([self._conn], [], []), ([], [server.conn], [])]
 
-    def test_http_tunnel(self):
-        self.proxy.client.conn.queue(CRLF.join([
+        self.proxy.client.conn.queue(proxy.CRLF.join([
             b'CONNECT localhost:%d HTTP/1.1' % self.http_server_port,
             b'Host: localhost:%d' % self.http_server_port,
-            b'User-Agent: proxy.py/%s' % version,
+            b'User-Agent: proxy.py/%s' % proxy.version,
             b'Proxy-Connection: Keep-Alive',
-            CRLF
+            proxy.CRLF
         ]))
-        self.proxy._process_request(self.proxy.client.recv())
-        self.assertFalse(self.proxy.server is None)
-        self.assertEqual(self.proxy.client.buffer, PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
-
-        parser = HttpParser(HttpParser.types.RESPONSE_PARSER)
+        self.proxy.run_once()
+        self.assertFalse(self.proxy.plugins['HttpProxyPlugin'].server is None)
+        self.assertEqual(self.proxy.client.buffer, proxy.HttpProxyPlugin.PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
+        mock_server_connection.assert_called_once()
+        server.connect.assert_called_once()
+        server.queue.assert_not_called()
+        server.closed = False
+
+        parser = proxy.HttpParser(proxy.HttpParser.types.RESPONSE_PARSER)
         parser.parse(self.proxy.client.buffer)
-        self.assertEqual(parser.state, HttpParser.states.HEADERS_COMPLETE)
+        self.assertEqual(parser.state, proxy.HttpParser.states.HEADERS_COMPLETE)
         self.assertEqual(int(parser.code), 200)
 
-        self.proxy.client.flush()
+        # Dispatch tunnel established response to client
+        self.proxy.run_once()
         self.assertEqual(self.proxy.client.buffer_size(), 0)
 
-        self.proxy.client.conn.queue(CRLF.join([
+        self.proxy.client.conn.queue(proxy.CRLF.join([
             b'GET / HTTP/1.1',
             b'Host: localhost:%d' % self.http_server_port,
-            b'User-Agent: proxy.py/%s' % version,
-            CRLF
+            b'User-Agent: proxy.py/%s' % proxy.version,
+            proxy.CRLF
         ]))
-        self.proxy._process_request(self.proxy.client.recv())
-        self.proxy.server.flush()
-        self.assertEqual(self.proxy.server.buffer_size(), 0)
-
-        parser = HttpParser(HttpParser.types.RESPONSE_PARSER)
-        data = self.proxy.server.recv()
-        while data:
-            parser.parse(data)
-            if parser.state == HttpParser.states.COMPLETE:
-                break
-            data = self.proxy.server.recv()
-
-        self.assertEqual(parser.state, HttpParser.states.COMPLETE)
-        self.assertEqual(int(parser.code), 200)
-
-    def test_proxy_connection_failed(self):
-        with self.assertRaises(ProxyConnectionFailed):
-            self.proxy._process_request(CRLF.join([
-                b'GET http://unknown.domain HTTP/1.1',
-                b'Host: unknown.domain',
-                CRLF
-            ]))
-
-    def test_proxy_authentication_failed(self):
-        self.proxy = HTTPProxy(TCPClientConnection(self._conn, self._addr), b'Basic %s' % base64.b64encode(b'user:pass'))
-
-        with self.assertRaises(ProxyAuthenticationFailed):
-            self.proxy._process_request(CRLF.join([
-                b'GET http://abhinavsingh.com HTTP/1.1',
-                b'Host: abhinavsingh.com',
-                CRLF
-            ]))
-
-    def test_authenticated_proxy_http_get(self):
-        self.proxy = HTTPProxy(TCPClientConnection(self._conn, self._addr), b'Basic %s' % base64.b64encode(b'user:pass'))
-
-        self.proxy.client.conn.queue((b'GET http://localhost:%d HTTP/1.1' % self.http_server_port) + CRLF)
-        self.proxy._process_request(self.proxy.client.recv())
-        self.assertNotEqual(self.proxy.request.state, HttpParser.states.COMPLETE)
-
-        self.proxy.client.conn.queue(CRLF.join([
-            b'User-Agent: proxy.py/%s' % version,
+        self.proxy.run_once()
+        server.queue.assert_called_once_with(proxy.CRLF.join([
+            b'GET / HTTP/1.1',
+            b'Host: localhost:%d' % self.http_server_port,
+            b'User-Agent: proxy.py/%s' % proxy.version,
+            proxy.CRLF
+        ]))
+        server.flush.assert_not_called()
+
+        self.proxy.run_once()
+        self.assertEqual(server.queue.call_count, 1)
+        server.flush.assert_called_once()
+
+    @mock.patch('select.select')
+    def test_proxy_connection_failed(self, mock_select):
+        mock_select.return_value = ([self._conn], [], [])
+        self.proxy.client.conn.queue(proxy.CRLF.join([
+            b'GET http://unknown.domain HTTP/1.1',
+            b'Host: unknown.domain',
+            proxy.CRLF
+        ]))
+        with self.assertRaises(proxy.ProxyConnectionFailed):
+            self.proxy.run_once()
+
+    @mock.patch('select.select')
+    def test_proxy_authentication_failed(self, mock_select):
+        mock_select.return_value = ([self._conn], [], [])
+        config = proxy.HttpProtocolConfig(auth_code=b'Basic %s' % base64.b64encode(b'user:pass'))
+        config.plugins = proxy.load_plugins('proxy.HttpProxyPlugin,proxy.HttpWebServerPlugin')
+        self.proxy = proxy.HttpProtocolHandler(proxy.TcpClientConnection(self._conn, self._addr),
+                                               config=config)
+        self.proxy.client.conn.queue(proxy.CRLF.join([
+            b'GET http://abhinavsingh.com HTTP/1.1',
+            b'Host: abhinavsingh.com',
+            proxy.CRLF
+        ]))
+        with self.assertRaises(proxy.ProxyAuthenticationFailed):
+            self.proxy.run_once()
+
+    @mock.patch('select.select')
+    @mock.patch('proxy.TcpServerConnection')
+    def test_authenticated_proxy_http_get(self, mock_server_connection, mock_select):
+        mock_select.return_value = ([self._conn], [], [])
+        server = mock_server_connection.return_value
+        server.connect.return_value = True
+
+        client = proxy.TcpClientConnection(self._conn, self._addr)
+        config = proxy.HttpProtocolConfig(auth_code=b'Basic %s' % base64.b64encode(b'user:pass'))
+        config.plugins = proxy.load_plugins('proxy.HttpProxyPlugin,proxy.HttpWebServerPlugin')
+
+        self.proxy = proxy.HttpProtocolHandler(client, config=config)
+        self.proxy.client.conn.queue(b'GET http://localhost:%d HTTP/1.1' % self.http_server_port)
+        self.proxy.run_once()
+        self.assertEqual(self.proxy.request.state, proxy.HttpParser.states.INITIALIZED)
+
+        self.proxy.client.conn.queue(proxy.CRLF)
+        self.proxy.run_once()
+        self.assertEqual(self.proxy.request.state, proxy.HttpParser.states.LINE_RCVD)
+
+        self.proxy.client.conn.queue(proxy.CRLF.join([
+            b'User-Agent: proxy.py/%s' % proxy.version,
             b'Host: localhost:%d' % self.http_server_port,
             b'Accept: */*',
             b'Proxy-Connection: Keep-Alive',
             b'Proxy-Authorization: Basic dXNlcjpwYXNz',
-            CRLF
+            proxy.CRLF
         ]))
+        self.proxy.run_once()
+        self.assertEqual(self.proxy.request.state, proxy.HttpParser.states.COMPLETE)
+        mock_server_connection.assert_called_once()
+        server.connect.assert_called_once()
+        server.closed = False
 
-        self.proxy._process_request(self.proxy.client.recv())
-        self.assertEqual(self.proxy.request.state, HttpParser.states.COMPLETE)
-        self.assertEqual(self.proxy.server.addr, (b'localhost', self.http_server_port))
-
-        self.proxy.server.flush()
-        self.assertEqual(self.proxy.server.buffer_size(), 0)
-
-        data = self.proxy.server.recv()
-        while data:
-            self.proxy._process_response(data)
-            if self.proxy.response.state == HttpParser.states.COMPLETE:
-                break
-            data = self.proxy.server.recv()
-
-        self.assertEqual(self.proxy.response.state, HttpParser.states.COMPLETE)
-        self.assertEqual(int(self.proxy.response.code), 200)
-
-    def test_authenticated_proxy_http_tunnel(self):
-        self.proxy = HTTPProxy(TCPClientConnection(self._conn, self._addr), b'Basic %s' % base64.b64encode(b'user:pass'))
+        server.queue.assert_called_once_with(proxy.CRLF.join([
+            b'GET / HTTP/1.1',
+            b'User-Agent: proxy.py/%s' % proxy.version,
+            b'Host: localhost:%d' % self.http_server_port,
+            b'Accept: */*',
+            b'Via: %s' % b'1.1 proxy.py v%s' % proxy.version,
+            b'Connection: Close',
+            proxy.CRLF
+        ]))
 
-        self.proxy.client.conn.queue(CRLF.join([
+    @mock.patch('select.select')
+    @mock.patch('proxy.TcpServerConnection')
+    def test_authenticated_proxy_http_tunnel(self, mock_server_connection, mock_select):
+        server = mock_server_connection.return_value
+        server.connect.return_value = True
+        mock_select.side_effect = [([self._conn], [], []), ([self._conn], [], []), ([], [server.conn], [])]
+
+        config = proxy.HttpProtocolConfig(auth_code=b'Basic %s' % base64.b64encode(b'user:pass'))
+        config.plugins = proxy.load_plugins('proxy.HttpProxyPlugin,proxy.HttpWebServerPlugin')
+        self.proxy = proxy.HttpProtocolHandler(proxy.TcpClientConnection(self._conn, self._addr),
+                                               config=config)
+        self.proxy.client.conn.queue(proxy.CRLF.join([
             b'CONNECT localhost:%d HTTP/1.1' % self.http_server_port,
             b'Host: localhost:%d' % self.http_server_port,
-            b'User-Agent: proxy.py/%s' % version,
+            b'User-Agent: proxy.py/%s' % proxy.version,
             b'Proxy-Connection: Keep-Alive',
             b'Proxy-Authorization: Basic dXNlcjpwYXNz',
-            CRLF
+            proxy.CRLF
         ]))
-        self.proxy._process_request(self.proxy.client.recv())
-        self.assertFalse(self.proxy.server is None)
-        self.assertEqual(self.proxy.client.buffer, PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
-
-        parser = HttpParser(HttpParser.types.RESPONSE_PARSER)
+        self.proxy.run_once()
+        self.assertFalse(self.proxy.plugins['HttpProxyPlugin'].server is None)
+        self.assertEqual(self.proxy.client.buffer, proxy.HttpProxyPlugin.PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT)
+        mock_server_connection.assert_called_once()
+        server.connect.assert_called_once()
+        server.closed = False
+
+        parser = proxy.HttpParser(proxy.HttpParser.types.RESPONSE_PARSER)
         parser.parse(self.proxy.client.buffer)
-        self.assertEqual(parser.state, HttpParser.states.HEADERS_COMPLETE)
+        self.assertEqual(parser.state, proxy.HttpParser.states.HEADERS_COMPLETE)
         self.assertEqual(int(parser.code), 200)
-
         self.proxy.client.flush()
         self.assertEqual(self.proxy.client.buffer_size(), 0)
 
-        self.proxy.client.conn.queue(CRLF.join([
+        self.proxy.client.conn.queue(proxy.CRLF.join([
             b'GET / HTTP/1.1',
             b'Host: localhost:%d' % self.http_server_port,
-            b'User-Agent: proxy.py/%s' % version,
-            CRLF
+            b'User-Agent: proxy.py/%s' % proxy.version,
+            proxy.CRLF
+        ]))
+        self.proxy.run_once()
+        server.queue.assert_called_once_with(proxy.CRLF.join([
+            b'GET / HTTP/1.1',
+            b'Host: localhost:%d' % self.http_server_port,
+            b'User-Agent: proxy.py/%s' % proxy.version,
+            proxy.CRLF
         ]))
-        self.proxy._process_request(self.proxy.client.recv())
-        self.proxy.server.flush()
-        self.assertEqual(self.proxy.server.buffer_size(), 0)
-
-        parser = HttpParser(HttpParser.types.RESPONSE_PARSER)
-        data = self.proxy.server.recv()
-        while data:
-            parser.parse(data)
-            if parser.state == HttpParser.states.COMPLETE:
-                break
-            data = self.proxy.server.recv()
 
-        self.assertEqual(parser.state, HttpParser.states.COMPLETE)
-        self.assertEqual(int(parser.code), 200)
+        self.proxy.run_once()
+        server.flush.assert_called_once()
 
 
 class TestWorker(unittest.TestCase):
 
     def setUp(self):
         self.queue = multiprocessing.Queue()
-        self.worker = Worker(self.queue)
+        self.worker = proxy.Worker(self.queue)
 
-    @patch('proxy.HTTPProxy')
+    @mock.patch('proxy.HttpProtocolHandler')
     def test_shutdown_op(self, mock_http_proxy):
-        self.queue.put((Worker.operations.SHUTDOWN, None))
-        self.worker.run()   # Worker should consume the prior shutdown operation
+        self.queue.put((proxy.Worker.operations.SHUTDOWN, None))
+        self.worker.run()  # Worker should consume the prior shutdown operation
         self.assertFalse(mock_http_proxy.called)
 
-    @patch('proxy.HTTPProxy')
+    @mock.patch('proxy.HttpProtocolHandler')
     def test_spawns_http_proxy_threads(self, mock_http_proxy):
-        self.queue.put((Worker.operations.DEFAULT, None))
-        self.queue.put((Worker.operations.SHUTDOWN, None))
+        self.queue.put((proxy.Worker.operations.HTTP_PROTOCOL, None))
+        self.queue.put((proxy.Worker.operations.SHUTDOWN, None))
         self.worker.run()
         self.assertTrue(mock_http_proxy.called)
 
 
 class TestMain(unittest.TestCase):
 
-    @patch('proxy.set_open_file_limit')
-    @patch('proxy.HTTPServer')
+    @mock.patch('proxy.set_open_file_limit')
+    @mock.patch('proxy.MultiCoreRequestDispatcher')
     def test_http_server_called(self, mock_set_open_file_limit, mock_http_server):
-        main()
+        proxy.main()
         self.assertTrue(mock_set_open_file_limit.called)
         self.assertTrue(mock_http_server.called)
 
+    def test_text(self):
+        self.assertEqual(proxy.text_(b'hello'), 'hello')
+
+    def test_bytes(self):
+        self.assertEqual(proxy.bytes_('hello'), b'hello')
+
 
 if __name__ == '__main__':
+    proxy.UNDER_TEST = True
     unittest.main()