Remove dramatiq and pubsub

Makefile

@@ -66,7 +66,7 @@ double-env:
 
 # run all e2e tests
 e2e-test:
-	./scripts/wait_for_server_ready.sh 12 # 12 services should have log, 2 gateways has no log
+	./scripts/wait_for_server_ready.sh 6 # 12 services should have log, 2 gateways has no log
 	cat double.vars
 	source double.vars && PIPENV_PIPFILE=backend/Pipfile pipenv run pytest \
 		backend/tests/e2e_tests -k "$(T)" -W ignore::DeprecationWarning

backend/Pipfile

@@ -7,7 +7,6 @@ url = "https://pypi.org/simple"
 verify_ssl = true
 
 [dev-packages]
-fakeredis = "*"
 black = "*"
 pyre-check = "*"
 pyre-extensions = "*"
@@ -17,9 +16,6 @@ ipython = "*"
 
 [packages]
 diem-sample-wallet = {editable = true,path = "."}
-dramatiq = {extras = ["redis", "watch"],version = "*"}
-flask-redis = "*"
-redis = "*"
 Flask = ">=1.1.1"
 Flask-WTF = "*"
 WTForms = "*"

backend/README.md

@@ -28,7 +28,7 @@ Then run the following to build the images from source and stand up a local depl
 
 The default gateway port is 8080, so you can visit http://localhost:8080. Develop mode also exposes the other ports from the backend services onto the host. The main backend webserver is on 5000. Swagger API docs can be found at http://localhost:5000/apidocs.
 
-Getting started in the code, we've provided detailed logs through each of the [workflows](wallet/background_tasks), which can be found at http://localhost:5000/execution_logs.
+Getting started in the code, we've provided detailed logs through each of the [workflows](wallet), which can be found at http://localhost:5000/execution_logs.
 
 ### Docker Compose
 
@@ -55,15 +55,6 @@ If you want to go the extra mile, you can also use helm to install the Diem Refe
 
 The Python backend relies on pipenv. To set up, run `pipenv install --dev` in /backend directory.
 
