The transport layer should not know about events

raiden/network/transport/matrix.py

@@ -22,7 +22,6 @@
 from matrix_client.errors import MatrixError, MatrixRequestError
 from matrix_client.user import User
 
-from raiden import messages
 from raiden.constants import ID_TO_NETWORKNAME
 from raiden.exceptions import (
     InvalidAddress,
@@ -46,8 +45,7 @@
 from raiden.network.utils import get_http_rtt
 from raiden.raiden_service import RaidenService
 from raiden.storage.serialize import JSONSerializer
-from raiden.transfer import events as transfer_events, views
-from raiden.transfer.mediated_transfer import events as mediated_transfer_events
+from raiden.transfer import views
 from raiden.transfer.queue_identifier import QueueIdentifier
 from raiden.transfer.state import (
     NODE_NETWORK_REACHABLE,
@@ -202,7 +200,7 @@ def start(
         self.greenlets = [self._client.sync_thread]
 
         self._client.set_presence_state(UserPresence.ONLINE.value)
-        self._send_queued_messages()  # uses property instead of initial_queues
+        self._send_queued_messages(initial_queues)
 
         self.log.info('TRANSPORT STARTED', config=self._config)
 
@@ -674,12 +672,9 @@ def send_delivered_for(message: SignedMessage):
             self.log.warning('Exception while processing message', exc_info=True)
             return
 
-    def _send_queued_messages(self):
-        for queue_identifier, events in self._queueids_to_queues.items():
-            node_address = self._raiden_service.address
-            for event in events:
-                message = _event_to_message(event, node_address)
-                self._raiden_service.sign(message)
+    def _send_queued_messages(self, queueids_to_queues):
+        for queue_identifier, messages in queueids_to_queues.items():
+            for message in messages:
                 self.start_health_check(queue_identifier.recipient)
                 self.send_async(queue_identifier, message)
 
@@ -1069,19 +1064,3 @@ def _get_room_id_for_address(self, address: Address) -> Optional[str]:
             self._set_room_id_for_address(address, None)
             return None
         return room_id
-
-
-def _event_to_message(event, node_address):
-    eventtypes_to_messagetype = {
-        mediated_transfer_events.SendBalanceProof: messages.Secret,
-        mediated_transfer_events.SendLockedTransfer: messages.LockedTransfer,
-        mediated_transfer_events.SendRefundTransfer: messages.RefundTransfer,
-        mediated_transfer_events.SendRevealSecret: messages.RevealSecret,
-        mediated_transfer_events.SendSecretRequest: messages.SecretRequest,
-        transfer_events.SendDirectTransfer: messages.DirectTransfer,
-        transfer_events.SendProcessed: messages.Processed,
-    }
-    message_class = eventtypes_to_messagetype.get(type(event))
-    if message_class is None:
-        raise TypeError(f'Event type {type(event)} is not supported.')
-    return message_class.from_event(event)

raiden/network/transport/udp/udp_transport.py

@@ -14,8 +14,9 @@
     RaidenShuttingDown,
     UnknownAddress,
 )
+
 from raiden.message_handler import on_message
-from raiden.messages import Delivered, Message, Ping, Pong, decode, message_from_sendevent
+from raiden.messages import Delivered, Message, Ping, Pong, decode
 from raiden.network.transport.udp import healthcheck
 from raiden.network.transport.udp.udp_utils import (
     event_first_of,
@@ -207,7 +208,7 @@ def __init__(self, discovery, udpsocket, throttle_policy, config):
     def start(
             self,
             raiden: RaidenService,
-            queueids_to_queues: typing.Dict[QueueIdentifier, typing.List[Event]],
+            queueids_to_queues: typing.Dict[QueueIdentifier, typing.List[Message]],
     ):
         if not self.event_stop.ready():
             raise RuntimeError('UDPTransport started while running')
@@ -221,15 +222,10 @@ def start(
         self.server.set_handle(self._receive)
 
         for queue_identifier, queue in queueids_to_queues.items():
-            encoded_queue = list()
-
-            for sendevent in queue:
-                message = message_from_sendevent(sendevent, raiden.address)
-                raiden.sign(message)
-                encoded = message.encode()
-
-                encoded_queue.append((encoded, sendevent.message_identifier))
-
+            encoded_queue = [
+                (message.encode(), message.message_identifier)
+                for message in queue
+            ]
             self.init_queue_for(queue_identifier, encoded_queue)
 
         self.server.start()

raiden/raiden_service.py

@@ -16,7 +16,8 @@
 from raiden.connection_manager import ConnectionManager
 from raiden.constants import SNAPSHOT_STATE_CHANGES_COUNT
 from raiden.exceptions import InvalidAddress, RaidenRecoverableError, RaidenShuttingDown
-from raiden.messages import LockedTransfer, SignedMessage
+from raiden.messages import LockedTransfer, SignedMessage, message_from_sendevent
+
 from raiden.network.blockchain_service import BlockChainService
 from raiden.network.proxies import SecretRegistry, TokenNetworkRegistry
 from raiden.storage import serialize, sqlite, wal
@@ -314,12 +315,24 @@ def start(self):
 
         self.alarm.start()
 
-        queueids_to_queues = views.get_all_messagequeues(chain_state)
-        # repopulate identifier_to_results for pending transfers
-        for queue_messages in queueids_to_queues.values():
-            for message in queue_messages:
-                if isinstance(message, SendDirectTransfer):
-                    self.identifier_to_results[message.payment_identifier] = AsyncResult()
+        events_queues = views.get_all_messagequeues(chain_state)
+        queueids_to_queues = dict()
+
+        for queue_identifier, event_queue in events_queues.items():
+
+            # repopulate identifier_to_results for pending transfers
+            for event in event_queue:
+                if type(event) == SendDirectTransfer:
+                    self.identifier_to_results[event.payment_identifier] = AsyncResult()
+
+            message_queue = []
+            for event in event_queue:
+                message = message_from_sendevent(event, self.address)
+                self.sign(message)
+                message_queue.append(message)
+
+            queueids_to_queues[queue_identifier] = message_queue
+
         self.transport.start(self, queueids_to_queues)
 
         # exceptions on these subtasks should crash the app and bubble up