Improve subscriptions for observable handling (#6236)

* fix: subscribeNewHeads for improved observable handling

* fix: replace switchMap with mergeMap for improved subscription handling

Author: ap211unitech

packages/api-derive/src/chain/subscribeFinalizedBlocks.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { SignedBlockExtended } from '../type/types.js';
 import type { DeriveApi } from '../types.js';
 
-import { switchMap } from 'rxjs';
+import { mergeMap } from 'rxjs';
 
 import { memo } from '../util/index.js';
 
@@ -22,7 +22,7 @@ import { memo } from '../util/index.js';
 export function subscribeFinalizedBlocks (instanceId: string, api: DeriveApi): () => Observable<SignedBlockExtended> {
   return memo(instanceId, (): Observable<SignedBlockExtended> =>
     api.derive.chain.subscribeFinalizedHeads().pipe(
-      switchMap((header) =>
+      mergeMap((header) =>
         api.derive.chain.getBlock(header.createdAtHash || header.hash)
       )
     )

packages/api-derive/src/chain/subscribeFinalizedHeads.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { Hash, Header } from '@polkadot/types/interfaces';
 import type { DeriveApi } from '../types.js';
 
-import { from, of, switchMap } from 'rxjs';
+import { from, mergeMap, of, switchMap } from 'rxjs';
 
 import { memo } from '../util/index.js';
 
@@ -43,7 +43,7 @@ export function subscribeFinalizedHeads (instanceId: string, api: DeriveApi): ()
     let prevHash: Hash | null = null;
 
     return api.rpc.chain.subscribeFinalizedHeads().pipe(
-      switchMap((header) => {
+      mergeMap((header) => {
         const endHash = prevHash;
         const startHash = header.parentHash;
 

packages/api-derive/src/chain/subscribeNewBlocks.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { SignedBlockExtended } from '../type/types.js';
 import type { DeriveApi } from '../types.js';
 
-import { switchMap } from 'rxjs';
+import { mergeMap } from 'rxjs';
 
 import { memo } from '../util/index.js';
 
@@ -22,7 +22,7 @@ import { memo } from '../util/index.js';
 export function subscribeNewBlocks (instanceId: string, api: DeriveApi): () => Observable<SignedBlockExtended> {
   return memo(instanceId, (): Observable<SignedBlockExtended> =>
     api.derive.chain.subscribeNewHeads().pipe(
-      switchMap((header) =>
+      mergeMap((header) =>
         api.derive.chain.getBlock(header.createdAtHash || header.hash)
       )
     )

packages/api-derive/src/chain/subscribeNewHeads.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { HeaderExtended } from '../type/types.js';
 import type { DeriveApi } from '../types.js';
 
-import { map, switchMap } from 'rxjs';
+import { map, mergeMap } from 'rxjs';
 
 import { createHeaderExtended } from '../type/index.js';
 import { memo } from '../util/index.js';
@@ -25,7 +25,7 @@ import { getAuthorDetails } from './util.js';
 export function subscribeNewHeads (instanceId: string, api: DeriveApi): () => Observable<HeaderExtended> {
   return memo(instanceId, (): Observable<HeaderExtended> =>
     api.rpc.chain.subscribeNewHeads().pipe(
-      switchMap((header) =>
+      mergeMap((header) =>
         getAuthorDetails(api, header)
       ),
       map(([header, validators, author]): HeaderExtended => {