Expose selector type as config option (#764)

1 files changed, 5 insertions, 1 deletions

diff --git a/kafka/client_async.py b/kafka/client_async.py
index 2700069..9439451 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -64,6 +64,7 @@ class KafkaClient(object):
         'ssl_crlfile': None,
         'api_version': None,
         'api_version_auto_timeout_ms': 2000,
+        'selector': selectors.DefaultSelector,
     }
     API_VERSIONS = [
         (0, 10),
@@ -135,6 +136,9 @@ class KafkaClient(object):
             api_version_auto_timeout_ms (int): number of milliseconds to throw a
                 timeout exception from the constructor when checking the broker
                 api version. Only applies if api_version is None
+            selector (selectors.BaseSelector): Provide a specific selector
+                implementation to use for I/O multiplexing.
+                Default: selectors.DefaultSelector
         """
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
@@ -150,7 +154,7 @@ class KafkaClient(object):
         self._topics = set() # empty set will fetch all topic metadata
         self._metadata_refresh_in_progress = False
         self._last_no_node_available_ms = 0
-        self._selector = selectors.DefaultSelector()
+        self._selector = self.config['selector']()
         self._conns = {}
         self._connecting = set()
         self._refresh_on_disconnects = True