¿Hay un operador RxJS que espere hasta que la fuente se complete y luego emita un valor dado? Si no hay ninguno, ¿cómo podría proporcionarlo por mi cuenta?
Esto funcionaría de manera similar a toArray() , que también espera a que se complete la fuente. En lugar de recopilar todos los valores emitidos, quiero ignorarlos y devolver un valor diferente.
Aquí hay una implementación equivalente:
observable.pipe( ignoreElements(), endWith(myValue), );alternativamente:
observable.pipe( toArray(), map(ignore => myValue) ) A menudo hay situaciones en las que necesito esto. Llegué a la conclusión de que es peligroso convertir promesas then cadenas en observables mediante switchMap() o mergeMap() porque los observables internos podrían completarse sin emitir ningún valor. Recientemente tuvimos este problema:
return getEntitiesFromBackend().pipe( switchMap(entities => { return putEntitiesToObjectStore(entities); }), switchMap(() => { return storeSyncDate(); }) ); En algunas situaciones, la fecha de sincronización no se almacenaba y era difícil saber por qué. Al final, la razón fue que el método putEntities... emite un valor para su operación "put". Pero en esos casos, la matriz de entities estaba vacía, por lo que no se emitió ningún valor.
Esto es lo que realmente quería hacer, traducido al mundo de las promesas:
return getEntitiesFromBackend() .then(entities => { return putEntitiesToObjectStore(entities); }) .then(() => { return storeSyncDate(); }) ); La mayor parte del código que vi que usa switchMap / mergeMap no tiene este problema. Porque la mayoría de las veces se trata de solicitudes HTTP que solo se emiten una vez y luego se completan. Ver aquí por ejemplo. Esto hizo que me acostumbrara a convertir patrones de promesa típicos al mundo RxJS con switchMap sin pensar demasiado en su funcionamiento y propósito reales. Ahora que trabajamos con IndexedDB, la mayoría de nuestros métodos devuelven observables que emiten un valor para cada operación de base de datos. switchMap / mergeMap tendrá una pesadilla aquí.
Es por eso que solicito un operador de este tipo y me pregunto por qué no pude encontrarlo todavía, ya que es un caso tan común en nuestra aplicación. Podría resolver esto fácilmente usando las implementaciones alternativas anteriores, pero no quiero repetir esos dos operadores una y otra vez:
return getEntitiesFromBackend().pipe( switchMap(entities => { return putEntitiesToObjectStore(entities); }), ignoreElements(), endWith(), switchMap(() => { return storeSyncDate(); }) ); Por supuesto, podría usar toArray() y simplemente ignorar el argumento en el siguiente operador. No me gusta porque causaría una sobrecarga innecesaria.
Me parece que quieres lograr esto:
source.pipe( last(), // emits only last value, when the source completes map(() => myValue), // or mapTo(myValue) )Yo sugeriría dos cambios.
Use concatMap en lugar de switchMap o mergeMap . concatMap garantizaría que cada emisión del Observable getEntitiesFromBackend() se reenviaría secuencialmente una tras otra y no en paralelo ( mergeMap ) o se cancelaría ( switchMap ). Vea aquí una breve introducción a los diferentes tipos de operadores de mapeo de orden superior.
En lugar de una combinación de operadores como ignoreElements + map , podría usar el last operador con un predicado que siempre devuelve false con un valor predeterminado. De esta manera, el valor predeterminado se emitiría cuando se complete la fuente observable.
return getEntitiesFromBackend().pipe( concatMap(entities => { return putEntitiesToObjectStore(entities).pipe( last(() => false, myValue), ); }), concatMap((myValue) => { return storeSyncDate(); }) );Algunas opciones vienen a la mente.
defaultIfEmptyEsto no es exactamente lo que está describiendo, pero puede adaptarse a su caso de uso.
defaultIfEmpty: emite un valor dado si el Observable de origen se completa sin emitir ningún valor siguiente; de lo contrario, refleja el Observable de origen.
return getEntitiesFromBackend().pipe( switchMap(entities => putEntitiesToObjectStore(entities)), defualtIfEmpty(null), switchMap(_ => storeSyncDate()) ); Existe una versión estática de pipe que realiza la composición sin la aplicación a ningún flujo determinado.
Así por ejemplo:
const a = source.pipe( ignoreElements(), endWith(myValue) ); const b = pipe( ignoreElements(), endWith(myValue) ); Aquí, a es un observable. Pipe compone dos operadores, luego devuelve el resultado de aplicar el operador compuesto a la source .
b por otro lado es solo el primer paso. b es un operador en sí mismo. El operador no se ha aplicado a un observable.
Así que puedes hacer algo como esto:
source1.pipe(b); source2.pipe(b); He reutilizado mi operador b dos veces. ¡Ya hemos recorrido la mayor parte del camino!
Los operadores RxJS son muy útiles porque son devueltos por funciones que personalizan el operador según sus necesidades. En el caso anterior, cada vez que use b , tendrá el mismo valor para endwith .
Podemos envolver b en una función para personalizarla según el uso.
const myCustomOperator = myValue => pipe( ignoreElements(), endWith(myValue) ); source1.pipe( myCustomOperator(22) ); source2.pipe( myCustomOperator(23) );Esto funciona como cualquier otro operador, por lo que también se puede componer (ser "canalizado") con todos los demás operadores estándar.
Podría resolver esto fácilmente usando las implementaciones alternativas anteriores, pero no quiero repetir esos dos operadores una y otra vez:
¡Ahora tienes un código reutilizable!
JavaScript:
function ignoreThenConcat(genObs) { return pipe( ignoreElements(), concatWith(from(genObs())) ); } [...] return getEntitiesFromBackend().pipe( switchMap(entities => putEntitiesToObjectStore(entities)), ignoreThenConcat(() => storeSyncDate()) ); [...]Encuentro que ver el código tipeado estáticamente a menudo realmente ayuda a mi comprensión. Así que aquí está lo mismo en TypeScript:
function ignoreThenConcat<T,R>(genObs: () => ObservableInput<R>): OperatorFunction<T,R> { return pipe( ignoreElements(), concatWith(from(genObs())) ); } [...] return getEntitiesFromBackend().pipe( switchMap(entities => putEntitiesToObjectStore(entities)), ignoreThenConcat(() => storeSyncDate()) ); [...]