node.js - ¿Cómo monitorear el número de suscripciones RXJS?
Estoy usando un Observable para proporcionar una interfaz de suscripción a eventos para los clientes desde un recurso global, y necesito gestionar ese recurso según el número de suscripciones activas:
- Asignar recurso global cuando el número de suscripciones sea mayor que 0
- Liberar recurso global cuando el número de suscripciones sea 0
- Ajusta la estrategia de uso de recursos según el número de suscripciones
¿Cuál es la manera adecuada en RXJS de monitorear el número de suscripciones activas?
¿Cómo implementar lo siguiente dentro de la sintaxis de RXJS? -
const myEvent: Observable<any> = new Observable();
myEvent.onSubscription((newCount: number, prevCount: number) => {
if(newCount === 0) {
// release global resource
} else {
// allocate global resource, if not yet allocated
}
// for a scalable resource usage / load,
// re-configure it, based on newCount
});
No esperaría una notificación garantizada en cada cambio, de ahí los parámetros newCount + prevCount.
ACTUALIZACIÓN-1
Esto no es un duplicado de esto, porque necesito ser notificado cuando cambie el número de suscripciones, y no solo para obtener el contador en algún momento.
ACTUALIZACIÓN-2
Sin ninguna respuesta hasta ahora, rápidamente ideé una solución muy fea y limitada, mediante encapsulación completa, y específicamente para el tipo Subject. Esperando mucho encontrar una solución adecuada.
ACTUALIZACIÓN-3
Después de algunas respuestas, todavía no estoy seguro de cómo implementar lo que estoy intentando, que es lo siguiente:
class CustomType {
}
class CountedObservable<T> extends Observable<T> {
private message: string; // random property
public onCount; // magical Observable that needs to be implemented
constructor(message: string) {
// super(); ???
this.message = message;
}
// random method
public getMessage() {
return this.message;
}
}
const a = new CountedObservable<CustomType>('hello'); // can create directly
const msg = a.getMessage(); // can call methods
a.subscribe((data: CustomType) => {
// handle subscriptions here;
});
// need that magic onCount implemented, so I can do this:
a.onCount.subscribe((newCount: number, prevCont: number) => {
// manage some external resources
});
¿Cómo implementar tal clase CountedObservable arriba, que me permitiría suscribirme a sí misma, así como a su propiedad onCount para monitorear el número de sus clientes/suscripciones?
ACTUALIZACIÓN-4
Todas las soluciones sugeridas parecían excesivamente complejas, y aunque acepté una de las respuestas, terminé con una solución completamente personalizada propia.
Respuesta #1
Podrías lograrlo usando defer para rastrear suscripciones y finalize para rastrear finalizaciones, por ejemplo como un operador:
// a custom operator that will count number of subscribers
function customOperator(onCountUpdate = noop) {
return function refCountOperatorFunction(source$) {
let counter = 0;
return defer(()=>{
counter++;
onCountUpdate(counter);
return source$;
})
.pipe(
finalize(()=>{
counter--;
onCountUpdate(counter);
})
);
};
}
// just a stub for `onCountUpdate`
function noop(){}
Y luego úsalo así:
const source$ = new Subject();
const result$ = source$.pipe(
customOperator( n => console.log('Count updated: ', n) )
);
Aquí hay un fragmento de código que ilustra esto:
const { Subject, of, timer, pipe, defer } = rxjs;
const { finalize, takeUntil } = rxjs.operators;
const source$ = new Subject();
const result$ = source$.pipe(
customOperator( n => console.log('Count updated: ', n) )
);
// emit events
setTimeout(()=>{
source$.next('one');
}, 250);
setTimeout(()=>{
source$.next('two');
}, 1000);
setTimeout(()=>{
source$.next('three');
}, 1250);
setTimeout(()=>{
source$.next('four');
}, 1750);
// subscribe and unsubscribe
const subscriptionA = result$
.subscribe(value => console.log('A', value));
setTimeout(()=>{
result$.subscribe(value => console.log('B', value));
}, 500);
setTimeout(()=>{
result$.subscribe(value => console.log('C', value));
}, 1000);
setTimeout(()=>{
subscriptionA.unsubscribe();
}, 1500);
// complete source
setTimeout(()=>{
source$.complete();
}, 2000);
function customOperator(onCountUpdate = noop) {
return function refCountOperatorFunction(source$) {
let counter = 0;
return defer(()=>{
counter++;
onCountUpdate(counter);
return source$;
})
.pipe(
finalize(()=>{
counter--;
onCountUpdate(counter);
})
);
};
}
function noop(){}<script src="https://unpkg.com/[email /bundles/rxjs.umd.min.js"></script>* NOTA: si tu fuente$está fría, es posible que necesites compartirla.
Espero que ayude
Respuesta #2
Realmente estás haciendo tres preguntas separadas aquí, y me pregunto si realmente necesitas la capacidad completa que mencionas. Dado que la mayoría de las cosas de gestión de recursos que estás pidiendo ya están proporcionadas por la biblioteca, hacer un código de seguimiento personalizado parece redundante. Las dos primeras preguntas:
- Asignar recurso global cuando el número de suscripciones sea mayor que 0
- Liberar recurso global cuando el número de suscripciones sea 0
Se puede hacer con los operadores using + share:
class ExpensiveResource {
constructor () {
// Do construction
}
unsubscribe () {
// Do Tear down
}
}
// Creates a resource and ties its lifecycle with that of the created `Observable`
// generated by the second factory function
// Using will accept anything that is "Subscription-like" meaning it has a unsubscribe function.
const sharedStream$ = using(
// Creates an expensive resource
() => new ExpensiveResource(),
// Passes that expensive resource to an Observable factory function
er => timer(1000)
)
// Share the underlying source so that global creation and deletion are only
// processed when the subscriber count changes between 0 and 1 (or visa versa)
.pipe(share())
Después de eso, sharedStream$ puede ser pasado como un flujo base que gestionará el recurso subyacente (suponiendo que hayas implementado tu unsubscribe correctamente), de modo que el recurso se creará y se destruirá a medida que el número de suscriptores pase de 0 a 1.
Ajusta la estrategia de uso de recursos según el número de suscripciones
La tercera pregunta es sobre la que tengo más dudas, pero la responderé por completitud asumiendo que conoces tu aplicación mejor que yo (ya que no puedo pensar en una razón por la cual necesitarías un manejo específico en diferentes niveles de uso aparte de ir entre 0 y 1).
Básicamente usaría un enfoque similar al anterior, pero encapsularía la lógica de transición de manera ligeramente diferente.
// Same as above
class ExpensiveResource {
unsubscribe() { console.log('Tear down this resource!')}
}
const usingReferenceTracking =
(onUp, onDown) => (resourceFactory, streamFactory) => {
let instance, refCount = 0
// Again manage the global resource state with using
const r$ = using(
// Unfortunately the using pattern doesn't let the resource escape the closure
// so we need to cache it for ourselves to use later
() => instance || (instance = resourceFactory()),
// Forward stream creation as normal
streamFactory
)
).pipe(
// Don't forget to clean up the stream after all is said and done
// Because its behind a share this should only happen when all subscribers unsubscribe
finalize(() => instance = null)
share()
)
// Use defer to trigger "onSubscribe" side-effects
// Note as well that these side-effects could be merged with the above for improved performance
// But I prefer them separate for easier maintenance.
return defer(() => onUp(instance, refCount += 1) || r$)
// Use finalize to handle the "onFinish" side-effects
.pipe(finalize(() => onDown(instance, refCount -= 1)))
}
const referenceTracked$ = usingReferenceTracking(
(ref, count) => console.log('Ref count increased to ' + count),
(ref, count) => console.log('Ref count decreased to ' + count)
)(
() => new ExpensiveResource(),
ref => timer(1000)
)
referenceTracked$.take(1).subscribe(x => console.log('Sub1 ' +x))
referenceTracked$.take(1).subscribe(x => console.log('Sub2 ' +x))
// Ref count increased to 1
// Ref count increased to 2
// Sub1 0
// Ref count decreased to 1
// Sub2 0
// Ref count decreased to 0
// Tear down this resource!
Advertencia: Uno de los efectos secundarios de esto es que, por definición, el flujo estará cálido una vez que salga de la función usingReferenceTracking, y se calentará en la primera suscripción. Asegúrese de tener esto en cuenta durante la fase de suscripción.