Repository: abhinavsingh/proxy.py

Issue:
Use select.epoll to bypass FD_SETSIZE limit imposed by select.select usage
Hello, I'm new in python and I am using this proxy but I have a problem, when lots of requests are being performed the proxy stops working because of the file   @#descriptor error (**FD_SETSIZE** limit).

I was reading what this problem only happens with select.select() and a workaround or solution may be using select.epoll() instead of select.select() but I didn't know how to do this correctly.

Thanks

Discussion:

Patch:
diff --git a/.github/workflows/python-package.yml b/.github/workflows/testing.yml
similarity index 51%
rename from .github/workflows/python-package.yml
rename to .github/workflows/testing.yml
index 13b07dc..6bab40e 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/testing.yml
@@ -4,19 +4,20 @@ on: [push]
 
 jobs:
   build:
-    runs-on: ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}-latest
+    name: Python ${{ matrix.python }} on ${{ matrix.os }}
     strategy:
       matrix:
-        os: [macOS-latest, ubuntu-latest, windows-latest]
+        os: [macOS, ubuntu, windows]
         python: [3.6, 3.7]
       max-parallel: 4
       fail-fast: false
     steps:
       - uses: actions/checkout@v1
-      - name: Set up Python
+      - name: Setup Python
         uses: actions/setup-python@v1
         with:
-          python-version: ${{ matrix.python }}
+          python-version: ${{ matrix.python }}-dev
           architecture: x64
       - name: Install dependencies
         run: |
@@ -24,12 +25,10 @@ jobs:
           pip install -r requirements-testing.txt
       - name: Quality Check
         run: |
-          # stop the build if there are Python syntax errors or undefined names
-          flake8 proxy.py tests.py --count --select=E9,F63,F7,F82 --show-source --statistics
-          # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-          flake8 proxy.py tests.py --count --exit-zero --max-line-length=127 --statistics
+          # The GitHub editor is 127 chars wide
+          # W504 screams for line break after binary operators
+          flake8 --ignore=W504 --max-line-length=127 proxy.py tests.py
           # mypy compliance check
           mypy --strict --ignore-missing-imports proxy.py plugin_examples.py tests.py
       - name: Run Tests
-        run: |
-          pytest tests.py
+        run: pytest tests.py
diff --git a/.travis.yml b/.travis.yml
index 8fbe904..ed96ea4 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -21,6 +21,6 @@ matrix:
         - TESTING_ON_TRAVIS=1
 install:
   - pip3 install -r requirements-testing.txt
-script: python3 -m coverage run tests.py || python -m coverage run tests.py
+script: python3 -m coverage run --source=proxy tests.py || python -m coverage run --source=proxy tests.py
 after_success:
   - coveralls
