Made kafka generic for output, now it can be in format of kafka/127.0.0.1:9092

Author: ayush3298

blockchainetl/jobs/exporters/kafka_expoerter.py

@@ -9,11 +9,18 @@
 
 class KafkaItemExporter:
 
-    def __init__(self, connection_url, item_type_to_topic_mapping, converters=()):
-        self.connection_url = connection_url
+    def __init__(self, output, item_type_to_topic_mapping, converters=()):
         self.item_type_to_topic_mapping = item_type_to_topic_mapping
         self.converter = CompositeItemConverter(converters)
-        self.producer = KafkaProducer(bootstrap_servers=connection_url)
+        self.connection_url = self.get_connection_url(output)
+        print(self.connection_url)
+        self.producer = KafkaProducer(bootstrap_servers=self.connection_url)
+
+    def get_connection_url(self, output):
+        try:
+            return output.split('/')[1]
+        except KeyError:
+            raise Exception('Invalid kafka output param, It should be in format of "kafka/127.0.0.1:9092"')
 
     def open(self):
         pass
@@ -26,6 +33,7 @@ def export_item(self, item):
         item_type = item.get('type')
         if item_type is not None and item_type in self.item_type_to_topic_mapping:
             data = json.dumps(item).encode('utf-8')
+            print(data)
             return self.producer.send(self.item_type_to_topic_mapping[item_type], value=data)
         else:
             logging.warning('Topic for item type "{}" is not configured.'.format(item_type))

ethereumetl/cli/stream.py

@@ -40,7 +40,7 @@
 @click.option('-o', '--output', type=str,
               help='Either Google PubSub topic path e.g. projects/your-project/topics/crypto_ethereum; '
                    'or Postgres connection url e.g. postgresql+pg8000://postgres:admin@127.0.0.1:5432/ethereum. '
-                   'or kafka, in case of kafka, specify connection_url '
+                   'or kafka, output name and connection ip:port e.g. kafka/127.0.0.1:9092 '
                    'If not specified will print to console')
 @click.option('-s', '--start-block', default=None, show_default=True, type=int, help='Start block')
 @click.option('-e', '--entity-types', default=','.join(EntityType.ALL_FOR_INFURA), show_default=True, type=str,
@@ -51,10 +51,8 @@
 @click.option('-w', '--max-workers', default=5, show_default=True, type=int, help='The number of workers')
 @click.option('--log-file', default=None, show_default=True, type=str, help='Log file')
 @click.option('--pid-file', default=None, show_default=True, type=str, help='pid file')
-@click.option('--connection-url', default=None, show_default=True, type=str, help='Connection url for file, required for kafka')
 def stream(last_synced_block_file, lag, provider_uri, output, start_block, entity_types,
-           period_seconds=10, batch_size=2, block_batch_size=10, max_workers=5, log_file=None, pid_file=None,
-           connection_url=None):
+           period_seconds=10, batch_size=2, block_batch_size=10, max_workers=5, log_file=None, pid_file=None):
     """Streams all data types to console or Google Pub/Sub."""
     configure_logging(log_file)
     configure_signals()
@@ -71,7 +69,7 @@ def stream(last_synced_block_file, lag, provider_uri, output, start_block, entit
 
     streamer_adapter = EthStreamerAdapter(
         batch_web3_provider=ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=True)),
-        item_exporter=create_item_exporter(output, connection_url),
+        item_exporter=create_item_exporter(output),
         batch_size=batch_size,
         max_workers=max_workers,
         entity_types=entity_types

ethereumetl/streaming/item_exporter_creator.py

@@ -31,7 +31,7 @@ def create_item_exporters(outputs):
     return MultiItemExporter(item_exporters)
 
 
-def create_item_exporter(output, connection_url):
+def create_item_exporter(output):
     item_exporter_type = determine_item_exporter_type(output)
     if item_exporter_type == ItemExporterType.PUBSUB:
         from blockchainetl.jobs.exporters.google_pubsub_item_exporter import GooglePubSubItemExporter
@@ -77,7 +77,7 @@ def create_item_exporter(output, connection_url):
         item_exporter = ConsoleItemExporter()
     elif item_exporter_type == ItemExporterType.KAFKA:
         from blockchainetl.jobs.exporters.kafka_expoerter import KafkaItemExporter
-        item_exporter = KafkaItemExporter(connection_url, item_type_to_topic_mapping={
+        item_exporter = KafkaItemExporter(output, item_type_to_topic_mapping={
             'block': 'blocks',
             'transaction': 'transactions',
             'log': 'logs',