feat(events): implement websocket endpoints and message handling

hathor/builder.py

@@ -26,6 +26,7 @@
 from twisted.web import server
 from twisted.web.resource import Resource
 
+from hathor.event import EventManager
 from hathor.exception import BuilderError
 from hathor.indexes import IndexesManager
 from hathor.manager import HathorManager
@@ -56,6 +57,7 @@ def create_manager(self, reactor: PosixReactorBase, args: Namespace) -> HathorMa
         from hathor.conf.get_settings import get_settings_module
         from hathor.daa import TestMode, _set_test_mode
         from hathor.event.storage import EventMemoryStorage, EventRocksDBStorage, EventStorage
+        from hathor.event.websocket.factory import EventWebsocketFactory
         from hathor.p2p.netfilter.utils import add_peer_id_blacklist
         from hathor.p2p.peer_discovery import BootstrapPeerDiscovery, DNSPeerDiscovery
         from hathor.storage import RocksDBStorage
@@ -90,13 +92,15 @@ def create_manager(self, reactor: PosixReactorBase, args: Namespace) -> HathorMa
 
         tx_storage: TransactionStorage
         rocksdb_storage: RocksDBStorage
-        event_storage: Optional[EventStorage] = None
+        self.event_storage: Optional[EventStorage] = None
+        self.event_ws_factory: Optional[EventWebsocketFactory] = None
+
         if args.memory_storage:
             self.check_or_raise(not args.data, '--data should not be used with --memory-storage')
             # if using MemoryStorage, no need to have cache
             tx_storage = TransactionMemoryStorage()
             if args.x_enable_event_queue:
-                event_storage = EventMemoryStorage()
+                self.event_storage = EventMemoryStorage()
             self.check_or_raise(not args.x_rocksdb_indexes, 'RocksDB indexes require RocksDB data')
             self.log.info('with storage', storage_class=type(tx_storage).__name__)
         else:
@@ -109,7 +113,7 @@ def create_manager(self, reactor: PosixReactorBase, args: Namespace) -> HathorMa
                                                    with_index=(not args.cache),
                                                    use_memory_indexes=args.memory_indexes)
             if args.x_enable_event_queue:
-                event_storage = EventRocksDBStorage(rocksdb_storage)
+                self.event_storage = EventRocksDBStorage(rocksdb_storage)
 
         self.log.info('with storage', storage_class=type(tx_storage).__name__, path=args.data)
         if args.cache:
@@ -135,6 +139,17 @@ def create_manager(self, reactor: PosixReactorBase, args: Namespace) -> HathorMa
 
         pubsub = PubSubManager(reactor)
 
+        event_manager: Optional[EventManager] = None
+        if args.x_enable_event_queue:
+            assert self.event_storage is not None, 'cannot create EventManager without EventStorage'
+            self.event_ws_factory = EventWebsocketFactory(reactor, self.event_storage)
+            event_manager = EventManager(
+                event_storage=self.event_storage,
+                event_ws_factory=self.event_ws_factory,
+                pubsub=pubsub,
+                reactor=reactor
+            )
+
         if args.wallet_index and tx_storage.indexes is not None:
             self.log.debug('enable wallet indexes')
             self.enable_wallet_index(tx_storage.indexes, pubsub)
@@ -150,7 +165,7 @@ def create_manager(self, reactor: PosixReactorBase, args: Namespace) -> HathorMa
             network=network,
             hostname=hostname,
             tx_storage=tx_storage,
-            event_storage=event_storage,
+            event_manager=event_manager,
             wallet=self.wallet,
             stratum_port=args.stratum,
             ssl=True,
@@ -481,6 +496,10 @@ def create_resources(self, args: Namespace) -> server.Site:
 
         ws_factory.subscribe(self.manager.pubsub)
 
+        # Event websocket resource
+        if args.x_enable_event_queue and self.event_ws_factory is not None:
+            root.putChild(b'event_ws', WebSocketResource(self.event_ws_factory))
+
         # Websocket stats resource
         root.putChild(b'websocket_stats', WebsocketStatsResource(ws_factory))
 

hathor/conf/unittests.py

@@ -32,4 +32,5 @@
     GENESIS_TX2_HASH=bytes.fromhex('33e14cb555a96967841dcbe0f95e9eab5810481d01de8f4f73afb8cce365e869'),
     REWARD_SPEND_MIN_BLOCKS=10,
     SLOW_ASSERTS=True,
+    ENABLE_EVENT_QUEUE_FEATURE=True,
 )

hathor/event/base_event.py

@@ -12,24 +12,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from dataclasses import dataclass
 from typing import Dict, Optional
 
+from pydantic import NonNegativeInt
 
-@dataclass
-class BaseEvent:
+from hathor.utils.pydantic import BaseModel
+
+
+class BaseEvent(BaseModel):
     # Full node id, because different full nodes can have different sequences of events
     peer_id: str
     # Event unique id, determines event order
