Aiohtt>=2.x compatibility

Thanks @thecodingrobot and @evertlammerts for the work on this patch!

Author: honzakral

elasticsearch_async/connection.py

@@ -1,13 +1,12 @@
 import asyncio
 
 import aiohttp
-from aiohttp.errors import FingerprintMismatch, ClientError
+from aiohttp.client_exceptions import ServerFingerprintMismatch
 
 from elasticsearch.exceptions import ConnectionError, ConnectionTimeout, SSLError
 from elasticsearch.connection import Connection
 from elasticsearch.compat import urlencode
 
-from .helpers import ensure_future
 
 class AIOHttpConnection(Connection):
     def __init__(self, host='localhost', port=9200, http_auth=None,
@@ -26,10 +25,10 @@ def __init__(self, host='localhost', port=9200, http_auth=None,
 
         self.session = aiohttp.ClientSession(
             auth=http_auth,
+            conn_timeout=self.timeout,
             connector=aiohttp.TCPConnector(
                 loop=self.loop,
                 verify_ssl=verify_certs,
-                conn_timeout=self.timeout,
                 use_dns_cache=use_dns_cache,
             )
         )
@@ -40,7 +39,7 @@ def __init__(self, host='localhost', port=9200, http_auth=None,
         )
 
     def close(self):
-        return ensure_future(self.session.close())
+        return self.session.close()
 
     @asyncio.coroutine
     def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
@@ -59,7 +58,7 @@ def perform_request(self, method, url, params=None, body=None, timeout=None, ign
 
         except Exception as e:
             self.log_request_fail(method, url, url_path, body, self.loop.time() - start, exception=e)
-            if isinstance(e, FingerprintMismatch):
+            if isinstance(e, ServerFingerprintMismatch):
                 raise SSLError('N/A', str(e), e)
             if isinstance(e, asyncio.TimeoutError):
                 raise ConnectionTimeout('TIMEOUT', str(e), e)

test_elasticsearch_async/conftest.py

@@ -1,39 +1,38 @@
 import asyncio
 import json
-from urllib.parse import urlparse, parse_qsl
-
-from aiohttp.server import ServerHttpProtocol
-from aiohttp import Response
 
 from pytest import yield_fixture, fixture
 
+import aiohttp.web
 from elasticsearch_async import AIOHttpConnection, AsyncElasticsearch
 
 
 @yield_fixture
 def connection(event_loop, server, port):
     connection = AIOHttpConnection(port=port, loop=event_loop)
     yield connection
-    event_loop.run_until_complete(connection.close())
+    connection.close()
+
+
+class DummyElasticsearch(aiohttp.web.Server):
 
-class DummyElasticsearch(ServerHttpProtocol):
     def __init__(self, **kwargs):
-        super().__init__(**kwargs)
+        super().__init__(handler=self.handler, **kwargs)
         self._responses = {}
         self.calls = []
 
     def register_response(self, path, response={}, status=200):
         self._responses[path] = status, response
 
     @asyncio.coroutine
-    def handle_request(self, message, payload):
-        url = urlparse(message.path)
+    def handler(self, request):
+        url = request.url
 
-        params = dict(parse_qsl(url.query))
-        body = yield from payload.read()
+        params = dict(request.query)
+        body = yield from request.read()
         body = json.loads(body.decode('utf-8')) if body else ''
 
-        self.calls.append((message.method, url.path, body, params))
+        self.calls.append((request.method, url.path, body, params))
 
         if url.path in self._responses:
             status, body = self._responses.pop(url.path)
@@ -42,18 +41,16 @@ def handle_request(self, message, payload):
         else:
             status = 200
             body = {
-                'method': message.method,
+                'method': request.method,
                 'params': params,
                 'path': url.path,
                 'body': body
             }
 
-        out = json.dumps(body).encode('utf-8')
+        out = json.dumps(body)
+
+        return aiohttp.web.Response(body=out, status=status, content_type='application/json')
 
-        response = Response(self.writer, status)
-        response.send_headers()
-        response.write(out)
-        yield from response.write_eof()
 
 i = 0
 @fixture
@@ -62,12 +59,13 @@ def port():
     i += 1
     return 8080 + i
 
-@fixture
+@yield_fixture
 def server(event_loop, port):
     server = DummyElasticsearch(debug=True, keep_alive=75)
-    f = event_loop.create_server(lambda: server, '127.0.0.1', port)
+    f = event_loop.create_server(server, '127.0.0.1', port)
     event_loop.run_until_complete(f)
-    return server
+    yield server
+    event_loop.run_until_complete(server.shutdown(timeout=.5))
 
 @yield_fixture
 def client(event_loop, server, port):