Added #352 aggregate bonded announcing

Author: rg911

src/infrastructure/Listener.ts

@@ -239,15 +239,18 @@ export class Listener {
      * it emits a new Transaction in the event stream.
      *
      * @param address address we listen when a transaction is in confirmed state
+     * @param transactionHash transactionHash for filtering multiple transactions
      * @return an observable stream of Transaction with state confirmed
      */
-    public confirmed(address: Address): Observable<Transaction> {
+    public confirmed(address: Address, transactionHash?: string): Observable<Transaction> {
         this.subscribeTo(`confirmedAdded/${address.plain()}`);
         return this.messageSubject.asObservable().pipe(
             filter((_) => _.channelName === ListenerChannelName.confirmedAdded),
             filter((_) => _.message instanceof Transaction),
             map((_) => _.message as Transaction),
-            filter((_) => this.transactionFromAddress(_, address)));
+            filter((_) => this.transactionFromAddress(_, address)),
+            filter((_) => transactionHash === undefined || _.transactionInfo!.hash === transactionHash),
+        );
     }
 
     /**
@@ -289,15 +292,18 @@ export class Listener {
      * it emits a new {@link AggregateTransaction} in the event stream.
      *
      * @param address address we listen when a transaction with missing signatures state
+     * @param transactionHash transactionHash for filtering multiple transactions
      * @return an observable stream of AggregateTransaction with missing signatures state
      */
-    public aggregateBondedAdded(address: Address): Observable<AggregateTransaction> {
+    public aggregateBondedAdded(address: Address, transactionHash?: string): Observable<AggregateTransaction> {
         this.subscribeTo(`partialAdded/${address.plain()}`);
         return this.messageSubject.asObservable().pipe(
             filter((_) => _.channelName === ListenerChannelName.aggregateBondedAdded),
             filter((_) => _.message instanceof AggregateTransaction),
             map((_) => _.message as AggregateTransaction),
-            filter((_) => this.transactionFromAddress(_, address)));
+            filter((_) => this.transactionFromAddress(_, address)),
+            filter((_) => transactionHash === undefined || _.transactionInfo!.hash === transactionHash),
+        );
     }
 
     /**

src/model/transaction/SignedTransaction.ts

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+import { Address } from '../account/Address';
+import { PublicAccount } from '../account/PublicAccount';
 import {NetworkType} from '../blockchain/NetworkType';
 
 /**
@@ -65,4 +67,12 @@ export class SignedTransaction {
             networkType: this.networkType,
         };
     }
+
+    /**
+     * Return signer's address
+     * @returns {Address}
+     */
+    getSignerAddress(): Address {
+        return PublicAccount.createFromPublicKey(this.signerPublicKey, this.networkType).address;
+    }
 }

src/service/TransactionService.ts

@@ -16,13 +16,17 @@
 
 import {Observable} from 'rxjs';
 import { mergeMap, toArray} from 'rxjs/operators';
+import { flatMap } from 'rxjs/operators';
+import { Listener } from '../infrastructure/Listener';
 import { ReceiptHttp } from '../infrastructure/ReceiptHttp';
 import { TransactionHttp } from '../infrastructure/TransactionHttp';
 import { Address } from '../model/account/Address';
 import { MosaicId } from '../model/mosaic/MosaicId';
 import { NamespaceId } from '../model/namespace/NamespaceId';
 import { ResolutionType } from '../model/receipt/ResolutionType';
 import { Statement } from '../model/receipt/Statement';
+import { AggregateTransaction } from '../model/transaction/AggregateTransaction';
+import { SignedTransaction } from '../model/transaction/SignedTransaction';
 import { Transaction } from '../model/transaction/Transaction';
 import { ITransactionService } from './interfaces/ITransactionService';
 
