Merge pull request #99 from confio/implement_watchAccount

Implement CosmWasmConnection.watchAccount

packages/bcp/src/cosmwasmconnection.spec.ts

@@ -1,7 +1,9 @@
 import { CosmosAddressBech32Prefix, decodeSignature } from "@cosmwasm/sdk";
 import {
+  Account,
   Address,
   Algorithm,
+  Amount,
   ChainId,
   isBlockInfoPending,
   isBlockInfoSucceeded,
@@ -17,6 +19,7 @@ import { Random, Secp256k1, Secp256k1Signature, Sha256 } from "@iov/crypto";
 import { Bech32, Encoding } from "@iov/encoding";
 import { HdPaths, Secp256k1HdWallet, UserProfile } from "@iov/keycontrol";
 import { assert } from "@iov/utils";
+import BN from "bn.js";
 
 import { CosmWasmConnection, TokenConfiguration } from "./cosmwasmconnection";
 import { encodeFullSignature } from "./encode";
@@ -54,6 +57,11 @@ describe("CosmWasmConnection", () => {
   const defaultChainId = "cosmos:testing" as ChainId;
   const defaultEmptyAddress = "cosmos1h806c7khnvmjlywdrkdgk2vrayy2mmvf9rxk2r" as Address;
   const defaultRecipient = "cosmos1t70qnpr0az8tf7py83m4ue5y89w58lkjmx0yq2" as Address;
+  const defaultAmount: Amount = {
+    quantity: "7744887",
+    fractionalDigits: 6,
+    tokenTicker: cosm,
+  };
 
   const unusedAccount = {
     pubkey: {
@@ -267,6 +275,70 @@ describe("CosmWasmConnection", () => {
     });
   });
 
+  describe("watchAccount", () => {
+    it("can watch account by address", done => {
+      pendingWithoutWasmd();
+      const recipient = makeRandomAddress();
+      const events = new Array<Account | undefined>();
+
+      (async () => {
+        const connection = await CosmWasmConnection.establish(httpUrl, defaultPrefix, defaultConfig);
+        const subscription = connection.watchAccount({ address: recipient }).subscribe({
+          next: event => {
+            events.push(event);
+
+            if (events.length === 3) {
+              const [event1, event2, event3] = events;
+
+              expect(event1).toBeUndefined();
+
+              assert(event2, "Second event must not be undefined");
+              expect(event2.address).toEqual(recipient);
+              expect(event2.pubkey).toBeUndefined();
+              expect(event2.balance.length).toEqual(1);
+              expect(event2.balance[0].quantity).toEqual(defaultAmount.quantity);
+              expect(event2.balance[0].tokenTicker).toEqual(defaultAmount.tokenTicker);
+
+              assert(event3, "Third event must not be undefined");
+              expect(event3.address).toEqual(recipient);
+              expect(event3.pubkey).toBeUndefined();
+              expect(event3.balance.length).toEqual(1);
+              expect(event3.balance[0].quantity).toEqual(new BN(defaultAmount.quantity).imuln(2).toString());
+              expect(event3.balance[0].tokenTicker).toEqual(defaultAmount.tokenTicker);
+
+              subscription.unsubscribe();
+              connection.disconnect();
+              done();
+            }
+          },
+          complete: done.fail,
+          error: done.fail,
+        });
+
+        const profile = new UserProfile();
+        const wallet = profile.addWallet(Secp256k1HdWallet.fromMnemonic(faucet.mnemonic));
+        const sender = await profile.createIdentity(wallet.id, defaultChainId, faucet.path);
+        const senderAddress = connection.codec.identityToAddress(sender);
+
+        for (const i of [0, 1]) {
+          const sendTx = await connection.withDefaultFee<SendTransaction>({
+            kind: "bcp/send",
+            chainId: defaultChainId,
+            senderPubkey: sender.pubkey,
+            sender: senderAddress,
+            recipient: recipient,
+            amount: defaultAmount,
+            memo: `Trigger for new event ${i}`,
+          });
+          const nonce = await connection.getNonce({ address: senderAddress });
+          const signedTransaction = await profile.signTransaction(sender, sendTx, connection.codec, nonce);
+          const result = await connection.postTx(connection.codec.bytesToPost(signedTransaction));
+          await result.blockInfo.waitFor(info => !isBlockInfoPending(info));
+        }
+      })().catch(done.fail);
+    });
+  });
+
   describe("getTx", () => {
     it("can get a recently posted bank send transaction", async () => {
       pendingWithoutWasmd();

packages/bcp/src/cosmwasmconnection.ts

@@ -41,7 +41,7 @@ import { DefaultValueProducer, ValueAndUpdates } from "@iov/stream";
 import BN from "bn.js";
 import equal from "fast-deep-equal";
 import { ReadonlyDate } from "readonly-date";
-import { Stream } from "xstream";
+import { Producer, Stream } from "xstream";
 
 import { decodeCosmosPubkey, pubkeyToAddress } from "./address";
 import { Caip5 } from "./caip5";
@@ -66,6 +66,9 @@ function isDefined<X>(value: X | undefined): value is X {
   return value !== undefined;
 }
 
+/** Account and undefined are valid events. The third option means no event fired yet */
+type LastWatchAccountEvent = Account | undefined | "no_event_fired_yet";
+
 export class CosmWasmConnection implements BlockchainConnection {
   // we must know prefix and tokens a priori to understand the chain
   public static async establish(
@@ -185,8 +188,34 @@ export class CosmWasmConnection implements BlockchainConnection {
     }
   }
 
-  public watchAccount(_account: AccountQuery): Stream<Account | undefined> {
-    throw new Error("not implemented");
+  public watchAccount(query: AccountQuery): Stream<Account | undefined> {
+    let lastEvent: LastWatchAccountEvent = "no_event_fired_yet";
+    let pollInternal: NodeJS.Timeout | undefined;
+    const producer: Producer<Account | undefined> = {
+      start: async listener => {
+        const poll = async (): Promise<void> => {
+          try {
+            const event = await this.getAccount(query);
+            if (!equal(event, lastEvent)) {
+              listener.next(event);
+              lastEvent = event;
+            }
+          } catch (error) {
+            listener.error(error);
+          }
+        };
+
+        pollInternal = setInterval(poll, defaultPollInterval);
+        await poll();
+      },
+      stop: () => {
+        if (pollInternal) {
+          clearInterval(pollInternal);
+          pollInternal = undefined;
+        }
+      },
+    };
+    return Stream.create(producer);
   }
 
   public async getNonce(query: AddressQuery | PubkeyQuery): Promise<Nonce> {

packages/bcp/types/cosmwasmconnection.d.ts

@@ -64,7 +64,7 @@ export declare class CosmWasmConnection implements BlockchainConnection {
    */
   identifier(signed: SignedTransaction): Promise<TransactionId>;
   getAccount(query: AccountQuery): Promise<Account | undefined>;
-  watchAccount(_account: AccountQuery): Stream<Account | undefined>;
+  watchAccount(query: AccountQuery): Stream<Account | undefined>;
   getNonce(query: AddressQuery | PubkeyQuery): Promise<Nonce>;
   getNonces(query: AddressQuery | PubkeyQuery, count: number): Promise<readonly Nonce[]>;
   getBlockHeader(height: number): Promise<BlockHeader>;