bridge.scanAccountsOnDevice needs to return Observable for safety

src/bridge/EthereumJSBridge.js

@@ -156,117 +156,118 @@ const fetchCurrentBlock = (perCurrencyId => currency => {
 })({})
 
 const EthereumBridge: WalletBridge<Transaction> = {
-  scanAccountsOnDevice(currency, deviceId, { next, complete, error }) {
-    let finished = false
-    const unsubscribe = () => {
-      finished = true
-    }
-    const api = apiForCurrency(currency)
-
-    // in future ideally what we want is:
-    // return mergeMap(addressesObservable, address => fetchAccount(address))
-
-    let newAccountCount = 0
-
-    async function stepAddress(
-      index,
-      { address, path: freshAddressPath, publicKey },
-      isStandard,
-    ): { account?: Account, complete?: boolean } {
-      const balance = await api.getAccountBalance(address)
-      if (finished) return { complete: true }
-      const currentBlock = await fetchCurrentBlock(currency)
-      if (finished) return { complete: true }
-      let { txs } = await api.getTransactions(address)
-      if (finished) return { complete: true }
-
-      const freshAddress = address
-      const accountId = `ethereumjs:${currency.id}:${address}:${publicKey}`
-
-      if (txs.length === 0) {
-        // this is an empty account
-        if (isStandard) {
-          if (newAccountCount === 0) {
-            // first zero account will emit one account as opportunity to create a new account..
-            const account: $Exact<Account> = {
-              id: accountId,
-              xpub: '',
-              freshAddress,
-              freshAddressPath,
-              name: getNewAccountPlaceholderName(currency, index),
-              balance,
-              blockHeight: currentBlock.height,
-              index,
-              currency,
-              operations: [],
-              pendingOperations: [],
-              unit: currency.units[0],
-              lastSyncDate: new Date(),
+  scanAccountsOnDevice: (currency, deviceId) =>
+    Observable.create(o => {
+      let finished = false
+      const unsubscribe = () => {
+        finished = true
+      }
+      const api = apiForCurrency(currency)
+
+      // in future ideally what we want is:
+      // return mergeMap(addressesObservable, address => fetchAccount(address))
+
+      let newAccountCount = 0
+
+      async function stepAddress(
+        index,
+        { address, path: freshAddressPath, publicKey },
+        isStandard,
+      ): { account?: Account, complete?: boolean } {
+        const balance = await api.getAccountBalance(address)
+        if (finished) return { complete: true }
+        const currentBlock = await fetchCurrentBlock(currency)
+        if (finished) return { complete: true }
+        let { txs } = await api.getTransactions(address)
+        if (finished) return { complete: true }
+
+        const freshAddress = address
+        const accountId = `ethereumjs:${currency.id}:${address}:${publicKey}`
+
+        if (txs.length === 0) {
+          // this is an empty account
+          if (isStandard) {
+            if (newAccountCount === 0) {
+              // first zero account will emit one account as opportunity to create a new account..
+              const account: $Exact<Account> = {
+                id: accountId,
+                xpub: '',
+                freshAddress,
+                freshAddressPath,
+                name: getNewAccountPlaceholderName(currency, index),
+                balance,
+                blockHeight: currentBlock.height,
+                index,
+                currency,
+                operations: [],
+                pendingOperations: [],
+                unit: currency.units[0],
+                lastSyncDate: new Date(),
+              }
+              return { account, complete: true }
             }
-            return { account, complete: true }
+            newAccountCount++
           }
-          newAccountCount++
+          // NB for legacy addresses maybe we will continue at least for the first 10 addresses
+          return { complete: true }
         }
-        // NB for legacy addresses maybe we will continue at least for the first 10 addresses
-        return { complete: true }
-      }
 
-      const account: $Exact<Account> = {
-        id: accountId,
-        xpub: '',
-        freshAddress,
-        freshAddressPath,
-        name: getAccountPlaceholderName(currency, index, !isStandard),
-        balance,
-        blockHeight: currentBlock.height,
-        index,
-        currency,
-        operations: [],
-        pendingOperations: [],
-        unit: currency.units[0],
-        lastSyncDate: new Date(),
-      }
-      for (let i = 0; i < 50; i++) {
-        const api = apiForCurrency(account.currency)
-        const { block } = txs[txs.length - 1]
-        if (!block) break
-        const next = await api.getTransactions(account.freshAddress, block.hash)
-        if (next.txs.length === 0) break
-        txs = txs.concat(next.txs)
+        const account: $Exact<Account> = {
+          id: accountId,
+          xpub: '',
+          freshAddress,
+          freshAddressPath,
+          name: getAccountPlaceholderName(currency, index, !isStandard),
+          balance,
+          blockHeight: currentBlock.height,
+          index,
+          currency,
+          operations: [],
+          pendingOperations: [],
+          unit: currency.units[0],
+          lastSyncDate: new Date(),
+        }
+        for (let i = 0; i < 50; i++) {
+          const api = apiForCurrency(account.currency)
+          const { block } = txs[txs.length - 1]
+          if (!block) break
+          const next = await api.getTransactions(account.freshAddress, block.hash)
+          if (next.txs.length === 0) break
+          txs = txs.concat(next.txs)
+        }
+        txs.reverse()
+        account.operations = mergeOps([], flatMap(txs, txToOps(account)))
+        return { account }
       }
-      txs.reverse()
-      account.operations = mergeOps([], flatMap(txs, txToOps(account)))
-      return { account }
-    }
-
-    async function main() {
-      try {
-        const derivations = getDerivations(currency)
-        const last = derivations[derivations.length - 1]
-        for (const derivation of derivations) {
-          const isStandard = last === derivation
-          for (let index = 0; index < 255; index++) {
-            const freshAddressPath = derivation({ currency, x: index, segwit: false })
-            const res = await getAddressCommand
-              .send({ currencyId: currency.id, devicePath: deviceId, path: freshAddressPath })
-              .toPromise()
-            const r = await stepAddress(index, res, isStandard)
-            if (r.account) next(r.account)
-            if (r.complete) {
-              break
+
+      async function main() {
+        try {
+          const derivations = getDerivations(currency)
+          const last = derivations[derivations.length - 1]
+          for (const derivation of derivations) {
+            const isStandard = last === derivation
+            for (let index = 0; index < 255; index++) {
+              const freshAddressPath = derivation({ currency, x: index, segwit: false })
+              const res = await getAddressCommand
+                .send({ currencyId: currency.id, devicePath: deviceId, path: freshAddressPath })
+                .toPromise()
+              const r = await stepAddress(index, res, isStandard)
+              if (r.account) o.next(r.account)
+              if (r.complete) {
+                break
+              }
             }
           }
+          o.complete()
+        } catch (e) {
+          o.error(e)
         }
-        complete()
-      } catch (e) {
-        error(e)
       }
-    }
 
-    main()
+      main()
 
-    return { unsubscribe }
-  },
+      return unsubscribe
+    }),
 
   synchronize: ({ freshAddress, blockHeight, currency, operations }) =>
     Observable.create(o => {

src/bridge/LibcoreBridge.js

@@ -79,14 +79,13 @@ const getFees = async (a, transaction) => {
 }
 
 const LibcoreBridge: WalletBridge<Transaction> = {
-  scanAccountsOnDevice(currency, devicePath, observer) {
+  scanAccountsOnDevice(currency, devicePath) {
     return libcoreScanAccounts
       .send({
         devicePath,
         currencyId: currency.id,
       })
       .pipe(map(decodeAccount))
-      .subscribe(observer)
   },
 
   synchronize: account =>

src/bridge/RippleJSBridge.js

@@ -239,113 +239,114 @@ const getServerInfo = (map => endpointConfig => {
 })({})
 
 const RippleJSBridge: WalletBridge<Transaction> = {
-  scanAccountsOnDevice(currency, deviceId, { next, complete, error }) {
-    let finished = false
-    const unsubscribe = () => {
-      finished = true
-    }
+  scanAccountsOnDevice: (currency, deviceId) =>
+    Observable.create(o => {
+      let finished = false
+      const unsubscribe = () => {
+        finished = true
+      }
 
-    async function main() {
-      const api = apiForEndpointConfig()
-      try {
-        await api.connect()
-        const serverInfo = await getServerInfo()
-        const ledgers = serverInfo.completeLedgers.split('-')
-        const minLedgerVersion = Number(ledgers[0])
-        const maxLedgerVersion = Number(ledgers[1])
-
-        const derivations = getDerivations(currency)
-        for (const derivation of derivations) {
-          const legacy = derivation !== derivations[derivations.length - 1]
-          for (let index = 0; index < 255; index++) {
-            const freshAddressPath = derivation({ currency, x: index, segwit: false })
-            const { address, publicKey } = await await getAddress
-              .send({ currencyId: currency.id, devicePath: deviceId, path: freshAddressPath })
-              .toPromise()
-            if (finished) return
-
-            const accountId = `ripplejs:${currency.id}:${address}:${publicKey}`
-
-            let info
-            try {
-              info = await api.getAccountInfo(address)
-            } catch (e) {
-              if (e.message !== 'actNotFound') {
-                throw e
-              }
-            }
+      async function main() {
+        const api = apiForEndpointConfig()
+        try {
+          await api.connect()
+          const serverInfo = await getServerInfo()
+          const ledgers = serverInfo.completeLedgers.split('-')
+          const minLedgerVersion = Number(ledgers[0])
+          const maxLedgerVersion = Number(ledgers[1])
 
-            // fresh address is address. ripple never changes.
-            const freshAddress = address
-
-            if (!info) {
-              // account does not exist in Ripple server
-              // we are generating a new account locally
-              if (!legacy) {
-                next({
-                  id: accountId,
-                  xpub: '',
-                  name: getNewAccountPlaceholderName(currency, index),
-                  freshAddress,
-                  freshAddressPath,
-                  balance: 0,
-                  blockHeight: maxLedgerVersion,
-                  index,
-                  currency,
-                  operations: [],
-                  pendingOperations: [],
-                  unit: currency.units[0],
-                  archived: false,
-                  lastSyncDate: new Date(),
-                })
+          const derivations = getDerivations(currency)
+          for (const derivation of derivations) {
+            const legacy = derivation !== derivations[derivations.length - 1]
+            for (let index = 0; index < 255; index++) {
+              const freshAddressPath = derivation({ currency, x: index, segwit: false })
+              const { address, publicKey } = await await getAddress
+                .send({ currencyId: currency.id, devicePath: deviceId, path: freshAddressPath })
+                .toPromise()
+              if (finished) return
+
+              const accountId = `ripplejs:${currency.id}:${address}:${publicKey}`
+
+              let info
+              try {
+                info = await api.getAccountInfo(address)
+              } catch (e) {
+                if (e.message !== 'actNotFound') {
+                  throw e
+                }
               }
-              break
-            }
 
-            if (finished) return
-            const balance = parseAPIValue(info.xrpBalance)
-            invariant(
-              !isNaN(balance) && isFinite(balance),
-              `Ripple: invalid balance=${balance} for address ${address}`,
-            )
+              // fresh address is address. ripple never changes.
+              const freshAddress = address
+
+              if (!info) {
+                // account does not exist in Ripple server
+                // we are generating a new account locally
+                if (!legacy) {
+                  o.next({
+                    id: accountId,
+                    xpub: '',
+                    name: getNewAccountPlaceholderName(currency, index),
+                    freshAddress,
+                    freshAddressPath,
+                    balance: 0,
+                    blockHeight: maxLedgerVersion,
+                    index,
+                    currency,
+                    operations: [],
+                    pendingOperations: [],
+                    unit: currency.units[0],
+                    archived: false,
+                    lastSyncDate: new Date(),
+                  })
+                }
+                break
+              }
 
-            const transactions = await api.getTransactions(address, {
-              minLedgerVersion,
-              maxLedgerVersion,
-            })
-            if (finished) return
-
-            const account: $Exact<Account> = {
-              id: accountId,
-              xpub: '',
-              name: getAccountPlaceholderName(currency, index, legacy),
-              freshAddress,
-              freshAddressPath,
-              balance,
-              blockHeight: maxLedgerVersion,
-              index,
-              currency,
-              operations: [],
-              pendingOperations: [],
-              unit: currency.units[0],
-              lastSyncDate: new Date(),
+              if (finished) return
+              const balance = parseAPIValue(info.xrpBalance)
+              invariant(
+                !isNaN(balance) && isFinite(balance),
+                `Ripple: invalid balance=${balance} for address ${address}`,
+              )
+
+              const transactions = await api.getTransactions(address, {
+                minLedgerVersion,
+                maxLedgerVersion,
+              })
+              if (finished) return
+
+              const account: $Exact<Account> = {
+                id: accountId,
+                xpub: '',
+                name: getAccountPlaceholderName(currency, index, legacy),
+                freshAddress,
+                freshAddressPath,
+                balance,
+                blockHeight: maxLedgerVersion,
+                index,
+                currency,
+                operations: [],
+                pendingOperations: [],
+                unit: currency.units[0],
+                lastSyncDate: new Date(),
+              }
+              account.operations = transactions.map(txToOperation(account))
+              o.next(account)
             }
-            account.operations = transactions.map(txToOperation(account))
-            next(account)
           }
+          o.complete()
+        } catch (e) {
+          o.error(e)
+        } finally {
+          api.disconnect()
         }
-        complete()
-      } catch (e) {
-        error(e)
-      } finally {
-        api.disconnect()
       }
-    }
 
-    main()
+      main()
 
-    return { unsubscribe }
-  },
+      return unsubscribe
+    }),
 
   synchronize: ({ endpointConfig, freshAddress, blockHeight }) =>
     Observable.create(o => {

src/bridge/UnsupportedBridge.js

@@ -10,10 +10,10 @@ const UnsupportedBridge: WalletBridge<*> = {
       o.error(genericError)
     }),
 
-  scanAccountsOnDevice(currency, deviceId, { error }) {
-    Promise.resolve(genericError).then(error)
-    return { unsubscribe() {} }
-  },
+  scanAccountsOnDevice: () =>
+    Observable.create(o => {
+      o.error(genericError)
+    }),
 
   pullMoreOperations: () => Promise.reject(genericError),
 

src/bridge/makeMockBridge.js

@@ -75,36 +75,37 @@ function makeMockBridge(opts?: Opts): WalletBridge<*> {
         }
       }),
 
-    scanAccountsOnDevice(currency, deviceId, { next, complete, error }) {
-      let unsubscribed = false
-
-      async function job() {
-        if (Math.random() > scanAccountDeviceSuccessRate) {
-          await delay(1000)
-          if (!unsubscribed) error(new Error('scan failed'))
-          return
-        }
-        const nbAccountToGen = 3
-        for (let i = 0; i < nbAccountToGen && !unsubscribed; i++) {
-          await delay(500)
-          const account = genAccount(String(Math.random()), {
-            operationsSize: 0,
-            currency,
-          })
-          account.unit = currency.units[0]
-          if (!unsubscribed) next(account)
+    scanAccountsOnDevice: (currency, deviceId) =>
+      Observable.create(o => {
+        let unsubscribed = false
+
+        async function job() {
+          if (Math.random() > scanAccountDeviceSuccessRate) {
+            await delay(1000)
+            if (!unsubscribed) o.error(new Error('scan failed'))
+            return
+          }
+          const nbAccountToGen = 3
+          for (let i = 0; i < nbAccountToGen && !unsubscribed; i++) {
+            await delay(500)
+            const account = genAccount(String(Math.random()), {
+              operationsSize: 0,
+              currency,
+            })
+            account.unit = currency.units[0]
+            if (!unsubscribed) o.next(account)
+          }
+          if (!unsubscribed) o.complete()
         }
-        if (!unsubscribed) complete()
-      }
 
-      job()
+        job()
 
-      return {
-        unsubscribe() {
-          unsubscribed = true
-        },
-      }
-    },
+        return {
+          unsubscribe() {
+            unsubscribed = true
+          },
+        }
+      }),
 
     pullMoreOperations: async (_accountId, _desiredCount) => {
       await delay(1000)

src/bridge/types.js

@@ -33,11 +33,7 @@ export interface WalletBridge<Transaction> {
   // the scan can stop once all accounts are discovered.
   // the function returns a Subscription and you MUST stop everything if it is unsubscribed.
   // TODO return Observable
-  scanAccountsOnDevice(
-    currency: Currency,
-    deviceId: DeviceId,
-    observer: Observer<Account>,
-  ): Subscription;
+  scanAccountsOnDevice(currency: Currency, deviceId: DeviceId): Observable<Account>;
 
   // synchronize an account. meaning updating the account object with latest state.
   // function receives the initialAccount object so you can actually know what the user side currently have

src/components/modals/AddAccounts/steps/03-step-import.js

@@ -72,7 +72,7 @@ class StepImport extends PureComponent<StepProps> {
       // TODO: use the real device
       const devicePath = currentDevice.path
 
-      this.scanSubscription = bridge.scanAccountsOnDevice(currency, devicePath, {
+      this.scanSubscription = bridge.scanAccountsOnDevice(currency, devicePath).subscribe({
         next: account => {
           const { scannedAccounts, checkedAccountsIds, existingAccounts } = this.props
           const hasAlreadyBeenScanned = !!scannedAccounts.find(a => account.id === a.id)