Add subscription to big_map updates (#45)

Author: droserasprout

src/demo_tezos_domains_big_map/dipdup.yml

@@ -9,12 +9,19 @@ contracts:
   edo_name_registry:
     address: KT1JJbWfW8CHUY95hG9iq2CEMma1RiKhMHDR
     typename: name_registry
+  mainnet_name_registry:
+    address: KT1GBZmSxmnKJXGMdMLbugPfLyUPmuLSMwKS
+    typename: name_registry
 
 datasources:
   tzkt_staging_edo:
     kind: tzkt
     url: ${TZKT_URL:-https://staging.api.edo2net.tzkt.io}
 
+  tzkt_staging_mainnet:
+    kind: tzkt
+    url: ${TZKT_URL:-https://staging.api.tzkt.io}
+
 templates:
   tezos_domains_big_map:
     kind: big_map
@@ -33,3 +40,9 @@ indexes:
     values:
       datasource: tzkt_staging_edo
       name_registry: edo_name_registry
+
+  tezos_domains_big_map_mainnet:
+    template: tezos_domains_big_map
+    values:
+      datasource: tzkt_staging_mainnet
+      name_registry: mainnet_name_registry

src/dipdup/config.py

@@ -565,7 +565,10 @@ def __post_init_post_parse__(self):
 
             elif isinstance(index_config, BigMapIndexConfig):
                 if isinstance(index_config.datasource, str):
-                    index_config.datasource = self.datasources[index_config.datasource]
+                    try:
+                        index_config.datasource = self.datasources[index_config.datasource]
+                    except KeyError as e:
+                        raise ConfigurationError(f'Datasource `{index_config.datasource}` not found in `datasources` config section') from e
 
                 for handler in index_config.handlers:
                     callback_patterns[handler.callback].append(handler.pattern)

src/dipdup/datasources/tzkt/datasource.py

@@ -330,8 +330,7 @@ async def on_connect(self):
         for address, types in self._operation_subscriptions.items():
             await self.subscribe_to_operations(address, types)
         for address, paths in self._big_map_subscriptions.items():
-            for path in paths:
-                await self.subscribe_to_big_maps(address, paths)
+            await self.subscribe_to_big_maps(address, paths)
 
     def on_error(self, message: CompletionMessage):
         raise Exception(message.error)
@@ -352,8 +351,22 @@ async def subscribe_to_operations(self, address: str, types: List[OperationType]
             ],
         )
 
-    async def subscribe_to_big_maps(self, address: Address, path: Path) -> None:
-        self._logger.info('Subscribing to %s, %s', address, path)
+    async def subscribe_to_big_maps(self, address: Address, paths: List[Path]) -> None:
+        self._logger.info('Subscribing to big map updates of %s, %s', address, paths)
+
+        while self._get_client().transport.state != ConnectionState.connected:
+            await asyncio.sleep(0.1)
+
+        for path in paths:
+            await self._get_client().send(
+                'SubscribeToBigMaps',
+                [
+                    {
+                        'address': address,
+                        'path': path,
+                    }
+                ],
+            )
 
     async def fetch_operations(self, index_config: OperationIndexConfig, last_level: int) -> None:
         self._logger.info('Fetching operations prior to level %s', last_level)
@@ -561,7 +574,7 @@ async def add_operation_subscription(self, address: str, types: Optional[List[Op
     async def add_big_map_subscription(self, address: str, path: str) -> None:
         if address not in self._big_map_subscriptions:
             self._big_map_subscriptions[address] = []
-        self._big_map_subscriptions[address].append('path')
+        self._big_map_subscriptions[address].append(path)
 
     async def on_operation_match(
         self,