Added integration tests

e2e/infrastructure/BlockHttp.spec.ts

@@ -15,7 +15,7 @@
  */
 
 import { expect } from 'chai';
-import { mergeMap } from 'rxjs/operators';
+import { mergeMap, toArray } from 'rxjs/operators';
 import { BlockRepository } from '../../src/infrastructure/BlockRepository';
 import { ReceiptRepository } from '../../src/infrastructure/ReceiptRepository';
 import { Account } from '../../src/model/account/Account';
@@ -26,6 +26,9 @@ import { TransferTransaction } from '../../src/model/transaction/TransferTransac
 import { UInt64 } from '../../src/model/UInt64';
 import { IntegrationTestHelper } from './IntegrationTestHelper';
 import { BlockSearchCriteria } from '../../src/infrastructure/searchCriteria/BlockSearchCriteria';
+import { BlockPaginationStreamer } from '../../src/infrastructure/paginationStreamer/BlockPaginationStreamer';
+import { deepEqual } from 'assert';
+import { take } from 'rxjs/operators';
 
 describe('BlockHttp', () => {
     const helper = new IntegrationTestHelper();
@@ -96,10 +99,21 @@ describe('BlockHttp', () => {
 
     describe('searchBlock', () => {
         it('should return block info given height and limit', async () => {
-            const blocksInfo = await blockRepository.search({} as BlockSearchCriteria).toPromise();
+            const blocksInfo = await blockRepository.search({}).toPromise();
             expect(blocksInfo.getData().length).to.be.greaterThan(0);
         });
     });
+
+    describe('searchBlock with streamer', () => {
+        it('should return block info given height and limit', async () => {
+            const streamer = new BlockPaginationStreamer(blockRepository);
+            const blockInfoStreamer = await streamer.search({ pageSize: 20 }).pipe(take(20), toArray()).toPromise();
+            const blocksInfo = await blockRepository.search({ pageSize: 20 }).toPromise();
+            expect(blockInfoStreamer.length).to.be.greaterThan(0);
+            deepEqual(blockInfoStreamer, blocksInfo.getData());
+        });
+    });
+
     describe('getMerkleReceipts', () => {
         it('should return Merkle Receipts', async () => {
             const merkleReceipts = await receiptRepository

e2e/infrastructure/MosaicHttp.spec.ts

@@ -29,6 +29,9 @@ import { MosaicDefinitionTransaction } from '../../src/model/transaction/MosaicD
 import { NamespaceRegistrationTransaction } from '../../src/model/transaction/NamespaceRegistrationTransaction';
 import { UInt64 } from '../../src/model/UInt64';
 import { IntegrationTestHelper } from './IntegrationTestHelper';
+import { MosaicPaginationStreamer } from '../../src/infrastructure/paginationStreamer/MosaicPaginationStreamer';
+import { toArray, take } from 'rxjs/operators';
+import { deepEqual } from 'assert';
 
 describe('MosaicHttp', () => {
     let mosaicId: MosaicId;
@@ -166,6 +169,20 @@ describe('MosaicHttp', () => {
         });
     });
 
+    describe('searchMosaics with streamer', () => {
+        it('should call searchMosaics successfully', async () => {
+            const streamer = new MosaicPaginationStreamer(mosaicRepository);
+            const mosaicsStreamer = await streamer
+                .search({ ownerAddress: account.address, pageSize: 3 })
+                .pipe(take(3), toArray())
+                .toPromise();
+            const mosaics = await mosaicRepository.search({ ownerAddress: account.address, pageSize: 3 }).toPromise();
+            expect(mosaicsStreamer.length).to.be.greaterThan(0);
+            expect(mosaicsStreamer.find((m) => m.id.toHex() === mosaicId.toHex()) !== undefined).to.be.true;
+            deepEqual(mosaics.getData(), mosaicsStreamer);
+        });
+    });
+
     /**
      * =========================
      * House Keeping

e2e/infrastructure/TransactionHttp.spec.ts

@@ -69,6 +69,9 @@ import { VrfKeyLinkTransaction } from '../../src/model/transaction/VrfKeyLinkTra
 import { VotingKeyLinkTransaction } from '../../src/model/transaction/VotingKeyLinkTransaction';
 import { NodeKeyLinkTransaction } from '../../src/model/transaction/NodeKeyLinkTransaction';
 import { AddressRestrictionFlag, MosaicRestrictionFlag, OperationRestrictionFlag } from '../../src/model/model';
+import { TransactionPaginationStreamer } from '../../src/infrastructure/paginationStreamer/TransactionPaginationStreamer';
+import { toArray, take } from 'rxjs/operators';
+import { deepEqual } from 'assert';
 
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const CryptoJS = require('crypto-js');
@@ -1466,4 +1469,16 @@ describe('TransactionHttp', () => {
             expect(transactions.getData().length).to.be.greaterThan(0);
         });
     });
+
+    describe('searchTransactions using steamer', () => {
+        it('should return transaction info given address', async () => {
+            const streamer = new TransactionPaginationStreamer(transactionRepository);
+            const transactionsNoStreamer = await transactionRepository
+                .search({ address: account.address, pageSize: 3 } as TransactionSearchCriteria)
+                .toPromise();
+            const transactions = await streamer.search({ address: account.address, pageSize: 3 }).pipe(take(3), toArray()).toPromise();
+            expect(transactions.length).to.be.greaterThan(0);
+            deepEqual(transactionsNoStreamer.getData(), transactions);
+        });
+    });
 });

src/infrastructure/TransactionRepository.ts

@@ -84,12 +84,4 @@ export interface TransactionRepository extends Searcher<Transaction, Transaction
      * @returns Observable<TransactionAnnounceResponse>
      */
     announceAggregateBondedCosignature(cosignatureSignedTransaction: CosignatureSignedTransaction): Observable<TransactionAnnounceResponse>;
-
-    // /**
-    //  * Search transactions based on the criteria's filtering and sorting returning a page.
-    //  *
-    //  * @param criteria the criteria
-    //  * @return a page of transactions.
-    //  */
-    // searchTransactions(criteria: TransactionSearchCriteria): Observable<Page<Transaction>>;
 }

src/infrastructure/transaction/CreateTransactionFromDTO.ts

@@ -122,6 +122,23 @@ const extractMessage = (message: any): PlainMessage | EncryptedMessage => {
     return msgObj;
 };
 
+/**
+ * @internal
+ * Extract transaction meta data
+ *
+ * @param meta - Transaction meta data
+ * @param id - TransactionId
+ * @return {TransactionInfo | AggregateTransactionInfo | undefined}
+ */
+const extractTransactionMeta = (meta: any, id: string): TransactionInfo | AggregateTransactionInfo | undefined => {
+    if (!meta) {
+        return undefined;
+    }
+    if (meta.aggregateHash || meta.aggregateId) {
+        return new AggregateTransactionInfo(UInt64.fromNumericString(meta.height), meta.index, id, meta.aggregateHash, meta.aggregateId);
+    }
+    return new TransactionInfo(UInt64.fromNumericString(meta.height), meta.index, id, meta.hash, meta.merkleComponentHash);
+};
 /**
  * @internal
  * @param transactionDTO
@@ -502,15 +519,7 @@ export const CreateTransactionFromDTO = (transactionDTO): Transaction => {
     ) {
         const innerTransactions = transactionDTO.transaction.transactions
             ? transactionDTO.transaction.transactions.map((innerTransactionDTO) => {
-                  const aggregateTransactionInfo = innerTransactionDTO.meta
-                      ? new AggregateTransactionInfo(
-                            UInt64.fromNumericString(innerTransactionDTO.meta.height),
-                            innerTransactionDTO.meta.index,
-                            innerTransactionDTO.id,
-                            innerTransactionDTO.meta.aggregateHash,
-                            innerTransactionDTO.meta.aggregateId,
-                        )
-                      : undefined;
+                  const aggregateTransactionInfo = extractTransactionMeta(innerTransactionDTO.meta, innerTransactionDTO.id);
                   innerTransactionDTO.transaction.maxFee = transactionDTO.transaction.maxFee;
                   innerTransactionDTO.transaction.deadline = transactionDTO.transaction.deadline;
                   innerTransactionDTO.transaction.signature = transactionDTO.transaction.signature;
@@ -536,26 +545,12 @@ export const CreateTransactionFromDTO = (transactionDTO): Transaction => {
             transactionDTO.transaction.signerPublicKey
                 ? PublicAccount.createFromPublicKey(transactionDTO.transaction.signerPublicKey, transactionDTO.transaction.network)
                 : undefined,
-            transactionDTO.meta
-                ? new TransactionInfo(
-                      UInt64.fromNumericString(transactionDTO.meta.height),
-                      transactionDTO.meta.index,
-                      transactionDTO.id,
-                      transactionDTO.meta.hash,
-                      transactionDTO.meta.merkleComponentHash,
-                  )
-                : undefined,
+            extractTransactionMeta(transactionDTO.meta, transactionDTO.id),
         );
     } else {
-        const transactionInfo = transactionDTO.meta
-            ? new TransactionInfo(
-                  UInt64.fromNumericString(transactionDTO.meta.height),
-                  transactionDTO.meta.index,
-                  transactionDTO.id,
-                  transactionDTO.meta.hash,
-                  transactionDTO.meta.merkleComponentHash,
-              )
-            : undefined;
-        return CreateStandaloneTransactionFromDTO(transactionDTO.transaction, transactionInfo);
+        return CreateStandaloneTransactionFromDTO(
+            transactionDTO.transaction,
+            extractTransactionMeta(transactionDTO.meta, transactionDTO.id),
+        );
     }
 };