diff --git a/Makefile b/Makefile
index d6dfe5a..4a7d4e6 100644
--- a/Makefile
+++ b/Makefile
@@ -38,14 +38,13 @@ release: package
 	twine upload dist/*
 
 coverage:
-	coverage3 run tests.py
+	coverage3 run --source=proxy tests.py
 	coverage3 html
 	open htmlcov/index.html
 
 lint:
+	flake8 --ignore=W504 --max-line-length=127 proxy.py tests.py
 	mypy --strict --ignore-missing-imports proxy.py plugin_examples.py tests.py
-	flake8 --ignore=E501,W504 proxy.py
-	flake8 --ignore=E501,W504 tests.py
 
 autopep8:
 	autopep8 --recursive --in-place --aggressive proxy.py
diff --git a/README.md b/README.md
index 7994466..b8e19b3 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,4 @@
-[![Proxy.Py](ProxyPy.png)](https://github.com/abhinavsingh/proxy.py)
+[![Proxy.Py](https://raw.githubusercontent.com/abhinavsingh/proxy.py/develop/ProxyPy.png)](https://github.com/abhinavsingh/proxy.py)
 
 [![License](https://img.shields.io/github/license/abhinavsingh/proxy.py.svg)](https://opensource.org/licenses/BSD-3-Clause)
 [![PyPi Downloads](https://img.shields.io/pypi/dm/proxy.py.svg?color=green)](https://pypi.org/project/proxy.py/)
@@ -7,24 +7,25 @@
 [![No Dependencies](https://img.shields.io/static/v1?label=dependencies&message=none&color=green)](https://github.com/abhinavsingh/proxy.py)
 [![Coverage](https://coveralls.io/repos/github/abhinavsingh/proxy.py/badge.svg?branch=develop)](https://coveralls.io/github/abhinavsingh/proxy.py?branch=develop)
 
+[![Tested With MacOS](https://img.shields.io/static/v1?label=tested%20with&message=mac%20OS%20%F0%9F%92%BB&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
+[![Tested With Ubuntu](https://img.shields.io/static/v1?label=tested%20with&message=Ubuntu%20%F0%9F%96%A5&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
+[![Tested With Windows](https://img.shields.io/static/v1?label=tested%20with&message=Windows%20%F0%9F%92%BB&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
 [![Tested With Android](https://img.shields.io/static/v1?label=tested%20with&message=Android%20%F0%9F%93%B1&color=brightgreen)](https://www.android.com/)
 [![Tested With Android Emulator](https://img.shields.io/static/v1?label=tested%20with&message=Android%20Emulator%20%F0%9F%93%B1&color=brightgreen)](https://developer.android.com/studio/run/emulator-networking.html#proxy)
 [![Tested With iOS](https://img.shields.io/static/v1?label=tested%20with&message=iOS%20%F0%9F%93%B1&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
 [![Tested With iOS Simulator](https://img.shields.io/static/v1?label=tested%20with&message=iOS%20Simulator%20%F0%9F%93%B1&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
-[![Tested With MacOS](https://img.shields.io/static/v1?label=tested%20with&message=mac%20OS%20%F0%9F%92%BB&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
-[![Tested With Ubuntu](https://img.shields.io/static/v1?label=tested%20with&message=Ubuntu%20%F0%9F%96%A5&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
-[![Tested With Windows](https://img.shields.io/static/v1?label=tested%20with&message=Windows%20%F0%9F%92%BB&color=brightgreen)](https://developer.apple.com/library/archive/documentation/IDEs/Conceptual/iOS_Simulator_Guide/Introduction/Introduction.html)
 
-[![Gitter](https://badges.gitter.im/proxy-py/community.svg)](https://gitter.im/proxy-py/community)
 [![Maintenance](https://img.shields.io/static/v1?label=maintained%3F&message=yes&color=green)](https://gitHub.com/abhinavsingh/proxy.py/graphs/commit-activity)
 [![Ask Me Anything](https://img.shields.io/static/v1?label=need%20help%3F&message=ask&color=green)](https://twitter.com/imoracle)
 [![Contributions Welcome](https://img.shields.io/static/v1?label=contributions&message=welcome%20%F0%9F%91%8D&color=green)](https://github.com/abhinavsingh/proxy.py/issues)
+[![Gitter](https://badges.gitter.im/proxy-py/community.svg)](https://gitter.im/proxy-py/community)
 
-[![Python 3.5](https://img.shields.io/badge/python-3.5-blue.svg)](https://www.python.org/downloads/release/python-350/)
 [![Python 3.6](https://img.shields.io/badge/python-3.6-blue.svg)](https://www.python.org/downloads/release/python-360/)
 [![Python 3.7](https://img.shields.io/badge/python-3.7-blue.svg)](https://www.python.org/downloads/release/python-370/)
 [![Checked with mypy](https://img.shields.io/static/v1?label=mypy&message=checked&color=blue)](http://mypy-lang.org/)
 
+[![Become a Backer](https://opencollective.com/proxypy/tiers/backer.svg?avatarHeight=36)](https://opencollective.com/proxypy)
+
 Table of Contents
 =================
 
@@ -44,10 +45,16 @@ Table of Contents
     * [Plugin Ordering](#plugin-ordering)
 * [End-to-End Encryption](#end-to-end-encryption)
 * [TLS Interception](#tls-interception)
+* [import proxy.py](#import-proxypy)
 * [Plugin Developer and Contributor Guide](#plugin-developer-and-contributor-guide)
     * [Everything is a plugin](#everything-is-a-plugin)
-    * [proxy.py Internals](#proxypy-internals)
-    * [Pull Request](#pull-request)
+    * [Internal Architecture](#internal-architecture)
+    * [Internal Documentation](#internal-documentation)
+    * [Sending a Pull Request](#sending-a-pull-request)
+* [Frequently Asked Questions](#frequently-asked-questions)
+    * [Unable to connect with proxy.py from remote host](#unable-to-connect-with-proxypy-from-remote-host)
+    * [Basic auth not working with a browser](#basic-auth-not-working-with-a-browser)
+    * [Docker image not working on MacOS](#docker-image-not-working-on-macos)
 * [Flags](#flags)
 
 Features
@@ -501,6 +508,42 @@ cached file instead of plain text.
 Now use CA flags other 
 [plugin examples](#plugin-examples) to make them work for `https` traffic.
 
+import proxy.py
+===============
+
+You can directly import `proxy.py` into your `Python` code.  Example:
+
+```
+$ python
+>>> import proxy
+>>>
+>>> # Generate HTTP GET request
+>>> proxy.build_http_request(b'GET', b'/')
+b'GET / HTTP/1.1\r\n\r\n'
+>>>
+>>> # Generate HTTP GET request with headers
+>>> proxy.build_http_request(b'GET', b'/', 
+        headers={b'Connection': b'close'})
+b'GET / HTTP/1.1\r\nConnection: close\r\n\r\n'
+>>>
+>>> # Generate HTTP POST request with headers and body
+>>> import json
+>>> proxy.build_http_request(b'POST', b'/form', 
+        headers={b'Content-type': b'application/json'}, 
+        body=proxy.bytes_(json.dumps({'email': 'hello@world.com'})))
+    b'POST /form HTTP/1.1\r\nContent-type: application/json\r\n\r\n{"email": "hello@world.com"}'
+```
+
+To start `proxy.py` server from imported `proxy.py` module, simply do:
+
+```
+import sys
+import proxy
+
+if __name__ == '__main__':
+  proxy.main(sys.argv[1:])
+```
+
 Plugin Developer and Contributor Guide
 ======================================
 
@@ -528,7 +571,7 @@ As you might have guessed by now, in `proxy.py` everything is a plugin.
   http(s) server. [HttpProxyPlugin](https://github.com/abhinavsingh/proxy.py/blob/b03629fa0df1595eb4995427bc601063be7fdca9/proxy.py#L941-L1182) 
   also implements `ProtocolHandlerPlugin`.
 
-## proxy.py Internals
+## Internal Architecture
 
 - [ProtocolHandler](https://github.com/abhinavsingh/proxy.py/blob/b03629fa0df1595eb4995427bc601063be7fdca9/proxy.py#L1263-L1440) 
 thread is started with the accepted [TcpClientConnection](https://github.com/abhinavsingh/proxy.py/blob/b03629fa0df1595eb4995427bc601063be7fdca9/proxy.py#L230-L237).
@@ -544,28 +587,120 @@ and invoke `HttpProxyBasePlugin` lifecycle hooks.
   processes.
 
 - `--num-workers` `Worker` processes are started by 
-  [WorkerPool](https://github.com/abhinavsingh/proxy.py/blob/b03629fa0df1595eb4995427bc601063be7fdca9/proxy.py#L368-L421) 
-  on start-up.  `Worker` processes receives `TcpClientConnection` over a pipe from `WorkerPool`.
+  [AcceptorPool](https://github.com/abhinavsingh/proxy.py/blob/b03629fa0df1595eb4995427bc601063be7fdca9/proxy.py#L368-L421) 
+  on start-up.
+
+- `AcceptorPool` listens on server socket and pass the handler to `Worker` processes.
+  Workers are responsible for accepting new client connections and starting
+  `ProtocolHandler` thread.
 
-- `WorkerPool` implements [TcpServer](https://github.com/abhinavsingh/proxy.py/blob/b03629fa0df1595eb4995427bc601063be7fdca9/proxy.py#L240-L302) 
-  abstract class. `TcpServer` accepts `TcpClientConnection`. `WorkerPool` 
-  ensures full utilization of available CPU cores, for which it dispatches 
-  accepted `TcpClientConnection` to `Worker` processes in a round-robin fashion.
+## Sending a Pull Request
 
-## Pull Request
+Install dependencies for local development testing:
+
+`$ pip install -r requirements-testing.txt`
 
 Every pull request goes through set of tests which must pass:
 
 - `mypy`: Run `make lint` locally for compliance check. 
   Fix all warnings and errors before sending out a PR.
 
-- `coverage`: Run `make coverage` for coverage report.
+- `coverage`: Run `make coverage` locally for coverage report.
   Its ideal to add tests for any critical change. Depending upon
   the change, it's ok if test coverage falls by `<0.5%`.
 
 - `formatting`: Run `make autopep8` locally to format the code in-place.
   `autopep8` is run with `--aggresive` flag.  Sometimes it _may_ result in
-  weird formatting.
+  weird formatting.  But let's stick to one consistent formatting tool.
+  I am open to flag changes for `autopep8`.
+
+## Internal Documentation
+
+Browse through internal class hierarchy and documentation using `pydoc3`.
+Example:
+
+```
+$ pydoc3 proxy
+Help on module proxy:
+
+NAME
+    proxy
+
+DESCRIPTION
+    proxy.py
+    ~~~~~~~~
+    Lightweight, Programmable, TLS interceptor Proxy for HTTP(S), HTTP2, WebSockets protocols in a single Python file.
+    
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+
+CLASSES
+    abc.ABC(builtins.object)
+        HttpProxyBasePlugin
+        HttpWebServerRoutePlugin
+            HttpWebServerPacFilePlugin
+        ProtocolHandlerPlugin
+            HttpProxyPlugin
+            HttpWebServerPlugin
+        TcpConnection
+            TcpClientConnection
+            TcpServerConnection
+    builtins.Exception(builtins.BaseException)
+        ProtocolException
+            HttpRequestRejected
+            ProxyAuthenticationFailed
+            ProxyConnectionFailed
+        TcpConnectionUninitializedException
+    builtins.object
+        AcceptorPool
+        ChunkParser
+        HttpParser
+        ProtocolConfig
+        Websocket
+        WebsocketFrame
+    builtins.tuple(builtins.object)
+        ChunkParserStates
+        HttpParserStates
+        HttpParserTypes
+        HttpProtocolTypes
+        TcpConnectionTypes
+    multiprocessing.context.Process(multiprocessing.process.BaseProcess)
+        Worker
+    threading.Thread(builtins.object)
+        ProtocolHandler
+... [ truncated ] ...
+```
+
+Frequently Asked Questions
+==========================
+
+## Unable to connect with proxy.py from remote host
+
+Make sure `proxy.py` is listening on correct network interface.
+Try following flags:
+
+- For IPv6 `--hostname ::`
+- For IPv4 `--hostname 0.0.0.0`
+
+## Basic auth not working with a browser
+
+Most likely it's a browser integration issue with system keychain.
+
+- First verify that basic auth is working using `curl`
+
+    `curl -v -x username:password@localhost:8899 https://httpbin.org/get`
+
+- See [this thread](https://github.com/abhinavsingh/proxy.py/issues/89#issuecomment-534845710)
+  for further details.
+
+## Docker image not working on macOS
+
+It's a compatibility issue with `vpnkit`.
+
+See [moby/vpnkit exhausts docker resources](https://github.com/abhinavsingh/proxy.py/issues/43)
+and [Connection refused: The proxy could not connect](https://github.com/moby/vpnkit/issues/469)
+for some background.
+
 
 Flags
 =====
diff --git a/fluentd.conf b/fluentd.conf
new file mode 100644
index 0000000..13e6ed3
--- /dev/null
+++ b/fluentd.conf
@@ -0,0 +1,21 @@
+# google-fluentd (Stackdriver) log input configuration file
+#
+# 1. Copy this configuration file as proxy.py.conf under:
+#       /etc/google-fluentd/config.d/
+# 2. Update `path` field to log file path as used with
+#    --log-file flag.  By default `/tmp/proxy.log` path is tailed.
+# 3. Reload google-fluentd:
+#       sudo service google-fluentd restart
+# 4. Now `proxy.py` logs can be browsed using GCE log viewer:
+#       https://console.cloud.google.com/logs/viewer
+#
+# See https://cloud.google.com/logging/docs/agent/configuration#third-party_application_log_input_configuration
+# for more options.
+<source>
+    @type tail
+    format none
+    path /tmp/proxy.log
+    pos_file /var/lib/google-fluentd/pos/proxy-py.pos
+    read_from_head true
+    tag proxy.py
+</source>
diff --git a/plugin_examples.py b/plugin_examples.py
index b40e132..968eeee 100644
--- a/plugin_examples.py
+++ b/plugin_examples.py
@@ -3,7 +3,7 @@
     ~~~~~~~~
     Lightweight, Programmable, TLS interceptor Proxy for HTTP(S), HTTP2, WebSockets protocols in a single Python file.
 
-    :copyright: (c) 2013-present by Abhinav Singh.
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
 import json
@@ -19,7 +19,7 @@ import proxy
 @proxy.route(b'/hello-world')
 def hello_world(_request: proxy.HttpParser) -> bytes:
     """A HttpWebServerRoutePlugin plugin for inbuilt web server."""
-    return proxy.HttpParser.build_response(200, body=b'Hello World')
+    return proxy.build_http_response(200, body=b'Hello World')
 
 
 class ProposedRestApiPlugin(proxy.HttpProxyBasePlugin):
@@ -59,14 +59,14 @@ class ProposedRestApiPlugin(proxy.HttpProxyBasePlugin):
         before connecting to upstream server."""
         if self.request.host == self.API_SERVER and self.request.url:
             if self.request.url.path in self.REST_API_SPEC:
-                self.client.send(proxy.HttpParser.build_response(
+                self.client.send(proxy.build_http_response(
                     200, reason=b'OK',
                     headers={b'Content-Type': b'application/json'},
                     body=proxy.bytes_(json.dumps(
                         self.REST_API_SPEC[self.request.url.path]))
                 ))
             else:
-                self.client.send(proxy.HttpParser.build_response(
+                self.client.send(proxy.build_http_response(
                     404, reason=b'NOT FOUND', body=b'Not Found'
                 ))
             return True
@@ -166,7 +166,7 @@ class ManInTheMiddlePlugin(proxy.HttpProxyBasePlugin):
         pass
 
     def handle_upstream_response(self, raw: bytes) -> bytes:
-        return proxy.HttpParser.build_response(
+        return proxy.build_http_response(
             200, reason=b'OK', body=b'Hello from man in the middle')
 
     def on_upstream_connection_close(self) -> None:
diff --git a/proxy.py b/proxy.py
index 57b06a7..d84bb78 100755
--- a/proxy.py
+++ b/proxy.py
@@ -5,13 +5,14 @@
     ~~~~~~~~
     Lightweight, Programmable, TLS interceptor Proxy for HTTP(S), HTTP2, WebSockets protocols in a single Python file.
 
-    :copyright: (c) 2013-present by Abhinav Singh.
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
 import argparse
 import base64
 import datetime
 import errno
+import hashlib
 import importlib
 import inspect
 import ipaddress
@@ -19,6 +20,8 @@ import logging
 import multiprocessing
 import os
 import pathlib
+import secrets
+import selectors
 import socket
 import ssl
 import subprocess
@@ -31,7 +34,7 @@ from multiprocessing.reduction import send_handle, recv_handle
 from typing import Any, Dict, List, Tuple, Optional, Union, NamedTuple, Type, Callable
 from urllib import parse as urlparse
 
-import select
+from typing_extensions import Protocol
 
 if os.name != 'nt':
     import resource
@@ -51,12 +54,18 @@ logger = logging.getLogger(__name__)
 # Defaults
 DEFAULT_BACKLOG = 100
 DEFAULT_BASIC_AUTH = None
+DEFAULT_CA_KEY_FILE = None
+DEFAULT_CA_CERT_DIR = None
+DEFAULT_CA_CERT_FILE = None
+DEFAULT_CA_SIGNING_KEY_FILE = None
+DEFAULT_CERT_FILE = None
 DEFAULT_BUFFER_SIZE = 1024 * 1024
 DEFAULT_CLIENT_RECVBUF_SIZE = DEFAULT_BUFFER_SIZE
 DEFAULT_SERVER_RECVBUF_SIZE = DEFAULT_BUFFER_SIZE
 DEFAULT_DISABLE_HEADERS: List[bytes] = []
 DEFAULT_IPV4_HOSTNAME = ipaddress.IPv4Address('127.0.0.1')
 DEFAULT_IPV6_HOSTNAME = ipaddress.IPv6Address('::1')
+DEFAULT_KEY_FILE = None
 DEFAULT_PORT = 8899
 DEFAULT_DISABLE_HTTP_PROXY = False
 DEFAULT_ENABLE_WEB_SERVER = False
@@ -66,7 +75,7 @@ DEFAULT_PAC_FILE = None
 DEFAULT_PAC_FILE_URL_PATH = b'/'
 DEFAULT_PID_FILE = None
 DEFAULT_NUM_WORKERS = 0
-DEFAULT_PLUGINS = ''
+DEFAULT_PLUGINS = ''    # Comma separated list of plugins
 DEFAULT_VERSION = False
 DEFAULT_LOG_FORMAT = '%(asctime)s - %(levelname)s - pid:%(process)d - %(funcName)s:%(lineno)d - %(message)s'
 DEFAULT_LOG_FILE = None
@@ -120,12 +129,6 @@ TcpConnectionTypes = NamedTuple('TcpConnectionTypes', [
 ])
 tcpConnectionTypes = TcpConnectionTypes(1, 2)
 
-WorkerOperations = NamedTuple('WorkerOperations', [
-    ('HTTP_PROTOCOL', int),
-    ('SHUTDOWN', int),
-])
-workerOperations = WorkerOperations(1, 2)
-
 ChunkParserStates = NamedTuple('ChunkParserStates', [
     ('WAITING_FOR_SIZE', int),
     ('WAITING_FOR_DATA', int),
@@ -152,8 +155,14 @@ httpParserTypes = HttpParserTypes(1, 2)
 HttpProtocolTypes = NamedTuple('HttpProtocolTypes', [
     ('HTTP', int),
     ('HTTPS', int),
+    ('WEBSOCKET', int),
 ])
-httpProtocolTypes = HttpProtocolTypes(1, 2)
+httpProtocolTypes = HttpProtocolTypes(1, 2, 3)
+
+
+class _HasFileno(Protocol):
+    def fileno(self) -> int:
+        ...     # pragma: no cover
 
 
 class TcpConnectionUninitializedException(Exception):
@@ -172,7 +181,7 @@ class TcpConnection(ABC):
     @abstractmethod
     def connection(self) -> Union[ssl.SSLSocket, socket.socket]:
         """Must return the socket connection to use in this class."""
-        raise TcpConnectionUninitializedException()
+        raise TcpConnectionUninitializedException()     # pragma: no cover
 
     def send(self, data: bytes) -> int:
         """Users must handle BrokenPipeError exceptions"""
@@ -212,11 +221,31 @@ class TcpConnection(ABC):
         return len(data)
 
     def flush(self) -> int:
+        if self.buffer_size() == 0:
+            return 0
         sent: int = self.send(self.buffer)
         self.buffer = self.buffer[sent:]
         logger.debug('flushed %d bytes to %s' % (sent, self.tag))
         return sent
 
+    @staticmethod
+    def new(addr: Tuple[str, int]) -> socket.socket:
+        try:
+            ip = ipaddress.ip_address(addr[0])
+            if ip.version == 4:
+                conn = socket.socket(
+                    socket.AF_INET, socket.SOCK_STREAM, 0)
+                conn.connect(addr)
+            else:
+                conn = socket.socket(
+                    socket.AF_INET6, socket.SOCK_STREAM, 0)
+                conn.connect((addr[0], addr[1], 0, 0))
+        except ValueError:
+            # Not a valid IP address, most likely its a domain name,
+            # try to establish dual stack IPv4/IPv6 connection.
+            conn = socket.create_connection(addr)
+        return conn
+
 
 class TcpServerConnection(TcpConnection):
     """Establishes connection to upstream server."""
@@ -235,21 +264,7 @@ class TcpServerConnection(TcpConnection):
     def connect(self) -> None:
         if self._conn is not None:
             return
-
-        try:
-            ip = ipaddress.ip_address(text_(self.addr[0]))
-            if ip.version == 4:
-                self._conn = socket.socket(
-                    socket.AF_INET, socket.SOCK_STREAM, 0)
-                self._conn.connect((self.addr[0], self.addr[1]))
-            else:
-                self._conn = socket.socket(
-                    socket.AF_INET6, socket.SOCK_STREAM, 0)
-                self._conn.connect((self.addr[0], self.addr[1], 0, 0))
-        except ValueError:
-            # Not a valid IP address, most likely its a domain name,
-            # try to establish dual stack IPv4/IPv6 connection.
-            self._conn = socket.create_connection((self.addr[0], self.addr[1]))
+        self._conn = self.new(self.addr)
 
 
 class TcpClientConnection(TcpConnection):
@@ -268,99 +283,48 @@ class TcpClientConnection(TcpConnection):
         return self._conn
 
 
-class TcpServer(ABC):
-    """TcpServer server implementation.
+class AcceptorPool:
+    """AcceptorPool.
 
-    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.
+    Pre-spawns worker processes to utilize all cores available on the system.  Server socket connection is
+    dispatched over a pipe to workers.  Each worker accepts incoming client request and spawns a
+    separate thread to handle the client request.
     """
 
     def __init__(self,
                  hostname: Union[ipaddress.IPv4Address,
-                                 ipaddress.IPv6Address] = DEFAULT_IPV6_HOSTNAME,
-                 port: int = DEFAULT_PORT,
-                 backlog: int = DEFAULT_BACKLOG):
-        self.port: int = port
-        self.backlog: int = backlog
-        self.socket: Optional[socket.socket] = None
+                                 ipaddress.IPv6Address],
+                 port: int, backlog: int, num_workers: int,
+                 work_klass: type, **kwargs: Any) -> None:
         self.running: bool = False
-        self.family: socket.AddressFamily = socket.AF_INET6 if hostname.version == 6 else socket.AF_INET
+
         self.hostname: Union[ipaddress.IPv4Address,
                              ipaddress.IPv6Address] = hostname
+        self.port: int = port
+        self.family: socket.AddressFamily = socket.AF_INET6 if hostname.version == 6 else socket.AF_INET
+        self.backlog: int = backlog
+        self.socket: Optional[socket.socket] = None
 
-    @abstractmethod
-    def setup(self) -> None:
-        pass  # pragma: no cover
-
-    @abstractmethod
-    def handle(self, client: TcpClientConnection) -> None:
-        raise NotImplementedError()  # pragma: no cover
-
-    @abstractmethod
-    def shutdown(self) -> None:
-        pass  # pragma: no cover
-
-    def stop(self) -> None:
-        self.running = False
-
-    def run_once(self) -> None:
-        if self.socket:
-            r, w, x = select.select([self.socket], [], [], 1)
-            if self.socket in r:
-                try:
-                    conn, addr = self.socket.accept()
-                    client = TcpClientConnection(conn, addr)
-                    self.handle(client)
-                except ssl.SSLError as e:
-                    logger.exception('SSLError encountered', exc_info=e)
-
-    def run(self) -> None:
-        self.running = True
-        self.setup()
-        try:
-            self.socket = socket.socket(self.family, socket.SOCK_STREAM)
-            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            self.socket.bind((str(self.hostname), self.port))
-            self.socket.listen(self.backlog)
-            logger.info('Started server on %s:%d' % (self.hostname, self.port))
-            while self.running:
-                self.run_once()
-        except Exception as e:
-            logger.exception(
-                'Unexpected error, tearing down server.',
-                exc_info=e)
-        finally:
-            self.shutdown()
-            logger.info('Closing server socket')
-            if self.socket:
-                self.socket.close()
-
-
-class WorkerPool(TcpServer):
-    """WorkerPool.
-
-    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: Union[ipaddress.IPv4Address,
-                                       ipaddress.IPv6Address],
-                 port: int, backlog: int, num_workers: int,
-                 work_klass: type, **kwargs: Any) -> None:
-        super().__init__(hostname=hostname, port=port, backlog=backlog)
+        self.current_worker_id = 0
         self.num_workers = num_workers
-
         self.workers: List[Worker] = []
         self.work_queues: List[Tuple[connection.Connection,
                                      connection.Connection]] = []
-        self.current_worker_id = 0
 
         self.work_klass = work_klass
         self.kwargs = kwargs
 
-    def setup(self) -> None:
+    def listen(self) -> None:
+        self.socket = socket.socket(self.family, socket.SOCK_STREAM)
+        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.socket.bind((str(self.hostname), self.port))
+        self.socket.listen(self.backlog)
+        self.socket.setblocking(False)
+        self.socket.settimeout(0)
+        logger.info('Listening on %s:%d' % (self.hostname, self.port))
+
+    def start_workers(self) -> None:
+        """Start worker processes."""
         for worker_id in range(self.num_workers):
             work_queue = multiprocessing.Pipe()
 
@@ -372,37 +336,35 @@ class WorkerPool(TcpServer):
             self.work_queues.append(work_queue)
         logger.info('Started %d workers' % self.num_workers)
 
-    def handle(self, client: TcpClientConnection) -> None:
-        # Dispatch in round robin fashion
-        work_queue = self.work_queues[self.current_worker_id]
-        logger.debug(
-            'Dispatched client request to worker id %d',
-            self.current_worker_id)
-        # Dispatch non-socket data first, followed by fileno using reduction
-        work_queue[0].send((workerOperations.HTTP_PROTOCOL, client.addr))
-        send_handle(work_queue[0], client.connection.fileno(),
-                    self.workers[self.current_worker_id].pid)
-        # Close parent handler
-        client.close()
-        self.current_worker_id += 1
-        self.current_worker_id %= self.num_workers
-
     def shutdown(self) -> None:
         logger.info('Shutting down %d workers' % self.num_workers)
-        for work_queue in self.work_queues:
-            work_queue[0].send((workerOperations.SHUTDOWN, None))
-            work_queue[0].close()
         for worker in self.workers:
             worker.join()
 
+    def setup(self) -> None:
+        """Listen on port, setup workers and pass server socket to workers."""
+        self.running = True
+        self.listen()
+        self.start_workers()
+
+        # Send server socket to workers.
+        assert self.socket is not None
+        for work_queue in self.work_queues:
+            work_queue[0].send(self.family)
+            send_handle(work_queue[0], self.socket.fileno(),
+                        self.workers[self.current_worker_id].pid)
+        self.socket.close()
+
 
 class Worker(multiprocessing.Process):
-    """Generic worker class implementation.
+    """Socket client acceptor.
 
-    Worker instance accepts (operation, payload) over work queue and
-    depending upon requested operation starts a new thread to handle the work.
+    Accepts client connection over received server socket handle and
+    starts a new work thread.
     """
 
+    lock = multiprocessing.Lock()
+
     def __init__(
             self,
             work_queue: connection.Connection,
@@ -413,30 +375,37 @@ class Worker(multiprocessing.Process):
         self.work_klass = work_klass
         self.kwargs = kwargs
 
-    def run_once(self) -> bool:
-        try:
-            op, payload = self.work_queue.recv()
-            if op == workerOperations.HTTP_PROTOCOL:
-                fileno = recv_handle(self.work_queue)
-                proxy = self.work_klass(
-                    fileno=fileno, addr=payload, **self.kwargs)
-                proxy.setDaemon(True)
-                proxy.start()
-            elif op == workerOperations.SHUTDOWN:
-                logger.debug('Worker shutting down....')
-                self.work_queue.close()
-                return True
-        except ConnectionRefusedError:
-            return False
-        except KeyboardInterrupt:  # pragma: no cover
-            return True
-        return False
-
     def run(self) -> None:
-        while True:
-            teardown = self.run_once()
-            if teardown:
-                break
+        family = self.work_queue.recv()
+        sock = socket.fromfd(
+            recv_handle(self.work_queue),
+            family=family,
+            type=socket.SOCK_STREAM
+        )
+        selector = selectors.DefaultSelector()
+        try:
+            while True:
+                with self.lock:
+                    selector.register(sock, selectors.EVENT_READ)
+                    events = selector.select(timeout=1)
+                    selector.unregister(sock)
+                    if len(events) == 0:
+                        continue
+                try:
+                    conn, addr = sock.accept()
+                except BlockingIOError:  # as e:
+                    # logger.exception('BlockingIOError', exc_info=e)
+                    continue
+                work = self.work_klass(
+                    fileno=conn.fileno(),
+                    addr=addr,
+                    **self.kwargs)
+                work.setDaemon(True)
+                work.start()
+        except KeyboardInterrupt:
+            pass
+        finally:
+            sock.close()
 
 
 class ChunkParser:
@@ -461,7 +430,7 @@ class ChunkParser:
             raw = self.chunk + raw
             self.chunk = b''
             # Extract following chunk data size
-            line, raw = HttpParser.find_line(raw)
+            line, raw = find_http_line(raw)
             # CRLF not received or Blank line was received.
             if line is None or line.strip() == b'':
                 self.chunk = raw
@@ -486,6 +455,62 @@ class ChunkParser:
         return len(raw) > 0, raw
 
 
+def build_http_request(method: bytes, url: bytes,
+                       protocol_version: bytes = b'HTTP/1.1',
+                       headers: Optional[Dict[bytes, bytes]] = None,
+                       body: Optional[bytes] = None) -> bytes:
+    if headers is None:
+        headers = {}
+    return build_http_pkt(
+        [method, url, protocol_version], headers, body)
+
+
+def build_http_response(status_code: int,
+                        protocol_version: bytes = b'HTTP/1.1',
+                        reason: Optional[bytes] = None,
+                        headers: Optional[Dict[bytes, bytes]] = None,
+                        body: Optional[bytes] = None) -> bytes:
+    line = [protocol_version, bytes_(status_code)]
+    if reason:
+        line.append(reason)
+    if headers is None:
+        headers = {}
+    if body is not None and not any(
+            k.lower() == b'content-length' for k in headers):
+        headers[b'Content-Length'] = bytes_(len(body))
+    return build_http_pkt(line, headers, body)
+
+
+def build_http_header(k: bytes, v: bytes) -> bytes:
+    return k + COLON + WHITESPACE + v
+
+
+def build_http_pkt(line: List[bytes],
+                   headers: Optional[Dict[bytes, bytes]] = None,
+                   body: Optional[bytes] = None) -> bytes:
+    req = WHITESPACE.join(line) + CRLF
+    if headers is not None:
+        for k in headers:
+            req += build_http_header(k, headers[k]) + CRLF
+    req += CRLF
+    if body:
+        req += body
+    return req
+
+
+def find_http_line(raw: bytes) -> Tuple[Optional[bytes], bytes]:
+    """Finds first line of request ending in CRLF.
+
+    If no CRLF is found, line is None.
+    Also returns pending buffer after received line of request."""
+    pos = raw.find(CRLF)
+    if pos == -1:
+        return None, raw
+    line = raw[:pos]
+    rest = raw[pos + len(CRLF):]
+    return line, rest
+
+
 class HttpParser:
     """HTTP request/response parser."""
 
@@ -519,6 +544,14 @@ class HttpParser:
         self.host: Optional[bytes] = None
         self.port: Optional[int] = None
 
+    def header(self, key: bytes) -> bytes:
+        if key.lower() not in self.headers:
+            raise KeyError('%s not found in headers', text_(key))
+        return self.headers[key.lower()][1]
+
+    def has_header(self, key: bytes) -> bool:
+        return key.lower() in self.headers
+
     def add_header(self, key: bytes, value: bytes) -> None:
         self.headers[key.lower()] = (key, value)
 
@@ -591,7 +624,7 @@ class HttpParser:
 
     def process(self, raw: bytes) -> Tuple[bool, bytes]:
         """Returns False when no CRLF could be found in received bytes."""
-        line, raw = HttpParser.find_line(raw)
+        line, raw = find_http_line(raw)
         if line is None:
             return False, raw
 
@@ -669,68 +702,13 @@ class HttpParser:
         assert self.method and self.version
         if disable_headers is None:
             disable_headers = DEFAULT_DISABLE_HEADERS
-        return HttpParser.build_request(
+        return build_http_request(
             self.method, self.build_url(), self.version,
             headers={} if not self.headers else {self.headers[k][0]: self.headers[k][1] for k in self.headers if
                                                  k.lower() not in disable_headers},
             body=self.body
         )
 
-    @staticmethod
-    def build_request(method: bytes, url: bytes, protocol_version: bytes,
-                      headers: Optional[Dict[bytes, bytes]] = None,
-                      body: Optional[bytes] = None) -> bytes:
-        if headers is None:
-            headers = {}
-        return HttpParser.build_pkt(
-            [method, url, protocol_version], headers, body)
-
-    @staticmethod
-    def build_response(status_code: int,
-                       protocol_version: bytes = b'HTTP/1.1',
-                       reason: Optional[bytes] = None,
-                       headers: Optional[Dict[bytes, bytes]] = None,
-                       body: Optional[bytes] = None) -> bytes:
-        line = [protocol_version, bytes_(status_code)]
-        if reason:
-            line.append(reason)
-        if headers is None:
-            headers = {}
-        if body is not None and not any(
-                k.lower() == b'content-length' for k in headers):
-            headers[b'Content-Length'] = bytes_(len(body))
-        return HttpParser.build_pkt(line, headers, body)
-
-    @staticmethod
-    def build_pkt(line: List[bytes],
-                  headers: Optional[Dict[bytes, bytes]] = None,
-                  body: Optional[bytes] = None) -> bytes:
-        req = WHITESPACE.join(line) + CRLF
-        if headers is not None:
-            for k in headers:
-                req += HttpParser.build_header(k, headers[k]) + CRLF
-        req += CRLF
-        if body:
-            req += body
-        return req
-
-    @staticmethod
-    def build_header(k: bytes, v: bytes) -> bytes:
-        return k + COLON + WHITESPACE + v
-
-    @staticmethod
-    def find_line(raw: bytes) -> Tuple[Optional[bytes], bytes]:
-        """Finds first line of request ending in CRLF.
-
-        If no CRLF is found, line is None.
-        Also returns pending buffer after received line of request."""
-        pos = raw.find(CRLF)
-        if pos == -1:
-            return None, raw
-        line = raw[:pos]
-        rest = raw[pos + len(CRLF):]
-        return line, rest
-
     ##########################################################################
     # HttpParser was originally written to parse the incoming raw Http requests.
     # Since request / response objects passed to ProtocolHandlerPlugin methods
@@ -763,7 +741,6 @@ class HttpRequestRejected(ProtocolException):
                  status_code: Optional[int] = None,
                  reason: Optional[bytes] = None,
                  body: Optional[bytes] = None):
-        super().__init__()
         self.status_code: Optional[int] = status_code
         self.reason: Optional[bytes] = reason
         self.body: Optional[bytes] = body
@@ -872,15 +849,15 @@ class ProtocolHandlerPlugin(ABC):
 
     @abstractmethod
     def get_descriptors(
-            self) -> Tuple[List[socket.socket], List[socket.socket], List[socket.socket]]:
-        return [], [], []  # pragma: no cover
+            self) -> Tuple[List[socket.socket], List[socket.socket]]:
+        return [], []  # pragma: no cover
 
     @abstractmethod
-    def flush_to_descriptors(self, w: List[socket.socket]) -> bool:
+    def flush_to_descriptors(self, w: List[Union[int, _HasFileno]]) -> bool:
         pass  # pragma: no cover
 
     @abstractmethod
-    def read_from_descriptors(self, r: List[socket.socket]) -> bool:
+    def read_from_descriptors(self, r: List[Union[int, _HasFileno]]) -> bool:
         pass  # pragma: no cover
 
     @abstractmethod
@@ -899,10 +876,6 @@ class ProtocolHandlerPlugin(ABC):
         Return optionally modified chunk to return back to client."""
         return chunk  # pragma: no cover
 
-    @abstractmethod
-    def access_log(self) -> None:
-        pass  # pragma: no cover
-
     @abstractmethod
     def on_client_connection_close(self) -> None:
         pass  # pragma: no cover
@@ -911,7 +884,7 @@ class ProtocolHandlerPlugin(ABC):
 class ProxyConnectionFailed(ProtocolException):
     """Exception raised when HttpProxyPlugin is unable to establish connection to upstream server."""
 
-    RESPONSE_PKT = HttpParser.build_response(
+    RESPONSE_PKT = build_http_response(
         502, reason=b'Bad Gateway',
         headers={PROXY_AGENT_HEADER_KEY: PROXY_AGENT_HEADER_VALUE,
                  b'Connection': b'close'},
@@ -935,7 +908,7 @@ class ProxyAuthenticationFailed(ProtocolException):
     """Exception raised when Http Proxy auth is enabled and
     incoming request doesn't present necessary credentials."""
 
-    RESPONSE_PKT = HttpParser.build_response(
+    RESPONSE_PKT = build_http_response(
         407, reason=b'Proxy Authentication Required',
         headers={PROXY_AGENT_HEADER_KEY: PROXY_AGENT_HEADER_VALUE,
                  b'Connection': b'close',
@@ -996,7 +969,7 @@ class HttpProxyBasePlugin(ABC):
 class HttpProxyPlugin(ProtocolHandlerPlugin):
     """ProtocolHandler plugin which implements HttpProxy specifications."""
 
-    PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT = HttpParser.build_response(
+    PROXY_TUNNEL_ESTABLISHED_RESPONSE_PKT = build_http_response(
         200, reason=b'Connection established'
     )
 
@@ -1020,9 +993,9 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
                 self.plugins[instance.name()] = instance
 
     def get_descriptors(
-            self) -> Tuple[List[socket.socket], List[socket.socket], List[socket.socket]]:
+            self) -> Tuple[List[socket.socket], List[socket.socket]]:
         if not self.request.has_upstream_server():
-            return [], [], []
+            return [], []
 
         r: List[socket.socket] = []
         w: List[socket.socket] = []
@@ -1031,11 +1004,13 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
         if self.server and not self.server.closed and \
                 self.server.has_buffer() and self.server.connection:
             w.append(self.server.connection)
-        return r, w, []
+        return r, w
 
-    def flush_to_descriptors(self, w: List[socket.socket]) -> bool:
+    def flush_to_descriptors(self, w: List[Union[int, _HasFileno]]) -> bool:
         if self.request.has_upstream_server() and \
-                self.server and not self.server.closed and self.server.connection in w:
+                self.server and not self.server.closed and \
+                self.server.buffer_size() > 0 and \
+                self.server.connection in w:
             logger.debug('Server is write ready, flushing buffer')
             try:
                 self.server.flush()
@@ -1045,7 +1020,7 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
                 return True
         return False
 
-    def read_from_descriptors(self, r: List[socket.socket]) -> bool:
+    def read_from_descriptors(self, r: List[Union[int, _HasFileno]]) -> bool:
         if self.request.has_upstream_server(
         ) and self.server and not self.server.closed and self.server.connection in r:
             logger.debug('Server is ready for reads, reading')
@@ -1069,15 +1044,38 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
         return False
 
     def on_client_connection_close(self) -> None:
-        if self.request.has_upstream_server() and self.server:
+        if not self.request.has_upstream_server():
+            return
+        server_host, server_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_(server_host),
+                 text_(server_port),
+                 self.response.total_size))
+        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_(server_host), server_port,
+                 text_(self.request.build_url()),
+                 text_(self.response.code),
+                 text_(self.response.reason),
+                 self.response.total_size))
+        # Invoke plugin.on_upstream_connection_close
+        if self.server and not self.server.closed:
+            for plugin in self.plugins.values():
+                plugin.on_upstream_connection_close()
+            self.server.close()
             logger.debug(
                 'Closed server connection with pending server buffer size %d bytes' %
                 self.server.buffer_size())
-            if not self.server.closed:
-                # Invoke plugin.on_upstream_connection_close
-                for plugin in self.plugins.values():
-                    plugin.on_upstream_connection_close()
-                self.server.close()
 
     def handle_response_chunk(self, chunk: bytes) -> bytes:
         return chunk
@@ -1104,8 +1102,7 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
                 if not os.path.isfile(cert_file_path):
                     logger.debug('Generating certificates %s', cert_file_path)
                     # TODO: Use ssl.get_server_certificate to populate generated certificate metadata
-                    # Currently we only set CN=example.org on the generated
-                    # certificates.
+                    # Currently we only set CN= field for generated certificates.
                     gen_cert = subprocess.Popen(
                         ['/usr/bin/openssl', 'req', '-new', '-key', self.config.ca_signing_key_file, '-subj',
                          '/CN=%s' % text_(self.request.host)],
@@ -1113,7 +1110,7 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
                         stderr=subprocess.PIPE)
                     sign_cert = subprocess.Popen(
                         ['/usr/bin/openssl', 'x509', '-req', '-days', '365', '-CA', self.config.ca_cert_file, '-CAkey',
-                         self.config.ca_key_file, '-set_serial', str(int(time.time() * 1000)), '-out', cert_file_path],
+                         self.config.ca_key_file, '-set_serial', str(int(time.time())), '-out', cert_file_path],
                         stdin=gen_cert.stdout,
                         stderr=subprocess.PIPE)
                     # TODO: Ensure sign_cert success.
@@ -1138,24 +1135,16 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
         for plugin in self.plugins.values():
             plugin.on_upstream_connection()
 
-        # 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)
             # If interception is enabled, generate server certificates
             if self.config.ca_key_file and self.config.ca_cert_file and self.config.ca_signing_key_file:
-                # Flush client buffer before wrapping, but is client ready for
-                # writes?
+                # Flush client buffer before wrapping,
+                # but is client ready for writes?
                 self.client.flush()
                 generated_cert = self.generate_upstream_certificate()
                 if generated_cert:
-                    # If client is communicating over https,
-                    # self.client.conn has already been wrapped before.
-                    # We could unwrap, but then we can't maintain our https
-                    # connection to the client. Below we handle the scenario
-                    # when client is communicating to proxy.py using http.
                     if not (self.config.keyfile and self.config.certfile) and \
                             self.server and isinstance(self.server.connection, socket.socket):
                         self.client._conn = ssl.wrap_socket(
@@ -1173,8 +1162,6 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
                         logger.info(
                             'TLS interception using %s', generated_cert)
                         return self.client.connection
-        # for general http requests, re-build request packet
-        # and queue for the server with appropriate headers
         elif self.server:
             # - proxy-connection header is a mistake, it doesn't seem to be
             #   officially documented in any specification, drop it.
@@ -1195,33 +1182,6 @@ class HttpProxyPlugin(ProtocolHandlerPlugin):
                     disable_headers=self.config.disable_headers))
         return False
 
-    def access_log(self) -> None:
-        if not self.request.has_upstream_server():
-            return
-
-        server_host, server_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_(server_host),
-                 text_(server_port),
-                 self.response.total_size))
-        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_(server_host), server_port,
-                    text_(self.request.build_url()),
-                    text_(self.response.code),
-                    text_(self.response.reason),
-                    self.response.total_size))
-
     def authenticate(self) -> None:
         if self.config.auth_code:
             if b'proxy-authorization' not in self.request.headers or \
@@ -1279,13 +1239,15 @@ def route(path: bytes, protocols: Optional[List[int]] = None) -> \
             def routes(self) -> List[Tuple[int, bytes]]:
                 p = protocols
                 if p is None:
-                    p = [httpProtocolTypes.HTTP, httpProtocolTypes.HTTPS]
+                    p = [i + 1 for i in range(len(httpProtocolTypes))]
                 return [(protocol, path) for protocol in p]
 
             def handle_request(self, request: HttpParser) -> None:
                 self.client.queue(func(request))
                 self.client.flush()
+
         return HttpWebServerRouteHandler
+
     return decorator
 
 
@@ -1314,12 +1276,10 @@ class HttpWebServerPacFilePlugin(HttpWebServerRoutePlugin):
         if self.config.pac_file:
             try:
                 with open(self.config.pac_file, 'rb') as f:
-                    logger.debug('Will serve pac file from disk')
                     content = f.read()
             except IOError:
-                logger.debug('Will serve pac file content from buffer')
                 content = bytes_(self.config.pac_file)
-            self.pac_file_response = HttpParser.build_response(
+            self.pac_file_response = build_http_response(
                 200, reason=b'OK', headers={
                     b'Content-Type': b'application/x-ns-proxy-autoconfig',
                     b'Connection': b'close'
@@ -1327,15 +1287,120 @@ class HttpWebServerPacFilePlugin(HttpWebServerRoutePlugin):
             )
 
 
+class WebsocketFrame:
+    """Websocket frames parser and constructor."""
+
+    GUID = b'258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
+
+    def __init__(self) -> None:
+        self.fin: bool = False
+        self.rsv1: bool = False
+        self.rsv2: bool = False
+        self.rsv3: bool = False
+        self.opcode: int = 0
+        self.masked: bool = False
+        self.payload_length: int = 0
+        self.mask: Optional[bytes] = None
+        self.data: Optional[bytes] = None
+
+    def parse(self, raw: bytes) -> None:
+        self.fin = bool(raw[0] & 1 << 7)
+        self.rsv1 = bool(raw[0] & 1 << 6)
+        self.rsv2 = bool(raw[0] & 1 << 5)
+        self.rsv3 = bool(raw[0] & 1 << 4)
+        self.opcode = raw[0] & 0b00001111
+        self.masked = bool(raw[1] & 0b10000000)
+        # TODO: Handle case when payload length is 126 or 127
+        self.payload_length = raw[1] & 0b01111111
+        if self.masked:
+            self.mask = raw[2:6]
+        self.data = raw[6:]
+        if self.masked:
+            assert self.mask is not None
+            data = bytearray(self.data)
+            for i in range(len(self.data)):
+                data[i] = data[i] ^ self.mask[i % 4]
+            self.data = bytes(data)
+
+    @staticmethod
+    def key_to_accept(key: bytes) -> bytes:
+        sha1 = hashlib.sha1()
+        sha1.update(key + WebsocketFrame.GUID)
+        return base64.b64encode(sha1.digest())
+
+
+class Websocket:
+
+    def __init__(self, hostname: Union[ipaddress.IPv4Address, ipaddress.IPv6Address],
+                 port: int, on_message: Optional[Callable[[bytes], None]] = None) -> None:
+        self.hostname: Union[ipaddress.IPv4Address, ipaddress.IPv6Address] = hostname
+        self.port: int = port
+        self.sock: socket.socket = TcpConnection.new((str(self.hostname), self.port))
+        self.on_message: Optional[Callable[[bytes], None]] = on_message
+        self.upgrade()
+
+    @staticmethod
+    def build_handshake_request(
+            key: bytes,
+            method: bytes = b'GET',
+            url: bytes = b'/') -> bytes:
+        return build_http_request(
+            method, url,
+            headers={
+                b'Connection': b'upgrade',
+                b'Upgrade': b'websocket',
+                b'Sec-WebSocket-Key': key,
+                b'Sec-WebSocket-Version': b'13',
+            }
+        )
+
+    @staticmethod
+    def build_handshake_response(accept: bytes) -> bytes:
+        return build_http_response(
+            101, reason=b'Switching Protocols',
+            headers={
+                b'Upgrade': b'websocket',
+                b'Connection': b'Upgrade',
+                b'Sec-WebSocket-Accept': accept
+            }
+        )
+
+    def upgrade(self) -> None:
+        key = base64.b64encode(secrets.token_bytes(16))
+        self.sock.send(self.build_handshake_request(key))
+        response = HttpParser(httpParserTypes.RESPONSE_PARSER)
+        response.parse(self.sock.recv(DEFAULT_BUFFER_SIZE))
+        accept = response.header(b'Sec-Websocket-Accept')
+        assert WebsocketFrame.key_to_accept(key) == accept
+
+    def ping(self, data: Optional[bytes] = None) -> None:
+        pass
+
+    def pong(self, data: Optional[bytes] = None) -> None:
+        pass
+
+    def send(self, raw: bytes) -> None:
+        pass
+
+    def close(self, data: Optional[bytes] = None) -> None:
+        pass
+
+
 class HttpWebServerPlugin(ProtocolHandlerPlugin):
-    """ProtocolHandler plugin which handles incoming requests to local webserver."""
+    """ProtocolHandler plugin which handles incoming requests to local web server."""
 
-    DEFAULT_404_RESPONSE = HttpParser.build_response(
+    DEFAULT_404_RESPONSE = build_http_response(
         404, reason=b'NOT FOUND',
         headers={b'Server': PROXY_AGENT_HEADER_VALUE,
                  b'Connection': b'close'}
     )
 
+    DEFAULT_501_RESPONSE = build_http_response(
+        501, reason=b'NOT IMPLEMENTED',
+        headers={b'Server': PROXY_AGENT_HEADER_VALUE,
+                 b'Connection': b'close'}
+    )
+
     def __init__(
             self,
             config: ProtocolConfig,
@@ -1343,6 +1408,8 @@ class HttpWebServerPlugin(ProtocolHandlerPlugin):
             request: HttpParser):
         super().__init__(config, client, request)
 
+        self.switched_protocol: Optional[bytes] = None
+
         self.routes: Dict[int, Dict[bytes, HttpWebServerRoutePlugin]] = {
             httpProtocolTypes.HTTP: {},
             httpProtocolTypes.HTTPS: {},
@@ -1360,14 +1427,36 @@ class HttpWebServerPlugin(ProtocolHandlerPlugin):
 
         url = self.request.build_url()
 
+        # Connection upgrade
+        if self.request.has_header(b'connection') and \
+                self.request.header(b'connection').lower() == b'upgrade':
+            if self.request.has_header(b'upgrade') and \
+                    self.request.header(b'upgrade').lower() == b'websocket':
+                # Complete websocket handshake
+                if not self.request.has_header(b'Sec-WebSocket-Key'):
+                    # TODO: Return invalid request?
+                    return True
+
+                self.client.queue(
+                    Websocket.build_handshake_response(
+                        WebsocketFrame.key_to_accept(
+                            self.request.header(b'Sec-WebSocket-Key'))))
+                self.client.flush()
+                self.switched_protocol = b'websocket'
+                return False
+            else:
+                self.client.queue(self.DEFAULT_501_RESPONSE)
+                self.client.flush()
+                return True
+
         # Routing
         if self.request.method != b'CONNECT':
             registered_routes = self.routes[httpProtocolTypes.HTTP]
         else:
             registered_routes = self.routes[httpProtocolTypes.HTTPS]
-        for route_ in registered_routes:
-            if url == route_:
-                registered_routes[route_].handle_request(self.request)
+        for r in registered_routes:
+            if r == url:
+                registered_routes[r].handle_request(self.request)
                 # But is client ready for flush?
                 self.client.flush()
                 return True
@@ -1378,33 +1467,34 @@ class HttpWebServerPlugin(ProtocolHandlerPlugin):
         self.client.flush()
         return True
 
-    def access_log(self) -> None:
-        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())))
-
-    def flush_to_descriptors(self, w: List[socket.socket]) -> bool:
+    def flush_to_descriptors(self, w: List[Union[int, _HasFileno]]) -> bool:
         pass
 
-    def read_from_descriptors(self, r: List[socket.socket]) -> bool:
+    def read_from_descriptors(self, r: List[Union[int, _HasFileno]]) -> bool:
         pass
 
     def on_client_data(self, raw: bytes) -> Optional[bytes]:
+        if self.switched_protocol == b'websocket':
+            frame = WebsocketFrame()
+            frame.parse(raw)
+            return None
         return raw
 
     def handle_response_chunk(self, chunk: bytes) -> bytes:
         return chunk
 
     def on_client_connection_close(self) -> None:
-        pass
+        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())))
 
     def get_descriptors(
-            self) -> Tuple[List[socket.socket], List[socket.socket], List[socket.socket]]:
-        return [], [], []
+            self) -> Tuple[List[socket.socket], List[socket.socket]]:
+        return [], []
 
 
 class ProtocolHandler(threading.Thread):
@@ -1422,9 +1512,13 @@ class ProtocolHandler(threading.Thread):
         self.config: ProtocolConfig = config if config else ProtocolConfig()
         self.request: HttpParser = HttpParser(httpParserTypes.REQUEST_PARSER)
 
+        self.selector = selectors.DefaultSelector()
+
         conn = self.optionally_wrap_socket(self.fromfd(fileno))
         if conn is None:
             raise TcpConnectionUninitializedException()
+        conn.setblocking(False)
+
         self.client: TcpClientConnection = \
             TcpClientConnection(conn=conn, addr=addr)
 
@@ -1458,8 +1552,12 @@ class ProtocolHandler(threading.Thread):
                 return conn
             except Exception as e:
                 logger.exception('Error encountered', exc_info=e)
-                conn.shutdown(socket.SHUT_RDWR)
-                conn.close()
+                try:
+                    conn.shutdown(socket.SHUT_RDWR)
+                except Exception as e:
+                    logger.exception('Error trying to shutdown client socket', exc_info=e)
+                finally:
+                    conn.close()
                 return None
         return conn
 
@@ -1474,8 +1572,8 @@ class ProtocolHandler(threading.Thread):
         # TODO: Add input argument option for timeout
         return self.connection_inactive_for() > 30
 
-    def handle_writables(self, writables: List[socket.socket]) -> bool:
-        if self.client.connection in writables:
+    def handle_writables(self, writables: List[Union[int, _HasFileno]]) -> bool:
+        if self.client.buffer_size() > 0 and self.client.connection in writables:
             logger.debug('Client is write, flushing buffer')
             try:
                 self.client.flush()
@@ -1485,7 +1583,14 @@ class ProtocolHandler(threading.Thread):
                 return True
         return False
 
-    def handle_readables(self, readables: List[socket.socket]) -> bool:
+    def send_server_error(self, e: Exception) -> None:
+        logger.exception('Server error', exc_info=e)
+        self.client.queue(build_http_response(
+            500, b'Server Error'
+        ))
+        self.client.flush()
+
+    def handle_readables(self, readables: List[Union[int, _HasFileno]]) -> bool:
         if self.client.connection in readables:
             logger.debug('Client is ready for reads, reading')
             client_data = self.client.recv(self.config.client_recvbuf_size)
@@ -1500,6 +1605,8 @@ class ProtocolHandler(threading.Thread):
             plugins = list(self.plugins.values())
             while plugin_index < len(plugins) and client_data:
                 client_data = plugins[plugin_index].on_client_data(client_data)
+                if client_data is None:
+                    break
                 plugin_index += 1
 
             if client_data:
@@ -1531,30 +1638,39 @@ class ProtocolHandler(threading.Thread):
                         self.client.queue(response)
                         # But is client also ready for writes?
                         self.client.flush()
+                    else:
+                        self.send_server_error(e)
                     return True
                 except Exception as e:
-                    raise e
+                    self.send_server_error(e)
+                    return True
         return False
 
-    def run_once(self) -> bool:
-        """Returns True if proxy must teardown."""
-        # Prepare list of descriptors
-        read_desc: List[socket.socket] = [self.client.connection]
-        write_desc: List[socket.socket] = []
-        err_desc: List[socket.socket] = []
+    def get_events(self) -> Dict[socket.socket, int]:
+        events: Dict[socket.socket, int] = {
+            self.client.connection: selectors.EVENT_READ
+        }
         if self.client.has_buffer():
-            write_desc.append(self.client.connection)
+            events[self.client.connection] |= selectors.EVENT_WRITE
 
         # ProtocolHandlerPlugin.get_descriptors
         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
+            plugin_read_desc, plugin_write_desc = plugin.get_descriptors()
+            for r in plugin_read_desc:
+                if r not in events:
+                    events[r] = selectors.EVENT_READ
+                else:
+                    events[r] |= selectors.EVENT_READ
+            for w in plugin_write_desc:
+                if w not in events:
+                    events[w] = selectors.EVENT_WRITE
+                else:
+                    events[w] |= selectors.EVENT_WRITE
 
-        readables, writables, errored = select.select(
-            read_desc, write_desc, err_desc, 1)
+        return events
 
+    def handle_events(self, readables: List[Union[int, _HasFileno]], writables: List[Union[int, _HasFileno]]) -> bool:
+        """Returns True if proxy must teardown."""
         # Flush buffer for ready to write sockets
         teardown = self.handle_writables(writables)
         if teardown:
@@ -1587,6 +1703,32 @@ class ProtocolHandler(threading.Thread):
 
         return False
 
+    def run_once(self) -> bool:
+        events = self.get_events()
+        for fd in events:
+            self.selector.register(fd, events[fd])
+
+        # Select
+        e: List[Tuple[selectors.SelectorKey, int]] = self.selector.select(timeout=1)
+        readables = []
+        writables = []
+        for key, mask in e:
+            if mask & selectors.EVENT_READ:
+                readables.append(key.fileobj)
+            if mask & selectors.EVENT_WRITE:
+                writables.append(key.fileobj)
+
+        teardown = self.handle_events(readables, writables)
+
+        # Unregister
+        for fd in events.keys():
+            self.selector.unregister(fd)
+
+        if teardown:
+            return True
+
+        return False
+
     def run(self) -> None:
         logger.debug('Proxying connection %r' % self.client.connection)
         try:
@@ -1598,29 +1740,28 @@ class ProtocolHandler(threading.Thread):
             pass
         except Exception as e:
             logger.exception(
-                'Exception while handling connection %r with reason %r' %
-                (self.client.connection, e))
+                'Exception while handling connection %r' %
+                self.client.connection, exc_info=e)
         finally:
-            # Invoke plugin.access_log
-            for plugin in self.plugins.values():
-                plugin.access_log()
-
-            if not self.client.closed:
-                try:
-                    self.client.connection.shutdown(socket.SHUT_RDWR)
-                    self.client.close()
-                except OSError:
-                    pass
-
             # Invoke plugin.on_client_connection_close
             for plugin in self.plugins.values():
                 plugin.on_client_connection_close()
 
             logger.debug(
-                'Closed proxy for connection %r '
+                'Closing proxy for connection %r '
                 'at address %r with pending client buffer size %d bytes' %
                 (self.client.connection, self.client.addr, self.client.buffer_size()))
 
+            if not self.client.closed:
+                try:
+                    self.client.connection.shutdown(socket.SHUT_RDWR)
+                    logger.debug('Client connection shutdown successful')
+                except OSError:
+                    pass
+                finally:
+                    self.client.close()
+                    logger.debug('Client connection closed')
+
 
 def is_py3() -> bool:
     """Exists only to avoid mocking sys.version_info in tests."""
@@ -1654,7 +1795,7 @@ def load_plugins(plugins: bytes) -> Dict[bytes, List[type]]:
             continue
         module_name, klass_name = plugin.rsplit(DOT, 1)
         if module_name == 'proxy':
-            klass = getattr(sys.modules[__name__], text_(klass_name))
+            klass = getattr(__name__, text_(klass_name))
         else:
             klass = getattr(
                 importlib.import_module(
@@ -1714,35 +1855,35 @@ def init_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         '--ca-key-file',
         type=str,
-        default=None,
+        default=DEFAULT_CA_KEY_FILE,
         help='Default: None. CA key to use for signing dynamically generated '
              'HTTPS certificates.  If used, must also pass --ca-cert-file and --ca-signing-key-file'
     )
     parser.add_argument(
         '--ca-cert-dir',
         type=str,
-        default=None,
+        default=DEFAULT_CA_CERT_DIR,
         help='Default: ~/.proxy.py. Directory to store dynamically generated certificates. '
              'Also see --ca-key-file, --ca-cert-file and --ca-signing-key-file'
     )
     parser.add_argument(
         '--ca-cert-file',
         type=str,
-        default=None,
+        default=DEFAULT_CA_CERT_FILE,
         help='Default: None. Signing certificate to use for signing dynamically generated '
              'HTTPS certificates.  If used, must also pass --ca-key-file and --ca-signing-key-file'
     )
     parser.add_argument(
         '--ca-signing-key-file',
         type=str,
-        default=None,
+        default=DEFAULT_CA_SIGNING_KEY_FILE,
         help='Default: None. CA signing key to use for dynamic generation of '
              'HTTPS certificates.  If used, must also pass --ca-key-file and --ca-cert-file'
     )
     parser.add_argument(
         '--cert-file',
         type=str,
-        default=None,
+        default=DEFAULT_CERT_FILE,
         help='Default: None. Server certificate to enable end-to-end TLS encryption with clients. '
              'If used, must also pass --key-file.'
     )
@@ -1777,7 +1918,7 @@ def init_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         '--key-file',
         type=str,
-        default=None,
+        default=DEFAULT_KEY_FILE,
         help='Default: None. Server key file to enable end-to-end TLS encryption with clients. '
              'If used, must also pass --cert-file.'
     )
@@ -1873,6 +2014,14 @@ def main(input_args: List[str]) -> None:
         if args.basic_auth:
             auth_code = b'Basic %s' % base64.b64encode(bytes_(args.basic_auth))
 
+        default_plugins = ''
+        if not args.disable_http_proxy:
+            default_plugins += 'proxy.HttpProxyPlugin,'
+        if args.enable_web_server or args.pac_file is not None:
+            default_plugins += 'proxy.HttpWebServerPlugin,'
+        if args.pac_file is not None:
+            default_plugins += 'proxy.HttpWebServerPacFilePlugin,'
+
         config = ProtocolConfig(
             auth_code=auth_code,
             server_recvbuf_size=args.server_recvbuf_size,
@@ -1893,20 +2042,12 @@ def main(input_args: List[str]) -> None:
             backlog=args.backlog,
             num_workers=args.num_workers if args.num_workers > 0 else multiprocessing.cpu_count())
 
-        default_plugins = ''
-        if not args.disable_http_proxy:
-            default_plugins += 'proxy.HttpProxyPlugin,'
-        if args.enable_web_server or config.pac_file is not None:
-            default_plugins += 'proxy.HttpWebServerPlugin,'
-        if config.pac_file is not None:
-            default_plugins += 'proxy.HttpWebServerPacFilePlugin,'
-
         config.plugins = load_plugins(
             bytes_(
                 '%s%s' %
                 (default_plugins, args.plugins)))
 
-        server = WorkerPool(
+        acceptor_pool = AcceptorPool(
             hostname=config.hostname,
             port=config.port,
             backlog=config.backlog,
@@ -1916,7 +2057,14 @@ def main(input_args: List[str]) -> None:
         if args.pid_file:
             with open(args.pid_file, 'wb') as pid_file:
                 pid_file.write(bytes_(os.getpid()))
-        server.run()
+        acceptor_pool.setup()
+
+        try:
+            # TODO: Introduce cron feature instead of mindless sleep
+            while True:
+                time.sleep(1)
+        finally:
+            acceptor_pool.shutdown()
     except KeyboardInterrupt:  # pragma: no cover
         pass
     finally:
diff --git a/requirements-release.txt b/requirements-release.txt
index e506590..d57bd86 100644
--- a/requirements-release.txt
+++ b/requirements-release.txt
@@ -1,3 +1,3 @@
 twine==2.0.0
 wheel==0.33.6
-setuptools==40.6.2
+setuptools==41.2.0
diff --git a/setup.py b/setup.py
index 0935635..843b898 100644
--- a/setup.py
+++ b/setup.py
@@ -5,7 +5,7 @@
     
     Lightweight, Programmable, TLS interceptor Proxy for HTTP(S), HTTP2, WebSockets protocols in a single Python file.
     
-    :copyright: (c) 2013-present by Abhinav Singh.
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
 from setuptools import setup


Test patch:
diff --git a/requirements-testing.txt b/requirements-testing.txt
index 6c53a73..1c67b37 100644
--- a/requirements-testing.txt
+++ b/requirements-testing.txt
@@ -1,6 +1,6 @@
 python-coveralls==2.9.3
 coverage==4.5.4
 flake8==3.7.8
-pytest==5.1.3
+pytest==5.2.0
 autopep8==1.4.4
 mypy==0.730
diff --git a/tests.py b/tests.py
index 1eba932..4438f6f 100644
--- a/tests.py
+++ b/tests.py
@@ -4,18 +4,19 @@
     ~~~~~~~~
     Lightweight, Programmable, TLS interceptor Proxy for HTTP(S), HTTP2, WebSockets protocols in a single Python file.
 
-    :copyright: (c) 2013-present by Abhinav Singh.
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
-import ssl
 import base64
 import errno
+import ipaddress
 import logging
 import multiprocessing
 import os
+import selectors
 import socket
+import ssl
 import tempfile
-import time
 import unittest
 from contextlib import closing
 from http.server import HTTPServer, BaseHTTPRequestHandler
@@ -44,8 +45,28 @@ def get_available_port() -> int:
         return int(port)
 
 
-class TestTcpConnection(unittest.TestCase):
+class TestTextBytes(unittest.TestCase):
+
+    def test_text(self) -> None:
+        self.assertEqual(proxy.text_(b'hello'), 'hello')
+
+    def test_text_int(self) -> None:
+        self.assertEqual(proxy.text_(1), '1')
 
+    def test_text_nochange(self) -> None:
+        self.assertEqual(proxy.text_('hello'), 'hello')
+
+    def test_bytes(self) -> None:
+        self.assertEqual(proxy.bytes_('hello'), b'hello')
+
+    def test_bytes_int(self) -> None:
+        self.assertEqual(proxy.bytes_(1), b'1')
+
+    def test_bytes_nochange(self) -> None:
+        self.assertEqual(proxy.bytes_(b'hello'), b'hello')
+
+
+class TestTcpConnection(unittest.TestCase):
     class TcpConnectionToTest(proxy.TcpConnection):
 
         def __init__(self, conn: Optional[Union[ssl.SSLSocket, socket.socket]] = None,
@@ -126,109 +147,98 @@ class TestTcpConnection(unittest.TestCase):
             (str(proxy.DEFAULT_IPV4_HOSTNAME), proxy.DEFAULT_PORT))
 
 
-class TcpServerUnderTest(proxy.TcpServer):
+class TestAcceptorPool(unittest.TestCase):
+
+    @mock.patch('proxy.send_handle')
+    @mock.patch('multiprocessing.Pipe')
+    @mock.patch('socket.socket')
+    @mock.patch('proxy.Worker')
+    def test_setup(self,
+                   mock_worker: mock.Mock,
+                   mock_socket: mock.Mock,
+                   mock_pipe: mock.Mock,
+                   mock_send_handle: mock.Mock) -> None:
+        num_workers = 2
+        sock = mock_socket.return_value
+        work_klass = mock.MagicMock()
+        kwargs = {'config': proxy.ProtocolConfig()}
+        acceptor = proxy.AcceptorPool(
+            ipaddress.ip_address(proxy.DEFAULT_IPV6_HOSTNAME),
+            proxy.DEFAULT_PORT,
+            proxy.DEFAULT_BACKLOG,
+            num_workers,
+            work_klass=work_klass,
+            **kwargs
+        )
+        acceptor.setup()
+        mock_socket.assert_called_with(
+            socket.AF_INET6 if acceptor.hostname.version == 6 else socket.AF_INET,
+            socket.SOCK_STREAM
+        )
+        sock.setsockopt.assert_called_with(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        sock.bind.assert_called_with((str(acceptor.hostname), acceptor.port))
+        sock.listen.assert_called_with(acceptor.backlog)
+        sock.setblocking.assert_called_with(False)
+        sock.settimeout.assert_called_with(0)
 
-    def handle(self, client: proxy.TcpClientConnection) -> None:
-        data = client.recv(proxy.DEFAULT_BUFFER_SIZE)
-        if data != b'HELLO':
-            raise ValueError('Expected HELLO')
-        client.connection.sendall(b'WORLD')
-        client.close()
+        self.assertTrue(mock_pipe.call_count, num_workers)
+        self.assertTrue(mock_worker.call_count, num_workers)
 
-    def setup(self) -> None:
-        pass
+        sock.close.assert_called()
 
-    def shutdown(self) -> None:
-        pass
 
+class TestWorker(unittest.TestCase):
 
-@unittest.skipIf(os.getenv('TESTING_ON_TRAVIS', 0),
-                 'Opening sockets not allowed on Travis')
-class TestTcpServerIntegration(unittest.TestCase):
+    @mock.patch('proxy.ProtocolHandler')
+    def setUp(self, mock_protocol_handler: mock.Mock) -> None:
+        self.pipe = multiprocessing.Pipe()
+        self.protocol_config = proxy.ProtocolConfig()
+        self.worker = proxy.Worker(
+            self.pipe[1],
+            mock_protocol_handler,
+            config=self.protocol_config)
+        self.mock_protocol_handler = mock_protocol_handler
 
-    ipv4_port: Optional[int] = None
-    ipv6_port: Optional[int] = None
-    ipv4_server: Optional[TcpServerUnderTest] = None
-    ipv6_server: Optional[TcpServerUnderTest] = None
-    ipv4_thread: Optional[Thread] = None
-    ipv6_thread: Optional[Thread] = None
+    @mock.patch('multiprocessing.Lock')
+    @mock.patch('selectors.DefaultSelector')
+    @mock.patch('socket.fromfd')
+    @mock.patch('proxy.recv_handle')
+    def test_accepts_client_from_server_socket(
+            self,
+            mock_recv_handle: mock.Mock,
+            mock_fromfd: mock.Mock,
+            mock_selector: mock.Mock,
+            mock_lock: mock.Mock) -> None:
+        fileno = 10
+        conn = mock.MagicMock()
+        addr = mock.MagicMock()
+        sock = mock_fromfd.return_value
+        mock_fromfd.return_value.accept.return_value = (conn, addr)
+        mock_recv_handle.return_value = fileno
+        self.mock_protocol_handler.return_value.start.side_effect = KeyboardInterrupt()
+        selector = mock_selector.return_value
+        selector.select.return_value = [(None, None)]
+        mock_lock.__enter__.return_value = True
 
-    @classmethod
-    def setUpClass(cls) -> None:
-        cls.ipv4_port = get_available_port()
-        cls.ipv6_port = get_available_port()
-        cls.ipv4_server = TcpServerUnderTest(
-            hostname=proxy.DEFAULT_IPV4_HOSTNAME,
-            port=cls.ipv4_port)
-        cls.ipv6_server = TcpServerUnderTest(
-            hostname=proxy.DEFAULT_IPV6_HOSTNAME,
-            port=cls.ipv6_port)
-        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()
+        self.pipe[0].send(socket.AF_INET6)
+        self.worker.run()
 
-    @classmethod
-    def tearDownClass(cls) -> None:
-        if cls.ipv4_server:
-            cls.ipv4_server.stop()
-        if cls.ipv4_thread:
-            cls.ipv4_thread.join()
-        if cls.ipv6_server:
-            cls.ipv6_server.stop()
-        if cls.ipv6_thread:
-            cls.ipv6_thread.join()
-
-    def baseTestCase(self, ipv4: bool = True) -> None:
-        while True:
-            sock = None
-            try:
-                sock = socket.socket(
-                    socket.AF_INET if ipv4 else socket.AF_INET6,
-                    socket.SOCK_STREAM,
-                    0)
-                sock.connect(
-                    (str(
-                        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:
-                if sock:
-                    sock.close()
-
-    def testIpv4ClientConnection(self) -> None:
-        self.baseTestCase()
-
-    def testIpv6ClientConnection(self) -> None:
-        self.baseTestCase(ipv4=False)
-
-
-class TestTcpServer(unittest.TestCase):
-
-    # Can happen if client is sending invalid https request to our SSL server.
-    # Example, simply sending http traffic to HTTPS server.
-    @mock.patch('select.select')
-    @mock.patch('socket.socket')
-    def testAcceptSSLErrorsSilentlyIgnored(
-            self, mock_socket: mock.Mock, mock_select: mock.Mock) -> None:
-        mock_socket.accept.side_effect = ssl.SSLError()
-        mock_select.return_value = ([mock_socket], [], [])
-        server = TcpServerUnderTest(
-            hostname=proxy.DEFAULT_IPV6_HOSTNAME, port=1234)
-        server.socket = mock_socket
-        with mock.patch('proxy.logger') as mock_logger:
-            server.run_once()
-            mock_logger.exception.assert_called()
-            self.assertTrue(
-                mock_logger.exception.call_args[0][0],
-                'SSLError encountered')
+        selector.register.assert_called_with(sock, selectors.EVENT_READ)
+        selector.unregister.assert_called_with(sock)
+        mock_recv_handle.assert_called_with(self.pipe[1])
+        mock_fromfd.assert_called_with(
+            fileno,
+            family=socket.AF_INET6,
+            type=socket.SOCK_STREAM
+        )
+        self.mock_protocol_handler.assert_called_with(
+            fileno=conn.fileno(),
+            addr=addr,
+            **{'config': self.protocol_config}
+        )
+        self.mock_protocol_handler.return_value.setDaemon.assert_called()
+        self.mock_protocol_handler.return_value.start.assert_called()
+        sock.close.assert_called()
 
 
 class TestChunkParser(unittest.TestCase):
@@ -310,24 +320,24 @@ class TestHttpParser(unittest.TestCase):
 
     def test_build_request(self) -> None:
         self.assertEqual(
-            proxy.HttpParser.build_request(
+            proxy.build_http_request(
                 b'GET', b'http://localhost:12345', b'HTTP/1.1'),
             proxy.CRLF.join([
                 b'GET http://localhost:12345 HTTP/1.1',
                 proxy.CRLF
             ]))
         self.assertEqual(
-            proxy.HttpParser.build_request(b'GET', b'http://localhost:12345', b'HTTP/1.1',
-                                           headers={b'key': b'value'}),
+            proxy.build_http_request(b'GET', b'http://localhost:12345', b'HTTP/1.1',
+                                     headers={b'key': b'value'}),
             proxy.CRLF.join([
                 b'GET http://localhost:12345 HTTP/1.1',
                 b'key: value',
                 proxy.CRLF
             ]))
         self.assertEqual(
-            proxy.HttpParser.build_request(b'GET', b'http://localhost:12345', b'HTTP/1.1',
-                                           headers={b'key': b'value'},
-                                           body=b'Hello from proxy.py'),
+            proxy.build_http_request(b'GET', b'http://localhost:12345', b'HTTP/1.1',
+                                     headers={b'key': b'value'},
+                                     body=b'Hello from proxy.py'),
             proxy.CRLF.join([
                 b'GET http://localhost:12345 HTTP/1.1',
                 b'key: value',
@@ -336,15 +346,15 @@ class TestHttpParser(unittest.TestCase):
 
     def test_build_response(self) -> None:
         self.assertEqual(
-            proxy.HttpParser.build_response(
+            proxy.build_http_response(
                 200, reason=b'OK', protocol_version=b'HTTP/1.1'),
             proxy.CRLF.join([
                 b'HTTP/1.1 200 OK',
                 proxy.CRLF
             ]))
         self.assertEqual(
-            proxy.HttpParser.build_response(200, reason=b'OK', protocol_version=b'HTTP/1.1',
-                                            headers={b'key': b'value'}),
+            proxy.build_http_response(200, reason=b'OK', protocol_version=b'HTTP/1.1',
+                                      headers={b'key': b'value'}),
             proxy.CRLF.join([
                 b'HTTP/1.1 200 OK',
                 b'key: value',
@@ -354,9 +364,9 @@ class TestHttpParser(unittest.TestCase):
     def test_build_response_adds_content_length_header(self) -> None:
         body = b'Hello world!!!'
         self.assertEqual(
-            proxy.HttpParser.build_response(200, reason=b'OK', protocol_version=b'HTTP/1.1',
-                                            headers={b'key': b'value'},
-                                            body=body),
+            proxy.build_http_response(200, reason=b'OK', protocol_version=b'HTTP/1.1',
+                                      headers={b'key': b'value'},
+                                      body=body),
             proxy.CRLF.join([
                 b'HTTP/1.1 200 OK',
                 b'key: value',
@@ -366,19 +376,19 @@ class TestHttpParser(unittest.TestCase):
 
     def test_build_header(self) -> None:
         self.assertEqual(
-            proxy.HttpParser.build_header(
+            proxy.build_http_header(
                 b'key', b'value'), b'key: value')
 
     def test_find_line(self) -> None:
         self.assertEqual(
-            proxy.HttpParser.find_line(
+            proxy.find_http_line(
                 b'CONNECT python.org:443 HTTP/1.0\r\n\r\n'),
             (b'CONNECT python.org:443 HTTP/1.0',
              proxy.CRLF))
 
     def test_find_line_returns_None(self) -> None:
         self.assertEqual(
-            proxy.HttpParser.find_line(b'CONNECT python.org:443 HTTP/1.0'),
+            proxy.find_http_line(b'CONNECT python.org:443 HTTP/1.0'),
             (None,
              b'CONNECT python.org:443 HTTP/1.0'))
 
@@ -709,6 +719,39 @@ class TestHttpParser(unittest.TestCase):
             self.assertTrue(k in dictionary)
 
 
+class TestWebsocketFrame(unittest.TestCase):
+
+    def test_parse_with_mask(self) -> None:
+        raw = b'\x81\x85\xc6\ti\x8d\xael\x05\xe1\xa9'
+        frame = proxy.WebsocketFrame()
+        frame.parse(raw)
+        self.assertEqual(frame.fin, True)
+        self.assertEqual(frame.rsv1, False)
+        self.assertEqual(frame.rsv2, False)
+        self.assertEqual(frame.rsv3, False)
+        self.assertEqual(frame.opcode, 0x1)
+        self.assertEqual(frame.masked, True)
+        assert frame.mask is not None
+        self.assertEqual(frame.mask, b'\xc6\ti\x8d')
+        self.assertEqual(frame.payload_length, 5)
+        self.assertEqual(frame.data, b'hello')
+
+
+class TestWebsocketClient(unittest.TestCase):
+
+    @mock.patch('base64.b64encode')
+    @mock.patch('proxy.TcpConnection.new')
+    def test_handshake(self, mock_connect: mock.Mock, mock_b64encode: mock.Mock) -> None:
+        key = b'MySecretKey'
+        mock_b64encode.return_value = key
+        mock_connect.return_value.recv.return_value = \
+            proxy.Websocket.build_handshake_response(proxy.WebsocketFrame.key_to_accept(key))
+        _ = proxy.Websocket(proxy.DEFAULT_IPV4_HOSTNAME, 8899)
+        mock_connect.return_value.send.assert_called_with(
+            proxy.Websocket.build_handshake_request(key)
+        )
+
+
 class TestHttpProtocolHandler(unittest.TestCase):
     http_server = None
     http_server_port = None
@@ -745,24 +788,22 @@ class TestHttpProtocolHandler(unittest.TestCase):
         if cls.http_server_thread:
             cls.http_server_thread.join()
 
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    def setUp(self, mock_fromfd: mock.Mock) -> None:
+    def setUp(self, mock_fromfd: mock.Mock, mock_selector: mock.Mock) -> None:
         self.fileno = 10
         self._addr = ('127.0.0.1', 54382)
         self._conn = mock_fromfd.return_value
+        self.mock_selector = mock_selector
         self.proxy = proxy.ProtocolHandler(
             self.fileno, self._addr, config=self.config)
 
-    @mock.patch('select.select')
     @mock.patch('proxy.TcpServerConnection')
-    def test_http_get(self, mock_server_connection: mock.Mock,
-                      mock_select: mock.Mock) -> None:
+    def test_http_get(self, mock_server_connection: mock.Mock) -> None:
         server = mock_server_connection.return_value
         server.connect.return_value = True
-        mock_select.side_effect = [
-            ([self._conn], [], []),
-            ([self._conn], [], []),
-            ([], [server.connection], [])]
+        server.buffer_size.return_value = 0
+        self.mock_selector_for_client_read_read_server_write(self.mock_selector, server)
 
         # Send request line
         assert self.http_server_port is not None
@@ -808,18 +849,33 @@ class TestHttpProtocolHandler(unittest.TestCase):
         assert parser.code is not None
         self.assertEqual(int(parser.code), 200)
 
-    @mock.patch('select.select')
     @mock.patch('proxy.TcpServerConnection')
-    def test_http_tunnel(self, mock_server_connection: mock.Mock,
-                         mock_select: mock.Mock) -> None:
+    def test_http_tunnel(self, mock_server_connection: mock.Mock) -> None:
         server = mock_server_connection.return_value
         server.connect.return_value = True
+        server.buffer_size.return_value = 0
         server.has_buffer.side_effect = [False, False, False, True]
-        mock_select.side_effect = [
-            ([self._conn], [], []),         # client read ready
-            ([], [self._conn], []),         # client write ready
-            ([self._conn], [], []),         # client read ready
-            ([], [server.connection], [])   # server write ready
+        self.mock_selector.return_value.select.side_effect = [
+            [(selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=selectors.EVENT_READ,
+                data=None), selectors.EVENT_READ), ],
+            [(selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=0,
+                data=None), selectors.EVENT_WRITE), ],
+            [(selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=selectors.EVENT_READ,
+                data=None), selectors.EVENT_READ), ],
+            [(selectors.SelectorKey(
+                fileobj=server.connection,
+                fd=server.connection.fileno,
+                events=0,
+                data=None), selectors.EVENT_WRITE), ],
         ]
 
         assert self.http_server_port is not None
@@ -840,9 +896,8 @@ class TestHttpProtocolHandler(unittest.TestCase):
         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.Mock) -> None:
-        mock_select.return_value = ([self._conn], [], [])
+    def test_proxy_connection_failed(self) -> None:
+        self.mock_selector_for_client_read(self.mock_selector)
         self._conn.recv.return_value = proxy.CRLF.join([
             b'GET http://unknown.domain HTTP/1.1',
             b'Host: unknown.domain',
@@ -852,12 +907,12 @@ class TestHttpProtocolHandler(unittest.TestCase):
         received = self._conn.send.call_args[0][0]
         self.assertEqual(received, proxy.ProxyConnectionFailed.RESPONSE_PKT)
 
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    @mock.patch('select.select')
     def test_proxy_authentication_failed(
-            self, mock_select: mock.Mock, mock_fromfd: mock.Mock) -> None:
+            self, mock_fromfd: mock.Mock, mock_selector: mock.Mock) -> None:
         self._conn = mock_fromfd.return_value
-        mock_select.return_value = ([self._conn], [], [])
+        self.mock_selector_for_client_read(mock_selector)
         config = proxy.ProtocolConfig(
             auth_code=b'Basic %s' %
                       base64.b64encode(b'user:pass'))
@@ -875,17 +930,19 @@ class TestHttpProtocolHandler(unittest.TestCase):
             self._conn.send.call_args[0][0],
             proxy.ProxyAuthenticationFailed.RESPONSE_PKT)
 
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    @mock.patch('select.select')
     @mock.patch('proxy.TcpServerConnection')
     def test_authenticated_proxy_http_get(
             self, mock_server_connection: mock.Mock,
-            mock_select: mock.Mock,
-            mock_fromfd: mock.Mock) -> None:
+            mock_fromfd: mock.Mock,
+            mock_selector: mock.Mock) -> None:
         self._conn = mock_fromfd.return_value
-        mock_select.return_value = ([self._conn], [], [])
+        self.mock_selector_for_client_read(mock_selector)
+
         server = mock_server_connection.return_value
         server.connect.return_value = True
+        server.buffer_size.return_value = 0
 
         config = proxy.ProtocolConfig(
             auth_code=b'Basic %s' %
@@ -920,19 +977,18 @@ class TestHttpProtocolHandler(unittest.TestCase):
         ])
         self.assert_data_queued(mock_server_connection, server)
 
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    @mock.patch('select.select')
     @mock.patch('proxy.TcpServerConnection')
     def test_authenticated_proxy_http_tunnel(
             self, mock_server_connection: mock.Mock,
-            mock_select: mock.Mock,
-            mock_fromfd: mock.Mock) -> None:
+            mock_fromfd: mock.Mock,
+            mock_selector: mock.Mock) -> None:
         server = mock_server_connection.return_value
         server.connect.return_value = True
-
+        server.buffer_size.return_value = 0
         self._conn = mock_fromfd.return_value
-        mock_select.side_effect = [
-            ([self._conn], [], []), ([self._conn], [], []), ([], [server.connection], [])]
+        self.mock_selector_for_client_read_read_server_write(mock_selector, server)
 
         config = proxy.ProtocolConfig(
             auth_code=b'Basic %s' %
@@ -959,51 +1015,83 @@ class TestHttpProtocolHandler(unittest.TestCase):
         self.proxy.run_once()
         server.flush.assert_called_once()
 
+    def mock_selector_for_client_read_read_server_write(self, mock_selector: mock.Mock, server: mock.Mock) -> None:
+        mock_selector.return_value.select.side_effect = [
+            [(selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=selectors.EVENT_READ,
+                data=None), selectors.EVENT_READ), ],
+            [(selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=0,
+                data=None), selectors.EVENT_READ), ],
+            [(selectors.SelectorKey(
+                fileobj=server.connection,
+                fd=server.connection.fileno,
+                events=0,
+                data=None), selectors.EVENT_WRITE), ],
+        ]
+
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    @mock.patch('select.select')
     def test_pac_file_served_from_disk(
-            self, mock_select: mock.Mock, mock_fromfd: mock.Mock) -> None:
+            self, mock_fromfd: mock.Mock, mock_selector: mock.Mock) -> None:
         pac_file = 'proxy.pac'
         self._conn = mock_fromfd.return_value
-        mock_select.return_value = [self._conn], [], []
+        self.mock_selector_for_client_read(mock_selector)
         self.init_and_make_pac_file_request(pac_file)
         self.proxy.run_once()
         self.assertEqual(
             self.proxy.request.state,
             proxy.httpParserStates.COMPLETE)
         with open('proxy.pac', 'rb') as f:
-            self._conn.send.called_once_with(proxy.HttpParser.build_response(
+            self._conn.send.called_once_with(proxy.build_http_response(
                 200, reason=b'OK', headers={
                     b'Content-Type': b'application/x-ns-proxy-autoconfig',
                     b'Connection': b'close'
                 }, body=f.read()
             ))
 
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    @mock.patch('select.select')
     def test_pac_file_served_from_buffer(
-            self, mock_select: mock.Mock, mock_fromfd: mock.Mock) -> None:
+            self, mock_fromfd: mock.Mock, mock_selector: mock.Mock) -> None:
         self._conn = mock_fromfd.return_value
+        self.mock_selector_for_client_read(mock_selector)
         pac_file_content = b'function FindProxyForURL(url, host) { return "PROXY localhost:8899; DIRECT"; }'
-        mock_select.return_value = [self._conn], [], []
         self.init_and_make_pac_file_request(proxy.text_(pac_file_content))
         self.proxy.run_once()
         self.assertEqual(
             self.proxy.request.state,
             proxy.httpParserStates.COMPLETE)
-        self._conn.send.called_once_with(proxy.HttpParser.build_response(
+        self._conn.send.called_once_with(proxy.build_http_response(
             200, reason=b'OK', headers={
                 b'Content-Type': b'application/x-ns-proxy-autoconfig',
                 b'Connection': b'close'
             }, body=pac_file_content
         ))
 
+    def mock_selector_for_client_read(self, mock_selector: mock.Mock) -> None:
+        mock_selector.return_value.select.return_value = [(
+            selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=selectors.EVENT_READ,
+                data=None), selectors.EVENT_READ), ]
+
+    @mock.patch('selectors.DefaultSelector')
     @mock.patch('socket.fromfd')
-    @mock.patch('select.select')
     def test_default_web_server_returns_404(
-            self, mock_select: mock.Mock, mock_fromfd: mock.Mock) -> None:
+            self, mock_fromfd: mock.Mock, mock_selector: mock.Mock) -> None:
         self._conn = mock_fromfd.return_value
-        mock_select.return_value = [self._conn], [], []
+        mock_selector.return_value.select.return_value = [(
+            selectors.SelectorKey(
+                fileobj=self._conn,
+                fd=self._conn.fileno,
+                events=selectors.EVENT_READ,
+                data=None), selectors.EVENT_READ), ]
         config = proxy.ProtocolConfig()
         config.plugins = proxy.load_plugins(
             b'proxy.HttpProxyPlugin,proxy.HttpWebServerPlugin')
@@ -1036,7 +1124,6 @@ class TestHttpProtocolHandler(unittest.TestCase):
             mock_run_once.return_value = True
             self.proxy.run()
         self.assertTrue(self._conn.closed)
-        plugin.return_value.access_log.assert_called()
         plugin.return_value.on_client_connection_close.assert_called()
 
     def init_and_make_pac_file_request(self, pac_file: str) -> None:
@@ -1060,14 +1147,16 @@ class TestHttpProtocolHandler(unittest.TestCase):
         server.connect.assert_called_once()
         server.closed = False
         assert self.http_server_port is not None
-        server.queue.assert_called_once_with(proxy.CRLF.join([
+        pkt = 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,
             proxy.CRLF
-        ]))
+        ])
+        server.queue.assert_called_once_with(pkt)
+        server.buffer_size.return_value = len(pkt)
 
     def assert_data_queued_to_server(self, server: mock.Mock) -> None:
         assert self.http_server_port is not None
@@ -1082,48 +1171,18 @@ class TestHttpProtocolHandler(unittest.TestCase):
             proxy.CRLF
         ])
         self.proxy.run_once()
-        server.queue.assert_called_once_with(proxy.CRLF.join([
+
+        pkt = 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.queue.assert_called_once_with(pkt)
+        server.buffer_size.return_value = len(pkt)
         server.flush.assert_not_called()
 
 
-class TestWorker(unittest.TestCase):
-
-    @mock.patch('proxy.ProtocolHandler')
-    def setUp(self, mock_protocol_handler: mock.Mock) -> None:
-        self.pipe = multiprocessing.Pipe()
-        self.worker = proxy.Worker(
-            self.pipe[1],
-            mock_protocol_handler,
-            config=proxy.ProtocolConfig())
-        self.mock_protocol_handler = mock_protocol_handler
-
-    @mock.patch('proxy.ProtocolHandler')
-    def test_shutdown_op(self, mock_http_proxy: mock.Mock) -> None:
-        self.pipe[0].send((proxy.workerOperations.SHUTDOWN, None))
-        self.worker.run()
-        self.assertFalse(mock_http_proxy.called)
-
-    @mock.patch('proxy.recv_handle')
-    def test_spawns_http_proxy_threads(
-            self, mock_recv_handle: mock.Mock) -> None:
-        fileno = 10
-        mock_recv_handle.return_value = fileno
-        self.pipe[0].send((proxy.workerOperations.HTTP_PROTOCOL, None))
-        self.pipe[0].send((proxy.workerOperations.SHUTDOWN, None))
-        self.worker.run()
-        self.assertTrue(self.mock_protocol_handler.called)
-
-    def test_handles_work_queue_recv_connection_refused(self) -> None:
-        with mock.patch.object(self.worker.work_queue, 'recv') as mock_recv:
-            mock_recv.side_effect = ConnectionRefusedError()
-            self.assertFalse(self.worker.run_once())  # doesn't teardown
-
-
 class TestHttpRequestRejected(unittest.TestCase):
 
     def setUp(self) -> None:
@@ -1156,197 +1215,215 @@ class TestHttpRequestRejected(unittest.TestCase):
 
 class TestMain(unittest.TestCase):
 
+    @staticmethod
+    def mock_default_args(mock_args: mock.Mock) -> None:
+        mock_args.version = False
+        mock_args.cert_file = proxy.DEFAULT_CERT_FILE
+        mock_args.key_file = proxy.DEFAULT_KEY_FILE
+        mock_args.ca_key_file = proxy.DEFAULT_CA_KEY_FILE
+        mock_args.ca_cert_file = proxy.DEFAULT_CA_CERT_FILE
+        mock_args.ca_signing_key_file = proxy.DEFAULT_CA_SIGNING_KEY_FILE
+        mock_args.pid_file = proxy.DEFAULT_PID_FILE
+        mock_args.log_file = proxy.DEFAULT_LOG_FILE
+        mock_args.log_level = proxy.DEFAULT_LOG_LEVEL
+        mock_args.log_format = proxy.DEFAULT_LOG_FORMAT
+        mock_args.basic_auth = proxy.DEFAULT_BASIC_AUTH
+        mock_args.hostname = proxy.DEFAULT_IPV6_HOSTNAME
+        mock_args.port = proxy.DEFAULT_PORT
+        mock_args.num_workers = proxy.DEFAULT_NUM_WORKERS
+        mock_args.disable_http_proxy = proxy.DEFAULT_DISABLE_HTTP_PROXY
+        mock_args.enable_web_server = proxy.DEFAULT_ENABLE_WEB_SERVER
+        mock_args.pac_file = proxy.DEFAULT_PAC_FILE
+        mock_args.plugins = proxy.DEFAULT_PLUGINS
+        mock_args.server_recvbuf_size = proxy.DEFAULT_SERVER_RECVBUF_SIZE
+        mock_args.client_recvbuf_size = proxy.DEFAULT_CLIENT_RECVBUF_SIZE
+        mock_args.open_file_limit = proxy.DEFAULT_OPEN_FILE_LIMIT
+
+    @mock.patch('time.sleep')
+    @mock.patch('proxy.load_plugins')
+    @mock.patch('proxy.init_parser')
     @mock.patch('proxy.set_open_file_limit')
-    @mock.patch('proxy.WorkerPool')
+    @mock.patch('proxy.ProtocolConfig')
+    @mock.patch('proxy.AcceptorPool')
     @mock.patch('proxy.logging.basicConfig')
-    def test_log_file_setup(
+    def test_init_with_no_arguments(
             self,
-            mock_config: mock.Mock,
-            mock_multicore_dispatcher: mock.Mock,
-            mock_set_open_file_limit: mock.Mock) -> None:
-        log_file = get_temp_file('proxy.log')
-        proxy.main(['--log-file', log_file])
-        mock_set_open_file_limit.assert_called()
-        mock_multicore_dispatcher.assert_called()
-        mock_multicore_dispatcher.return_value.run.assert_called()
-        mock_config.assert_called_with(
-            filename=log_file,
-            filemode='a',
+            mock_logging_config: mock.Mock,
+            mock_acceptor_pool: mock.Mock,
+            mock_protocol_config: mock.Mock,
+            mock_set_open_file_limit: mock.Mock,
+            mock_init_parser: mock.Mock,
+            mock_load_plugins: mock.Mock,
+            mock_sleep: mock.Mock) -> None:
+        mock_sleep.side_effect = KeyboardInterrupt()
+
+        mock_args = mock_init_parser.return_value.parse_args.return_value
+        self.mock_default_args(mock_args)
+        proxy.main([])
+
+        mock_init_parser.assert_called()
+        mock_init_parser.return_value.parse_args.called_with([])
+
+        mock_load_plugins.assert_called_with(b'proxy.HttpProxyPlugin,')
+        mock_logging_config.assert_called_with(
             level=logging.INFO,
             format=proxy.DEFAULT_LOG_FORMAT
         )
+        mock_set_open_file_limit.assert_called_with(mock_args.open_file_limit)
+
+        mock_protocol_config.assert_called_with(
+            auth_code=mock_args.basic_auth,
+            backlog=mock_args.backlog,
+            ca_cert_dir=mock_args.ca_cert_dir,
+            ca_cert_file=mock_args.ca_cert_file,
+            ca_key_file=mock_args.ca_key_file,
+            ca_signing_key_file=mock_args.ca_signing_key_file,
+            certfile=mock_args.cert_file,
+            client_recvbuf_size=mock_args.client_recvbuf_size,
+            hostname=mock_args.hostname,
+            keyfile=mock_args.key_file,
+            num_workers=multiprocessing.cpu_count(),
+            pac_file=mock_args.pac_file,
+            pac_file_url_path=mock_args.pac_file_url_path,
+            port=mock_args.port,
+            server_recvbuf_size=mock_args.server_recvbuf_size,
+            disable_headers=[
+                header.lower() for header in proxy.bytes_(
+                    mock_args.disable_headers).split(proxy.COMMA) if header.strip() != b''],
+        )
+
+        mock_acceptor_pool.assert_called_with(
+            hostname=mock_protocol_config.return_value.hostname,
+            port=mock_protocol_config.return_value.port,
+            backlog=mock_protocol_config.return_value.backlog,
+            num_workers=mock_protocol_config.return_value.num_workers,
+            work_klass=proxy.ProtocolHandler,
+            config=mock_protocol_config.return_value,
+        )
+        mock_acceptor_pool.return_value.setup.assert_called()
+        mock_acceptor_pool.return_value.shutdown.assert_called()
+        mock_sleep.assert_called_with(1)
 
+    @mock.patch('time.sleep')
     @mock.patch('os.remove')
     @mock.patch('os.path.exists')
     @mock.patch('builtins.open')
-    @mock.patch('proxy.set_open_file_limit')
-    @mock.patch('proxy.WorkerPool')
-    @unittest.skipIf(
-        True,  # type: ignore
-        'This test passes while development on Intellij but fails via CLI :(')
+    @mock.patch('proxy.init_parser')
+    @mock.patch('proxy.AcceptorPool')
     def test_pid_file_is_written_and_removed(
             self,
-            mock_multicore_dispatcher,
-            mock_set_open_file_limit,
-            mock_open,
-            mock_exists,
-            mock_remove) -> None:
+            mock_acceptor_pool: mock.Mock,
+            mock_init_parser: mock.Mock,
+            mock_open: mock.Mock,
+            mock_exists: mock.Mock,
+            mock_remove: mock.Mock,
+            mock_sleep: mock.Mock) -> None:
         pid_file = get_temp_file('proxy.pid')
+        mock_sleep.side_effect = KeyboardInterrupt()
+        mock_args = mock_init_parser.return_value.parse_args.return_value
+        self.mock_default_args(mock_args)
+        mock_args.pid_file = pid_file
         proxy.main(['--pid-file', pid_file])
-        mock_set_open_file_limit.assert_called()
-        mock_multicore_dispatcher.assert_called()
-        mock_multicore_dispatcher.return_value.run.assert_called()
+        mock_init_parser.assert_called()
+        mock_acceptor_pool.assert_called()
+        mock_acceptor_pool.return_value.setup.assert_called()
         mock_open.assert_called_with(pid_file, 'wb')
         mock_open.return_value.__enter__.return_value.write.assert_called_with(
             proxy.bytes_(os.getpid()))
         mock_exists.assert_called_with(pid_file)
         mock_remove.assert_called_with(pid_file)
 
+    @mock.patch('time.sleep')
     @mock.patch('proxy.ProtocolConfig')
-    @mock.patch('proxy.set_open_file_limit')
-    @mock.patch('proxy.WorkerPool')
-    def test_main(
+    @mock.patch('proxy.AcceptorPool')
+    def test_basic_auth(
             self,
-            mock_multicore_dispatcher: mock.Mock,
-            mock_set_open_file_limit: mock.Mock,
-            mock_config: mock.Mock) -> None:
+            mock_acceptor_pool: mock.Mock,
+            mock_protocol_config: mock.Mock,
+            mock_sleep: mock.Mock) -> None:
+        mock_sleep.side_effect = KeyboardInterrupt()
         proxy.main(['--basic-auth', 'user:pass'])
-        self.assertTrue(mock_set_open_file_limit.called)
-        config = mock_config.return_value
-        mock_multicore_dispatcher.assert_called_with(
+        config = mock_protocol_config.return_value
+        mock_acceptor_pool.assert_called_with(
             hostname=config.hostname,
             port=config.port,
             backlog=config.backlog,
             num_workers=config.num_workers,
             work_klass=proxy.ProtocolHandler,
             config=config)
-        mock_config.assert_called_with(
-            auth_code=b'Basic dXNlcjpwYXNz',
-            client_recvbuf_size=proxy.DEFAULT_CLIENT_RECVBUF_SIZE,
-            server_recvbuf_size=proxy.DEFAULT_SERVER_RECVBUF_SIZE,
-            pac_file=proxy.DEFAULT_PAC_FILE,
-            pac_file_url_path=proxy.DEFAULT_PAC_FILE_URL_PATH,
-            disable_headers=proxy.DEFAULT_DISABLE_HEADERS,
-            hostname=proxy.DEFAULT_IPV6_HOSTNAME,
-            port=proxy.DEFAULT_PORT,
-            backlog=proxy.DEFAULT_BACKLOG,
-            num_workers=multiprocessing.cpu_count(),
-            keyfile=None,
-            certfile=None,
-            ca_cert_file=None,
-            ca_key_file=None,
-            ca_signing_key_file=None,
-            ca_cert_dir=None
-        )
+        self.assertEqual(mock_protocol_config.call_args[1]['auth_code'], b'Basic dXNlcjpwYXNz')
 
     @mock.patch('builtins.print')
-    @mock.patch('proxy.ProtocolConfig')
-    @mock.patch('proxy.set_open_file_limit')
-    @mock.patch('proxy.WorkerPool')
     def test_main_version(
             self,
-            mock_multicore_dispatcher: mock.Mock,
-            mock_set_open_file_limit: mock.Mock,
-            mock_config: mock.Mock,
             mock_print: mock.Mock) -> None:
         with self.assertRaises(SystemExit):
             proxy.main(['--version'])
             mock_print.assert_called_with(proxy.text_(proxy.version))
-        mock_multicore_dispatcher.assert_not_called()
-        mock_set_open_file_limit.assert_not_called()
-        mock_config.assert_not_called()
 
+    @mock.patch('time.sleep')
     @mock.patch('builtins.print')
-    @mock.patch('proxy.ProtocolConfig')
-    @mock.patch('proxy.set_open_file_limit')
-    @mock.patch('proxy.WorkerPool')
+    @mock.patch('proxy.AcceptorPool')
     @mock.patch('proxy.is_py3')
     def test_main_py3_runs(
             self,
             mock_is_py3: mock.Mock,
-            mock_multicore_dispatcher: mock.Mock,
-            mock_set_open_file_limit: mock.Mock,
-            mock_config: mock.Mock,
-            mock_print: mock.Mock) -> None:
+            mock_acceptor_pool: mock.Mock,
+            mock_print: mock.Mock,
+            mock_sleep: mock.Mock) -> None:
+        mock_sleep.side_effect = KeyboardInterrupt()
         mock_is_py3.return_value = True
         proxy.main([])
         mock_is_py3.assert_called()
         mock_print.assert_not_called()
-        mock_multicore_dispatcher.assert_called()
-        mock_set_open_file_limit.assert_called()
-        mock_config.assert_called()
+        mock_acceptor_pool.assert_called()
+        mock_acceptor_pool.return_value.setup.assert_called()
 
     @mock.patch('builtins.print')
-    @mock.patch('proxy.ProtocolConfig')
-    @mock.patch('proxy.set_open_file_limit')
-    @mock.patch('proxy.WorkerPool')
     @mock.patch('proxy.is_py3')
-    @unittest.skipIf(
-        True,  # type: ignore
-        'This test passes while development on Intellij but fails via CLI :(')
     def test_main_py2_exit(
             self,
-            mock_is_py3,
-            mock_multicore_dispatcher,
-            mock_set_open_file_limit,
-            mock_config,
-            mock_print) -> None:
+            mock_is_py3: mock.Mock,
+            mock_print: mock.Mock) -> None:
+        proxy.UNDER_TEST = False
         mock_is_py3.return_value = False
         with self.assertRaises(SystemExit):
             proxy.main([])
             mock_print.assert_called_with('DEPRECATION')
-            mock_is_py3.assert_called()
-        mock_multicore_dispatcher.assert_not_called()
-        mock_set_open_file_limit.assert_not_called()
-        mock_config.assert_not_called()
-
-    def test_text(self) -> None:
-        self.assertEqual(proxy.text_(b'hello'), 'hello')
-
-    def test_text_int(self) -> None:
-        self.assertEqual(proxy.text_(1), '1')
-
-    def test_text_nochange(self) -> None:
-        self.assertEqual(proxy.text_('hello'), 'hello')
-
-    def test_bytes(self) -> None:
-        self.assertEqual(proxy.bytes_('hello'), b'hello')
+        mock_is_py3.assert_called()
 
-    def test_bytes_int(self) -> None:
-        self.assertEqual(proxy.bytes_(1), b'1')
 
-    def test_bytes_nochange(self) -> None:
-        self.assertEqual(proxy.bytes_(b'hello'), b'hello')
+@unittest.skipIf(
+    os.name == 'nt',
+    'Open file limit tests disabled for Windows')
+class TestSetOpenFileLimit(unittest.TestCase):
 
-    @unittest.skipIf(
-        os.name == 'nt',
-        'Open file limit tests disabled for Windows')
-    @mock.patch('resource.getrlimit', return_value=(128, 1024))  # type: ignore
+    @mock.patch('resource.getrlimit', return_value=(128, 1024))
     @mock.patch('resource.setrlimit', return_value=None)
     def test_set_open_file_limit(
-            self, mock_set_rlimit, mock_get_rlimit) -> None:
+            self,
+            mock_set_rlimit: mock.Mock,
+            mock_get_rlimit: mock.Mock) -> None:
         proxy.set_open_file_limit(256)
         mock_get_rlimit.assert_called_with(resource.RLIMIT_NOFILE)
         mock_set_rlimit.assert_called_with(resource.RLIMIT_NOFILE, (256, 1024))
 
-    @unittest.skipIf(
-        os.name == 'nt',
-        'Open file limit tests disabled for Windows')
-    @mock.patch('resource.getrlimit', return_value=(256, 1024))  # type: ignore
+    @mock.patch('resource.getrlimit', return_value=(256, 1024))
     @mock.patch('resource.setrlimit', return_value=None)
     def test_set_open_file_limit_not_called(
-            self, mock_set_rlimit, mock_get_rlimit) -> None:
+            self,
+            mock_set_rlimit: mock.Mock,
+            mock_get_rlimit: mock.Mock) -> None:
         proxy.set_open_file_limit(256)
         mock_get_rlimit.assert_called_with(resource.RLIMIT_NOFILE)
         mock_set_rlimit.assert_not_called()
 
-    @unittest.skipIf(
-        os.name == 'nt',
-        'Open file limit tests disabled for Windows')
-    @mock.patch('resource.getrlimit', return_value=(256, 1024))  # type: ignore
+    @mock.patch('resource.getrlimit', return_value=(256, 1024))
     @mock.patch('resource.setrlimit', return_value=None)
-    def test_set_open_file_limit_not_called1(
-            self, mock_set_rlimit, mock_get_rlimit) -> None:
+    def test_set_open_file_limit_not_called_coz_upper_bound_check(
+            self,
+            mock_set_rlimit: mock.Mock,
+            mock_get_rlimit: mock.Mock) -> None:
         proxy.set_open_file_limit(1024)
         mock_get_rlimit.assert_called_with(resource.RLIMIT_NOFILE)
         mock_set_rlimit.assert_not_called()