-    id: int
+    id: NonNegativeInt
     # Timestamp in which the event was emitted, this follows the unix_timestamp format, it's only informative, events
     # aren't guaranteed to always have sequential timestamps, for example, if the system clock changes between two
     # events it's possible that timestamps will temporarily decrease.
     timestamp: float
     # One of the event types
-    type: str
+    type: str  # TODO: Convert type and data to enum and classes
     # Variable for event type
     data: Dict
     # Used to link events, for example, many TX_METADATA_CHANGED will have the same group_id when they belong to the
     # same reorg process
-    group_id: Optional[int] = None
+    group_id: Optional[NonNegativeInt] = None

hathor/event/event_manager.py

@@ -18,6 +18,7 @@
 
 from hathor.event.base_event import BaseEvent
 from hathor.event.storage import EventStorage
+from hathor.event.websocket import EventWebsocketFactory
 from hathor.pubsub import EventArguments, HathorEvents, PubSubManager
 from hathor.util import Reactor
 
@@ -85,61 +86,133 @@ def _extract_reorg(args: EventArguments) -> Dict[str, Any]:
 }
 
 
-def build_event_data(event: HathorEvents, event_args: EventArguments) -> Dict[str, Any]:
+def _build_event_data(event_type: HathorEvents, event_args: EventArguments) -> Dict[str, Any]:
     """Extract and build event data from event_args for a given event type."""
-    event_extract_fn = _EVENT_EXTRACT_MAP.get(event)
+    event_extract_fn = _EVENT_EXTRACT_MAP.get(event_type)
     if event_extract_fn is None:
-        raise ValueError(f'The given event type ({event}) is not a supported event')
+        raise ValueError(f'The given event type ({event_type}) is not a supported event')
     return event_extract_fn(event_args)
 
 
 class EventManager:
-    def __init__(self, event_storage: EventStorage, reactor: Reactor, peer_id: str):
+    """Class that manages integration events.
+
+    Events are received from PubSub, persisted on the storage and sent to WebSocket clients.
+    """
+
+    _peer_id: str
+
+    def __init__(
+        self,
+        event_storage: EventStorage,
+        event_ws_factory: EventWebsocketFactory,
+        pubsub: PubSubManager,
+        reactor: Reactor
+    ):
         self.log = logger.new()
-        self.clock = reactor
-        self.event_storage = event_storage
-        last_event = event_storage.get_last_event()
-        last_event_type = HathorEvents(last_event.type) if last_event is not None else None
+
+        self._clock = reactor
+        self._event_storage = event_storage
+        self._event_ws_factory = event_ws_factory
+        self._pubsub = pubsub
+
+        self._last_event = self._event_storage.get_last_event()
+        self._last_existing_group_id = self._event_storage.get_last_group_id()
+
+        self._assert_closed_event_group()
+        self._subscribe_events()
+
+    def start(self, peer_id: str) -> None:
+        self._peer_id = peer_id
+        self._event_ws_factory.start()
+
+    def stop(self):
+        self._event_ws_factory.stop()
+
+    def _assert_closed_event_group(self):
         # XXX: we must check that the last event either does not belong to an event group or that it just closed an
         #      event group, because we cannot resume an open group of events that wasn't properly closed before exit
         assert (
-            last_event is None or
-            last_event.group_id is None or
-            last_event_type in _GROUP_END_EVENTS
+            self._event_group_is_closed()
         ), 'an unclosed event group was detected, which indicates the node crashed, cannot resume'
-        self._next_event_id = 0 if last_event is None else last_event.id + 1
-        last_group_id = event_storage.get_last_group_id()
-        self._next_group_id = 0 if last_group_id is None else last_group_id + 1
-        self._current_group_id: Optional[int] = None
-        self._peer_id = peer_id
 
-    def subscribe(self, pubsub: PubSubManager) -> None:
+    def _event_group_is_closed(self):
+        return (
+            self._last_event is None or
+            self._last_event.group_id is None or
+            HathorEvents(self._last_event.type) in _GROUP_END_EVENTS
+        )
+
+    def _subscribe_events(self):
         """ Subscribe to defined events for the pubsub received
         """
         for event in _SUBSCRIBE_EVENTS:
-            pubsub.subscribe(event, self._persist_event)
+            self._pubsub.subscribe(event, self._handle_event)
+
+    def _handle_event(self, event_type: HathorEvents, event_args: EventArguments) -> None:
+        create_event_fn: Callable[[HathorEvents, EventArguments], BaseEvent]
 
-    def _persist_event(self, event: HathorEvents, event_args: EventArguments) -> None:
-        group_id: Optional[int]
-        if event in _GROUP_START_EVENTS:
-            assert self._current_group_id is None, 'cannot start an event group before the last one is ended'
-            group_id = self._next_group_id
+        if event_type in _GROUP_START_EVENTS:
+            create_event_fn = self._create_group_start_event
+        elif event_type in _GROUP_END_EVENTS:
+            create_event_fn = self._create_group_end_event
         else:
