Fix some Websocket issues, remove Index.head relation

.bumpversion.cfg

@@ -15,3 +15,7 @@ replace = __version__ = '{new_version}'
 [bumpversion:glob:src/*/docker/Dockerfile]
 search = FROM dipdup/dipdup:{current_version}
 replace = FROM dipdup/dipdup:{new_version}
+
+[bumpversion:glob:CHANGELOG.md]
+search = '## [unreleased]'
+replace = '## {new_version}'

CHANGELOG.md

@@ -15,6 +15,7 @@
 * Fixed creation of missing project package on `init`.
 * Fixed invalid handler callbacks generated on `init`.
 * Fixed detection of existing types in the project.
+* Fixed race condition caused by event emitter concurrency.
 
 ## 3.0.1 - 2021-09-24
 

pyproject.toml

@@ -30,7 +30,6 @@ tortoise-orm = "0.17.5"
 pydantic = "^1.8.1"
 aiosignalrcore = "^0.9.2"
 fcache = "^0.4.7"
-pyee = "^8.1.0"
 APScheduler = "^3.7.0"
 sentry-sdk = "^1.1.0"
 pyhumps = "^3.0.2"

src/dipdup/datasources/datasource.py

@@ -1,29 +1,19 @@
 from abc import abstractmethod
-from enum import Enum
 from logging import Logger
-from typing import Awaitable, List, Protocol
-
-from pyee import AsyncIOEventEmitter  # type: ignore
+from typing import Awaitable, List, Protocol, Set
 
 from dipdup.config import HTTPConfig
 from dipdup.http import HTTPGateway
 from dipdup.models import BigMapData, HeadBlockData, OperationData
 
 
-class EventType(Enum):
-    operations = 'operatitions'
-    big_maps = 'big_maps'
-    rollback = 'rollback'
-    head = 'head'
-
-
 class OperationsCallback(Protocol):
-    def __call__(self, datasource: 'IndexDatasource', operations: List[OperationData], block: HeadBlockData) -> Awaitable[None]:
+    def __call__(self, datasource: 'IndexDatasource', operations: List[OperationData]) -> Awaitable[None]:
         ...
 
 
 class BigMapsCallback(Protocol):
-    def __call__(self, datasource: 'IndexDatasource', big_maps: List[BigMapData], block: HeadBlockData) -> Awaitable[None]:
+    def __call__(self, datasource: 'IndexDatasource', big_maps: List[BigMapData]) -> Awaitable[None]:
         ...
 
 
@@ -33,7 +23,7 @@ def __call__(self, datasource: 'IndexDatasource', from_level: int, to_level: int
 
 
 class HeadCallback(Protocol):
-    def __call__(self, datasource: 'IndexDatasource', block: HeadBlockData) -> Awaitable[None]:
+    def __call__(self, datasource: 'IndexDatasource', head: HeadBlockData) -> Awaitable[None]:
         ...
 
 
@@ -45,39 +35,38 @@ async def run(self) -> None:
         ...
 
 
-class IndexDatasource(Datasource, AsyncIOEventEmitter):
+class IndexDatasource(Datasource):
     def __init__(self, url: str, http_config: HTTPConfig) -> None:
-        HTTPGateway.__init__(self, url, http_config)
-        AsyncIOEventEmitter.__init__(self)
-
-    def on(self, event, f=None) -> None:
-        raise RuntimeError('Do not use `on` directly')
+        super().__init__(url, http_config)
+        self._on_head: Set[HeadCallback] = set()
+        self._on_operations: Set[OperationsCallback] = set()
+        self._on_big_maps: Set[BigMapsCallback] = set()
+        self._on_rollback: Set[RollbackCallback] = set()
 
-    def emit(self, event: str, *args, **kwargs) -> None:
-        if event not in ('new_listener', 'error'):
-            raise RuntimeError('Do not use `emit` directly')
-        super().emit(event, *args, **kwargs)
+    def on_head(self, fn: HeadCallback) -> None:
+        self._on_head.add(fn)
 
     def on_operations(self, fn: OperationsCallback) -> None:
-        super().on(EventType.operations, fn)
+        self._on_operations.add(fn)
 
     def on_big_maps(self, fn: BigMapsCallback) -> None:
-        super().on(EventType.big_maps, fn)
+        self._on_big_maps.add(fn)
 
     def on_rollback(self, fn: RollbackCallback) -> None:
-        super().on(EventType.rollback, fn)
-
-    def on_head(self, fn: HeadCallback) -> None:
-        super().on(EventType.head, fn)
+        self._on_rollback.add(fn)
 
-    def emit_operations(self, operations: List[OperationData], block: HeadBlockData) -> None:
-        super().emit(EventType.operations, datasource=self, operations=operations, block=block)
+    async def emit_head(self, head: HeadBlockData) -> None:
+        for fn in self._on_head:
+            await fn(self, head)
 
-    def emit_big_maps(self, big_maps: List[BigMapData], block: HeadBlockData) -> None:
-        super().emit(EventType.big_maps, datasource=self, big_maps=big_maps, block=block)
+    async def emit_operations(self, operations: List[OperationData]) -> None:
+        for fn in self._on_operations:
+            await fn(self, operations)
 
-    def emit_rollback(self, from_level: int, to_level: int) -> None:
-        super().emit(EventType.rollback, datasource=self, from_level=from_level, to_level=to_level)
+    async def emit_big_maps(self, big_maps: List[BigMapData]) -> None:
+        for fn in self._on_big_maps:
+            await fn(self, big_maps)
 
-    def emit_head(self, block: HeadBlockData) -> None:
-        super().emit(EventType.head, datasource=self, block=block)
+    async def emit_rollback(self, from_level: int, to_level: int) -> None:
+        for fn in self._on_rollback:
+            await fn(self, from_level, to_level)