TypeScript / Observable Subject

Bienvenidos sean a este post, hoy veremos una clase interesante.

Si vienen de los posts anteriores hasta ahora vimos que los stream observables emiten un valor y como suscribirse a estos. El observable en si mismo es responsable por la emision de valores y los suscriptores reaccionan a los valores recibidos. Por lo tanto, cuando un stream observable es completo, todos los suscriptores completan su procesamiento y detienen su ejecucion. Pero que sucede si querremos mantener un stream abierto y registrar uno o mas suscriptores que esperan hasta que un nuevo valor sea emitido?

Para este tipo de evento en particular, rxjs nos provee a la clase Subject para esto. Esta clase mantiene una lista de listeners que han registrado su interes. Este tambien es un stream Observable y esto permite que los listeners puedan suscribirse a este usando la misma sintaxis y funciones que vinimos utilizando hasta ahora en nuestros streams. Pero que lo hace interesante? Es su habilidad de muticast, esto permite que tenga multiples suscriptores en el mismo stream y notificara a todos los suscriptores interesados cuando ocurra un evento.

Para entender este concepto vamos a analizar el siguiente codigo:

import {Observable, Subject, Subscription} from 'rxjs';
import {map, filter} from 'rxjs/operators';
import * as _ from "underscore";

enum ClaveEvento {
        TODAS = "evento-todos",
        UNA = "un-evento"
}
export interface IEventoBroadcast {
        key: ClaveEvento;
        datos: string;
}

export class ServicioBroadcast {
        private _busEvento = new Subject<IEventoBroadcast>();
        on(key: ClaveEvento): Observable<string> {
                return this._busEvento.asObservable().pipe(
                        filter(ev => ev.key === key ||
                                ev.key === ClaveEvento.TODAS),
                        map(ev => ev.datos));
        }
        broadcast(key: ClaveEvento, datos: string) {
                this._busEvento.next({key, datos});
        }
}

class Listener {
        private suscripcionEv: Subscription;
        constructor( servicioBroadcase: ServicioBroadcast,
                claveEvento: ClaveEvento,
                private nombreListener: string) {
                _.bindAll(this, "reaccionAevento");
                this.suscripcionEv = servicioBroadcast.on(claveEvento)
                        .subscribe(this.reaccionAevento);
        }
        private reaccionAevento(evento: string) {
                console.log('Listener: [' + this.nombreListener + ']'
                + '\nEvento recibido: ' + evento);
        }
        public unregister() {
                this.suscripcionEv.unsubscribe();
        }
}

const servicioBroadcast = new ServicioBroadcast();
const primero = new Listener(servicioBroadcast,
        ClaveEvento.TODAS, "primero");
const segundo = new Listener(servicioBroadcast,
        ClaveEvento.UNA, "segundo");

servicioBroadcast.broadcast(ClaveEvento.TODAS,
        "Broadcast de todos los eventos");
servicioBroadcast.broadcast(ClaveEvento.UNA,
        "Broadcast de un solo evento");
servicioBroadcast.broadcast(ClaveEvento.TODAS,
        "Broadcast de todos los eventos, nuevamente");

Las tres primeras lineas son las encargadas de importar todos nuestros elementos. La primera importara el tipo, funciones y clases para trabajar. La segunda es para los operadores observables y por ultimo importaremos todo el modulo underscore y le haremos un alias para usarlo de esa forma. Esta ultima la usaremos par un solo tema especifico en el codigo.

Nuestro primer paso sera definir un enum que contendra dos constantes, estas sera para indicar dos tipos de eventos con sus respectivos identificaciones. Luego definimos una interfaz para el evento de broadcast y su primera propiedad sera del tipo del enum y tendra uno para almacenar.

Despues de esto definiremos una clase para el servicio de broadcast. Lo primero que haremos sera definir un objeto de la clase Subject y tendra el tipo de la interfaz que declaramos anteriormente. Luego tenemos a la funcion on que se encaga de escuchar cualquier evento y en este caso solo recibira un argumento que sera la clave del evento. y devuelve un Observable de tipo string. En el bloque devolvera el objeto creado de Subject pero del objeto creado de esta clase, por eso el this, y luego le aplicamos el metodo asObservable para asegurarnos que sea de este tipo y a su vez asegurarnos que se crea un nuevo stream Observable. Por lo tanto podemos aplicarle un pipe para verificarlo. El primer operador es un filter para verificar que el evento coincida con los iniciados y en caso de ser verdadero notificara a los observadores registrados. Por lo tanto, la funcion avisara solamente a uno cuando ocurre el evento UNA pero notificara a todos cuando sea el evento TODAS. La ultima funcion es una de las importadas al comienzo y esta tiene solo dos parametros, la clave y el dato, los cuales usara para realizar el broadcast correspondiente a los observadores registrados.