-            group_id = self._current_group_id
-        if event in _GROUP_END_EVENTS:
-            assert self._current_group_id is not None, 'cannot end group twice'
-        event_to_store = BaseEvent(
-            id=self._next_event_id,
+            create_event_fn = self._create_non_group_edge_event
+
+        event = create_event_fn(event_type, event_args)
+
+        self._event_storage.save_event(event)
+        self._event_ws_factory.broadcast_event(event)
+
+        self._last_event = event
+
+    def _create_group_start_event(self, event_type: HathorEvents, event_args: EventArguments) -> BaseEvent:
+        assert self._event_group_is_closed(), 'A new event group cannot be started as one is already in progress.'
+
+        new_group_id = 0 if self._last_existing_group_id is None else self._last_existing_group_id + 1
+
+        self._last_existing_group_id = new_group_id
+
+        return self._create_event(
+            event_type=event_type,
+            event_args=event_args,
+            group_id=new_group_id,
+        )
+
+    def _create_group_end_event(self, event_type: HathorEvents, event_args: EventArguments) -> BaseEvent:
+        assert self._last_event is not None, 'Cannot end event group if there are no events.'
+        assert not self._event_group_is_closed(), 'Cannot end event group as none is in progress.'
+
+        return self._create_event(
+            event_type=event_type,
+            event_args=event_args,
+            group_id=self._last_event.group_id,
+        )
+
+    def _create_non_group_edge_event(self, event_type: HathorEvents, event_args: EventArguments) -> BaseEvent:
+        group_id = None
+
+        if not self._event_group_is_closed():
+            assert self._last_event is not None, 'Cannot continue event group if there are no events.'
+            group_id = self._last_event.group_id
+
+        return self._create_event(
+            event_type=event_type,
+            event_args=event_args,
+            group_id=group_id,
+        )
+
+    def _create_event(
+        self,
+        event_type: HathorEvents,
+        event_args: EventArguments,
+        group_id: Optional[int],
+    ) -> BaseEvent:
+        return BaseEvent(
+            id=0 if self._last_event is None else self._last_event.id + 1,
             peer_id=self._peer_id,
-            timestamp=self.clock.seconds(),
-            type=event.value,
-            data=build_event_data(event, event_args),
+            timestamp=self._clock.seconds(),
+            type=event_type.value,
+            data=_build_event_data(event_type, event_args),
             group_id=group_id,
         )
-        self.event_storage.save_event(event_to_store)
-        self._next_event_id += 1
-        if event in _GROUP_START_EVENTS:
-            self._current_group_id = self._next_group_id
-            self._next_group_id += 1
-        if event in _GROUP_END_EVENTS:
-            self._current_group_id = None

hathor/event/storage/event_storage.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from abc import ABC, abstractmethod
-from typing import Optional
+from typing import Iterator, Optional
 
 from hathor.event.base_event import BaseEvent
 
@@ -38,3 +38,8 @@ def get_last_event(self) -> Optional[BaseEvent]:
     def get_last_group_id(self) -> Optional[int]:
         """ Get the last group-id that was emitted, this is used to help resume when restarting."""
         raise NotImplementedError
+
+    @abstractmethod
+    def iter_from_event(self, key: int) -> Iterator[BaseEvent]:
+        """ Iterate through events starting from the event with the given key"""
+        raise NotImplementedError

hathor/event/storage/memory_storage.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import List, Optional
+from typing import Iterator, List, Optional
 
 from hathor.event.base_event import BaseEvent
 from hathor.event.storage.event_storage import EventStorage
@@ -25,8 +25,6 @@ def __init__(self):
         self._last_group_id: Optional[int] = None
 
     def save_event(self, event: BaseEvent) -> None:
-        if event.id < 0:
-            raise ValueError('event.id must be non-negative')
         if event.id != len(self._events):
             raise ValueError('invalid event.id, ids must be sequential and leave no gaps')
         self._last_event = event
@@ -36,7 +34,7 @@ def save_event(self, event: BaseEvent) -> None:
 
     def get_event(self, key: int) -> Optional[BaseEvent]:
         if key < 0:
-            raise ValueError('key must be non-negative')
+            raise ValueError(f'event.id \'{key}\' must be non-negative')
         if key >= len(self._events):
             return None
         event = self._events[key]
@@ -48,3 +46,11 @@ def get_last_event(self) -> Optional[BaseEvent]:
 
     def get_last_group_id(self) -> Optional[int]:
         return self._last_group_id
+
+    def iter_from_event(self, key: int) -> Iterator[BaseEvent]:
+        if key < 0:
+            raise ValueError(f'event.id \'{key}\' must be non-negative')
+
+        while key < len(self._events):
+            yield self._events[key]
+            key += 1