@@ -33,13 +37,15 @@ export class TransactionService implements ITransactionService {
 
     private readonly transactionHttp: TransactionHttp;
     private readonly receiptHttp: ReceiptHttp;
+    private readonly listener: Listener;
     /**
      * Constructor
      * @param url Base catapult-rest url
      */
     constructor(url: string) {
         this.transactionHttp = new TransactionHttp(url);
         this.receiptHttp = new ReceiptHttp(url);
+        this.listener = new Listener(url);
     }
 
     /**
@@ -65,7 +71,7 @@ export class TransactionService implements ITransactionService {
                 (resolution.unresolved as NamespaceId).equals(unresolved));
 
         if (!resolutionStatement) {
-            throw new Error('No resolution statement found');
+            throw new Error(`No resolution statement found for unsolved value: ${unresolved.toHex()}`);
         }
         // source (0,0) is reserved for blocks, source (n, 0) is for txes, where n is 1-based index
         const resolutionEntry = resolutionStatement.resolutionEntries
@@ -74,7 +80,7 @@ export class TransactionService implements ITransactionService {
                 entry.source.secondaryId === (aggregateTransactionIndex !== undefined ? transactionIndex + 1 : 0));
 
         if (!resolutionEntry) {
-            throw new Error('No resolution entry found');
+            throw new Error(`No resolution entry found for unsolved value: ${unresolved.toHex()}`);
         }
 
         return resolutionEntry.resolved;
@@ -91,4 +97,39 @@ export class TransactionService implements ITransactionService {
                 toArray(),
             );
     }
+
+    /**
+     * @param signedTransaction Signed transaction to be announced.
+     * @returns {Observable<Transaction>}
+     */
+    public announce(signedTransaction: SignedTransaction): Observable<Transaction> {
+        return this.transactionHttp.announce(signedTransaction).pipe(
+            flatMap(() => this.listener.confirmed(signedTransaction.getSignerAddress(), signedTransaction.hash)),
+        );
+    }
+
+    /**
+     * Announce aggregate transaction
+     * @param signedTransaction Signed aggregate bonded transaction.
+     * @returns {Observable<AggregateTransaction>}
+     */
+    public announceAggregateBonded(signedTransaction: SignedTransaction): Observable<AggregateTransaction> {
+        return this.transactionHttp.announceAggregateBonded(signedTransaction).pipe(
+            flatMap(() => this.listener.aggregateBondedAdded(signedTransaction.getSignerAddress(), signedTransaction.hash)),
+        );
+    }
+
+    /**
+     * Announce aggregate bonded transaction with lock fund
+     * @param signedHashLockTransaction Signed hash lock transaction.
+     * @param signedAggregateTransaction Signed aggregate bonded transaction.
+     * @returns {Observable<AggregateTransaction>}
+     */
+    public announceHashLockAggregateBonded(signedHashLockTransaction: SignedTransaction,
+                                           signedAggregateTransaction: SignedTransaction): Observable<AggregateTransaction> {
+        return this.announce(signedHashLockTransaction).pipe(
+            flatMap(() => this.announceAggregateBonded(signedAggregateTransaction)),
+        );
+
+    }
 }

src/service/interfaces/ITransactionService.ts

@@ -15,6 +15,8 @@
  */
 
 import {Observable} from 'rxjs';
+import { AggregateTransaction } from '../../model/transaction/AggregateTransaction';
+import { SignedTransaction } from '../../model/transaction/SignedTransaction';
 import { Transaction } from '../../model/transaction/Transaction';
 
 /**
@@ -24,7 +26,29 @@ export interface ITransactionService {
 
     /**
      * @param transationHashes List of transaction hashes.
-     * @returns Observable<Transaction[]>
+     * @returns {Observable<Transaction[]>}
      */
     resolveAliases(transationHashes: string[]): Observable<Transaction[]>;
+
+    /**
+     * @param signedTransaction Signed transaction to be announced.
+     * @returns {Observable<Transaction>}
+     */
+    announce(signedTransaction: SignedTransaction): Observable<Transaction>;
+
+    /**
+     * Announce aggregate transaction
+     * @param signedTransaction Signed aggregate bonded transaction.
+     * @returns {Observable<AggregateTransaction>}
+     */
+    announceAggregateBonded(signedTransaction: SignedTransaction): Observable<AggregateTransaction>;
+
+    /**
+     * Announce aggregate bonded transaction with lock fund
+     * @param signedHashLockTransaction Signed hash lock transaction.
+     * @param signedAggregateTransaction Signed aggregate bonded transaction.
+     * @returns {Observable<AggregateTransaction>}
+     */
+    announceHashLockAggregateBonded(signedHashLockTransaction: SignedTransaction,
+                                    signedAggregateTransaction: SignedTransaction): Observable<AggregateTransaction>;
 }