-The workflows use [dramatiq](https://dramatiq.io/) with the Redis broker. Install Redis and run it as a daemon:
-
-    $ redis-server --daemonize yes
-
-After ensuring you have Redis running:
-
-    $ ./run_web.sh  # runs main webserver
-    $ ./run_worker.sh  # runs dramatiq workers
-
 To test:
 
     $ ./format.sh  # runs black

backend/pubsub/__main__.py

@@ -1,34 +0,0 @@
-# Copyright (c) The Diem Core Contributors
-# SPDX-License-Identifier: Apache-2.0
-
-import argparse
-import json
-import sys
-
-from pubsub import DEFL_CONFIG, VASP_ADDR
-from pubsub.client import LRWPubSubClient
-
-
-parser = argparse.ArgumentParser(
-    description="Pubsub CLI tool. Takes in pubsub config file or VASP_ADDR environment variable"
-)
-parser.add_argument("-f", "--file", type=str, help="LRW pubsub config file path")
-args = parser.parse_args()
-
-if args.file:
-    try:
-        with open(args.file) as f:
-            conf = json.load(f)
-    except:
-        print("Missing or invalid config file, exiting...")
-        sys.exit(1)
-else:  # load in by env var
-    if not VASP_ADDR:
-        print("Missing VASP_ADDR environment variable, exiting...")
-        sys.exit(1)
-    conf = DEFL_CONFIG
-
-print(conf)
-
-client = LRWPubSubClient(conf)
-client.start()

backend/pubsub/client.py

@@ -9,9 +9,11 @@
 import logging
 import json
 
-from wallet.background_tasks.background import process_incoming_txn
 from .types import LRWPubSubEvent
+from wallet.services.transaction import process_incoming_transaction
+from diem_utils.types.currencies import DiemCurrency
 from diem import jsonrpc
+from . import DEFL_CONFIG
 
 
 class FileProgressStorage:
@@ -31,14 +33,14 @@ def save_state(self, state: Dict[str, int]) -> None:
 
 
 class LRWPubSubClient:
-    def __init__(self, config: Dict[str, Any]) -> None:
+    def __init__(self, config: Dict[str, Any] = DEFL_CONFIG) -> None:
         self.sync_interval_ms = config["sync_interval_ms"]
         self.accounts = config["accounts"]
 
         self.diem_node_uri = config["diem_node_uri"]
         self.progress_file_path = config["progress_file_path"]
         self.fetch_batch_size = 10
-        self.processor = config.get("processor", process_incoming_txn)
+        self.processor = config.get("processor", LRWPubSubClient.process_incoming_txn)
 
         logging.info(f"Loaded LRWPubSubClient with config: {config}")
 
@@ -63,7 +65,7 @@ def sync(
                 )
                 for event in events:
                     lrw_event = LRWPubSubEvent.from_jsonrpc_event(event)
-                    self.processor.send(lrw_event)
+                    self.processor(lrw_event)
                     logging.info(f"SUCCESS: sent to wallet onchain {lrw_event}")
 
                 after_sync_state[key] = sequence_num + len(events)
@@ -87,3 +89,23 @@ def init_progress_state(self) -> Dict[str, int]:
             if account.received_events_key not in state:
                 state[account.received_events_key] = 0
         return state
+
+    @staticmethod
+    def process_incoming_txn(txn: LRWPubSubEvent) -> None:
+        metadata = txn.metadata
+        blockchain_version = txn.version
+        sender_address = txn.sender
+        receiver_address = txn.receiver
+        sequence = txn.sequence
+        amount = txn.amount
+        currency = DiemCurrency[txn.currency]
+        process_incoming_transaction(
+            blockchain_version=blockchain_version,
+            sender_address=sender_address,
+            receiver_address=receiver_address,
+            sequence=sequence,
+            amount=amount,
+            currency=currency,
+            metadata=metadata,
+        )
+

backend/test.sh

@@ -11,7 +11,7 @@ set -euo pipefail
 
 # NOTE:
 # each of these test modules need to be run independently due to how they
-# instantiate dramatiq and other test dependencies in conftest.py
+# instantiate test dependencies in conftest.py
 pipenv run python3 $DIR/setup.py pytest --addopts="$DIR/tests/wallet_tests $@"
 pipenv run python3 $DIR/setup.py pytest --addopts="$DIR/tests/webapp_tests $@"
 pipenv run python3 $DIR/setup.py pytest --addopts="$DIR/tests/pubsub_tests $@"

backend/tests/conftest.py

@@ -89,14 +89,6 @@ def wait_for_transaction(*args):
     yield network
 
 
-@pytest.fixture(autouse=True)
-def no_background_tasks(monkeypatch) -> None:
-    def mocked() -> bool:
-        return False
-
-    monkeypatch.setattr(services, "run_bg_tasks", mocked)
-
-
 class LpClientMock:
     QUOTES: Dict[QuoteId, QuoteData] = {}
     TRADES: Dict[TradeId, TradeData] = {}

backend/tests/wallet_tests/services/test_transaction.py

@@ -53,7 +53,7 @@ def test_send_transaction() -> None:
     assert tx.destination_subaddress == "receiver_subaddress"
 
 
-def test_transaction_direction(no_background_tasks) -> None:
+def test_transaction_direction() -> None:
     account_id, send_tx = send_fake_tx()
     assert get_transaction_direction(account_id, send_tx) == TransactionDirection.SENT
 

backend/wallet/__init__.py

@@ -1,9 +1,4 @@
 # Copyright (c) The Diem Core Contributors
 # SPDX-License-Identifier: Apache-2.0
 
-# keep config import first, because we MUST configure dramatiq broker prior to background_tasks import
-# once @actor is being used (e.g. in background_tasks) and no global dramatiq broker configured -
-# dramatiq will try to use default RabbitMQ one (this is hardcoded and can't be changed) - which will fail
-
 from . import config
-from . import background_tasks