import { Observable } from "rxjs" ; import { unstable_batchedUpdates } from "react-dom" ; remarks export const batchUpdates = < T > ( ) => ( source$ ) : Observable < T > => { return new Observable < T > ( ( observer ) => { const obs = { n : ( v ) => observer . next ( v ) , c : ( ) => observer . complete ( ) , e : ( e ) => observer . error ( e ) , } ; let queue = [ ] ; let promise = null ; const flush = ( ) => { promise = null ; const originalQueue = queue ; queue = [ ] ; unstable_batchedUpdates ( ( ) => { originalQueue . forEach ( ( [ prop , val ] ) => obs [ prop ] ( val ) ) ; } ) ; } ; const push = ( type ) => ( val ? ) => { queue . push ( [ type , val ] ) ; if ( ! promise ) { promise = Promise . resolve ( ) . then ( flush ) ; } } ; return source$ . subscribe ( push ( "n" ) , push ( "e" ) , push ( "c" ) ) ; } ) ; } ;