Implement async variants of SR clients and serdes

Author: rohitsanj

LICENSE

@@ -652,3 +652,36 @@ For the files wingetopt.c wingetopt.h downloaded from https://github.com/alex85k
  */
 
 
+
+LICENSE.unasync
+--------------------------------------------------------------
+For unasync code in setup.py, derived from
+https://github.com/encode/httpcore/blob/ae46dfbd4330eefaa9cd6ab1560dec18a1d0bcb8/scripts/unasync.py
+
+Copyright © 2020, [Encode OSS Ltd](https://www.encode.io/).
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+* Neither the name of the copyright holder nor the names of its
+  contributors may be used to endorse or promote products derived from
+  this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

pyproject.toml

@@ -73,3 +73,6 @@ optional-dependencies.all = { file = [
     "requirements/requirements-avro.txt",
     "requirements/requirements-json.txt",
     "requirements/requirements-protobuf.txt"] }
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"

requirements/requirements-tests.txt

@@ -8,3 +8,4 @@ pytest-timeout
 requests-mock
 respx
 pytest_cov
+pytest-asyncio

setup.py

@@ -1,9 +1,13 @@
 #!/usr/bin/env python
 
 import os
-from setuptools import setup
-from distutils.core import Extension
 import platform
+import re
+import sys
+from setuptools import setup
+from setuptools import Extension
+from setuptools.command.build_py import build_py as _build_py
+from pprint import pprint
 
 work_dir = os.path.dirname(os.path.realpath(__file__))
 mod_dir = os.path.join(work_dir, 'src', 'confluent_kafka')
@@ -25,4 +29,128 @@
                             os.path.join(ext_dir, 'AdminTypes.c'),
                             os.path.join(ext_dir, 'Admin.c')])
 
-setup(ext_modules=[module])
+SUBS = [
+    ('from confluent_kafka.schema_registry.common import asyncinit', ''),
+    ('@asyncinit', ''),
+    ('import asyncio', ''),
+    ('asyncio.sleep', 'time.sleep'),
+
+    ('Async([A-Z][A-Za-z0-9_]*)', r'\2'),
+    ('_Async([A-Z][A-Za-z0-9_]*)', r'_\2'),
+    ('async_([a-z][A-Za-z0-9_]*)', r'\2'),
+
+    ('async def', 'def'),
+    ('await ', ''),
+    ('aclose', 'close'),
+    ('__aenter__', '__enter__'),
+    ('__aexit__', '__exit__'),
+    ('__aiter__', '__iter__'),    
+]
+COMPILED_SUBS = [
+    (re.compile(r'(^|\b)' + regex + r'($|\b)'), repl)
+    for regex, repl in SUBS
+]
+
+USED_SUBS = set()
+
+def unasync_line(line):
+    for index, (regex, repl) in enumerate(COMPILED_SUBS):
+        old_line = line
+        line = re.sub(regex, repl, line)
+        if old_line != line:
+            USED_SUBS.add(index)
+    return line
+
+
+def unasync_file(in_path, out_path):
+    with open(in_path, "r") as in_file:
+        with open(out_path, "w", newline="") as out_file:
+            for line in in_file.readlines():
+                line = unasync_line(line)
+                out_file.write(line)
+
+
+def unasync_file_check(in_path, out_path):
+    with open(in_path, "r") as in_file:
+        with open(out_path, "r") as out_file:
+            for in_line, out_line in zip(in_file.readlines(), out_file.readlines()):
+                expected = unasync_line(in_line)
+                if out_line != expected:
+                    print(f'unasync mismatch between {in_path!r} and {out_path!r}')
+                    print(f'Async code:         {in_line!r}')
+                    print(f'Expected sync code: {expected!r}')
+                    print(f'Actual sync code:   {out_line!r}')
+                    sys.exit(1)
+
+
+def unasync_dir(in_dir, out_dir, check_only=False):
+    for dirpath, dirnames, filenames in os.walk(in_dir):
+        for filename in filenames:
+            if not filename.endswith('.py'):
+                continue
+            rel_dir = os.path.relpath(dirpath, in_dir)
+            in_path = os.path.normpath(os.path.join(in_dir, rel_dir, filename))
+            out_path = os.path.normpath(os.path.join(out_dir, rel_dir, filename))
+            print(in_path, '->', out_path)
+            if check_only:
+                unasync_file_check(in_path, out_path)
+            else:
+                unasync_file(in_path, out_path)
+
+def unasync():
+    unasync_dir(
+        "src/confluent_kafka/schema_registry/_async",
+        "src/confluent_kafka/schema_registry/_sync",
+        check_only=False
+    )
+    unasync_dir(
+        "tests/integration/schema_registry/_async",
+        "tests/integration/schema_registry/_sync",
+        check_only=False
+    )
+
+
+    if len(USED_SUBS) != len(SUBS):
+        unused_subs = [SUBS[i] for i in range(len(SUBS)) if i not in USED_SUBS]
+
+        print("These patterns were not used:")
+        pprint(unused_subs)
+
+
+class build_py(_build_py):
+    """
+    Subclass build_py from setuptools to modify its behavior.
+
+    Convert files in _async dir from being asynchronous to synchronous
+    and saves them to the specified output directory.
+    """
+
+    def run(self):
+        self._updated_files = []
+
+        # Base class code
+        if self.py_modules:
+            self.build_modules()
+        if self.packages:
+            self.build_packages()
+            self.build_package_data()
+
+        # Our modification
+        unasync()
+
+        # Remaining base class code
+        self.byte_compile(self.get_outputs(include_bytecode=0))
+
+    def build_module(self, module, module_file, package):
+        outfile, copied = super().build_module(module, module_file, package)
+        if copied:
+            self._updated_files.append(outfile)
+        return outfile, copied
+
+
+setup(
+    ext_modules=[module],
+    cmdclass={
+        'build_py': build_py,
+    }
+)

src/confluent_kafka/schema_registry/__init__.py

@@ -31,7 +31,6 @@
   Schema,
   SchemaRegistryClient,
   SchemaRegistryError,
-  AsyncSchemaRegistryClient,
   SchemaReference,
   ServerConfig
 )
@@ -57,7 +56,6 @@
   "RuleSet",
   "Schema",
   "SchemaRegistryClient",
-  "AsyncSchemaRegistryClient"
   "SchemaRegistryError",
   "SchemaReference",
   "ServerConfig",

src/confluent_kafka/schema_registry/_async/__init__.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2020 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#