async version

Author: kulame

tests/test_pipeline.py

@@ -10,6 +10,6 @@
 config.read(".env")
 
 def test_kafkasink() -> None:
-    database_url = config['kafka'].get("url")
+    database_url = config['sink'].get("url")
     sink = KafkaSink(database_url)
 

tests/test_sql.py

@@ -31,14 +31,14 @@ def run_sync(*args, **kwargs):
 
 
 def test_parse() -> None:
-    database_url = config['mysql'].get("url")
+    database_url = config['source'].get("url")
     db = parse_url(database_url)
     debug(db)
 
 @async_adapter
 async def test_mock() -> None:
     meta = sqlalchemy.MetaData()
-    database_url = config['mysql'].get("url")
+    database_url = config['source'].get("url")
     debug(database_url)
 
     engine = sqlalchemy.create_engine(database_url)

unimeta/pipeline.py

@@ -18,7 +18,7 @@
 import requests
 from concurrent.futures import ProcessPoolExecutor
 import aiohttp
-from aiokafka import AIOKafkaProducer
+from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
 from aioch import Client
 
 
@@ -102,6 +102,7 @@ async def publish(self, event):
                    columns=",".join(columns))
         try:
             await self.client.execute(sql,[event.data])
+            logger.success("{sql} insert clickhouse".format(sql=sql))
         except:
             logger.error(event.data)
             logger.exception("what?")
@@ -118,13 +119,13 @@ def delivery_report(err, msg):
 class KafkaSink(Sink):
 
     def __init__(self, database_url):
-        Source.__init__(self)
+        Sink.__init__(self)
         settings = parse_url(database_url)
         loop = asyncio.get_event_loop()
+        bootstrap = '{host}:{port}'.format(host=settings['host'],port=settings['port'])
         self.producer = AIOKafkaProducer(
             loop=loop,
-            bootstrap_servers = '{host}:{port}'.format(host=settings['host'],port=settings['port']),
-            
+            bootstrap_servers = bootstrap
         )
         self.producer.start()
         self.topic = settings['name']
@@ -142,6 +143,31 @@ async def publish(self, event):
 
         #logger.success("publish {event}　success".format(event=str(event)))
 
+class KafkaSource(Source):
+
+    def __init__(self, database_url):
+        Source.__init__(self)
+        settings = parse_url(database_url)
+        name = settings['name']
+        user = settings['user']
+        if user is None:
+            user = "default"
+        loop = asyncio.get_event_loop()
+        bootstrap = '{host}:{port}'.format(host=settings['host'],port=settings['port'])
+        self.consumer = AIOKafkaConsumer(
+            name, loop=loop,
+            bootstrap_servers = bootstrap,
+            group_id = user
+        )
+
+    async def start(self):
+        await self.consumer.start()
+
+    async def subscribe(self):
+        async for msg in self.consumer:
+            print(msg)
+            yield msg
+        
 
 
 class MetaServer():
@@ -186,6 +212,8 @@ def __init__(self,source:str, sink:str, meta:str):
         sconf = parse_url(source)
         if sconf['scheme'] == 'mysql':
             self.source = MysqlSource(source)
+        elif sconf['scheme'] == 'kafka':
+            self.source = KafkaSource(source)
         else:
             raise Exception("unregister source")
 
@@ -223,5 +251,7 @@ def rebuild_table(self, table):
     async def sync(self):
         await self.sink.start()
         async for event in self.source.subscribe():
+            debug(event)
+            continue
             await self.metaserver.reg(event)
             await self.sink.publish(event)

unimeta/table.py

@@ -126,6 +126,9 @@ def avro_types(self) -> Optional[str,List[str]]:
             return "string"
 
 
+class PhoneField(StringColumn):
+    pass
+
 
 
 class TextColumn(Column):