La siguiente clase sera la «escuchadora» y lo primero que haremos sera declarar una propiedad que sera del tipo Subscription. Esta clase representa al lugar donde trabajaremos nuestro Observable. Luego tenemos un constructor donde primero definiremos un argumento que sera del tipo anterior, el siguiente es otro argumento del tipo del enum y por ultimo uno privado que sera para asignar el nombre del listener. En el bloque usaremos la funcion bindAll para unir el metodo pasado como argumento al objeto en el primer argumento, en este caso this o el objeto creado desde esta clase, y en la ultima linea iniciaremos a la propiedad que declaramos anteriormente. En este asignaremos el objeto de la clase anterior que definimos en el constructor y usamos el metodo on y el subscribe con el metodo reaccionAevento.

Nuestro siguiente paso sera definir al metodo reaccionAevento, este recibe un argumento del tipo string. En el bloque mostraremos en consola un mensaje con todos los datos que le pasamos mediante el bindAll anterior. Por ultimo, definimos un metodo para desregistrar a un suscriptor y para ello simplemennte utilizamos al objeto asignado y a la suscripcion y le aplicamos el metodo para quitarlo.

Con todo definido pasemos al programa en si. Primero definimos un objeto del tipo ServicioBroadcast. Despues definiremos dos objetos del tipo Listener, con dos identificadores distintos y una clave de evento para cada uno. Por ulitmo haremos tres llamados a broadcast:

• La primera pasa la clave TODAS y un mensaje
• La segunda pasa la clave UNA y otro mensaje
• La tercera pasa la clave TODAS y un ultimo mensaje

Con esto podemos probar como trabaja, para verlo en accion deben compilarlo y ejecutarlo:

$ node mirar.js
Listener: [primero]
Evento recibido: Broadcast de todos los eventos
Listener: [segundo]
Evento recibido: Broadcast de todos los eventos
Listener: [segundo]
Evento recibido: Broadcast de un solo evento
Listener: [primero]
Evento recibido: Broadcast de todos los eventos, nuevamente
Listener: [segundo]
Evento recibido: Broadcast de todos los eventos, nuevamente
$

En la salida pueden ver como se respeto el tema de la clave de los eventos. Por que al momento de usar a TODAS se notifica que los dos listener reciben el mensaje. En cambio cuando usamos la clave UNA solo se manda al listener correspondiente. Pero antes de finalizar vamos a desregistrar uno de los listener, para ello deben agregar un llamado a unregister antes de la ejecucion de los broadcasts:

... codigo anterior...

primero.unregister();
servicioBroadcast.broadcast(ClaveEvento.TODAS,
        "Broadcast de todos los eventos"); ... resto del codigo

Simplemente agregan esa linea y nos desregistrara al primer «escuchador» de la suscripcion. Si lo compilan y ejecutan veran lo siguiente:

$ node mirar.js
Listener: [segundo]
Evento recibido: Broadcast de todos los eventos
Listener: [segundo]
Evento recibido: Broadcast de un solo evento
Listener: [segundo]
Evento recibido: Broadcast de todos los eventos, nuevamente
$

En esta ocasion desaparecio el listener indicado como primero y segundo responde a todos los llamados porque como dijimos anteriormente TODAS es para todos los suscriptores.

En este codigo vimos lo que se llama bus de eventos. Donde al unirse a la subscripcion nos garantiza que recibiran el evento que estan esperando. Esta pensado para notificar a elementos no relacionados sobre un evento en particular. Por ejemplo, si un usuario cierra sesion en el sitio y suponiendo que dejo cosas en su carrito mediante esto podemos notificarle a los elementos relacionados al cierre de sesion, asi como tambien quitar acceso a ofertas y otros beneficios de tener sesion iniciada. Asi como esto podemos aplicarlo en muchas otras situaciones para beneficiar a nuestro codigo.

En resumen, hoy hemos visto a Subject, que es, para que sirve, como se utiliza, un ejemplo para verlo en accion y como trabaja. Espero les haya sido de utilidad sigueme en tumblr, Twitter o Facebook para recibir una notificacion cada vez que subo un nuevo post en este blog, nos vemos en el proximo post.