Added #352 aggregate bonded announcing

Author: rg911

src/infrastructure/Listener.ts

@@ -239,7 +239,7 @@ 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 transaction hash for the filter
+     * @param transactionHash transactionHash for filtering multiple transactions
      * @return an observable stream of Transaction with state confirmed
      */
     public confirmed(address: Address, transactionHash?: string): Observable<Transaction> {
@@ -249,7 +249,7 @@ export class Listener {
             filter((_) => _.message instanceof Transaction),
             map((_) => _.message as Transaction),
             filter((_) => this.transactionFromAddress(_, address)),
-            filter((_) => _.transactionInfo!.hash === transactionHash || transactionHash === undefined),
+            filter((_) => transactionHash === undefined || _.transactionInfo!.hash === transactionHash),
         );
     }
 
@@ -292,7 +292,7 @@ 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 transaction hash for the filter
+     * @param transactionHash transactionHash for filtering multiple transactions
      * @return an observable stream of AggregateTransaction with missing signatures state
      */
     public aggregateBondedAdded(address: Address, transactionHash?: string): Observable<AggregateTransaction> {
@@ -302,7 +302,7 @@ export class Listener {
             filter((_) => _.message instanceof AggregateTransaction),
             map((_) => _.message as AggregateTransaction),
             filter((_) => this.transactionFromAddress(_, address)),
-            filter((_) => _.transactionInfo!.hash === transactionHash || transactionHash === undefined),
+            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,6 +16,8 @@
 
 import {Observable, of} from 'rxjs';
 import { map, 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 { NamespaceId } from '../model/namespace/NamespaceId';
@@ -40,13 +42,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);
     }
 
     /**
@@ -143,4 +147,39 @@ export class TransactionService implements ITransactionService {
             map((statement) => transaction.resolveAliases(statement, aggregateIndex)),
         );
     }
+
+    /**
+     * @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>;
 }