Búsqueda de sitios web

typescript - Añadiendo pipes después de suscribirse a un servicio de notificaciones push

Situación:
He encontrado un caso de uso para el sistema Observable de rxjs, donde puede que necesite agregar comandos pipe a una Subscription después de que haya sido iniciada.

En mi caso, la aplicación en la que estoy trabajando tiene que escuchar de manera pasiva un sistema de notificaciones push. Se pueden enviar varios mensajes a través de este sistema, a los cuales mi sistema necesita responder. Sin embargo, hay un caso previsible en el que una vista cargada dinámicamente, que se implementará en el futuro, necesitará agregar un oyente al sistema de notificaciones push.

Pregunta:
Dado que mi aplicación está en un estado donde mi Subscription ya existe, ¿puedo agregar un pipe adicional después de que se haya invocado .subscribe(() => {})?

// this.something is an Observable<any>, for discussion purposes.
const subscription = this.something.subscribe(() => { // commands });

this.something.pipe(
  map((something) => {
    // ...Commands that I want to add to the subscription...
  })
);

...¿Y si hago eso, entonces qué pasa, si es que pasa algo?

Solución:
Las dos respuestas de @user2216584 y @SerejaBogolubov ambas tenían un aspecto de la respuesta a esta pregunta.

Mi servicio de escucha de notificaciones push de alto nivel necesitaba hacer dos cosas:

  1. Mantén la suscripción, y
  2. Ser capaz de seleccionar de una lista de oyentes.

La complicación es que cada oyente necesita estar escuchando un mensaje diferente. Dicho de otro modo, si recibo un mensaje en foo_DEV, la aplicación necesita hacer algo diferente a si el sistema de notificaciones push envía un mensaje en bar_DEV.

Entonces, esto es lo que se me ocurrió:

export interface PushNotificationListener {
  name: string,
  onMessageReceived: (msg: PushNotificationMessage) => any,
  messageSubject$: Subject<PushNotificationMessage>
}

export class PushNotificationListenerService {
  private connection$: Observable<PushNotificationConnection>;
  private subscription$: Subscription;

  private listeners: PushNotificationListener[] = [];

  constructor(
    private connectionManager: PushNotificationConnectionManager
  ) {
  }

  connect() {
    // Step 1 - Open the socket connection!
    this.connection$ = this.connectionManager.connect(
      // The arguments for setting up the websocket are unimportant here.
      // The underlying implementation is similarly unimportant.
    );
  } 

  setListener(
    name: string,
    onMessageReceived: (msg: PushNotificationMessage) => any
  ) {
    // Step 3...or maybe 2...(shrug)...
    // Set listeners that the subscription to the high-order connection
    // will employ.
    const newListener: PushNotificationListener = {
      name: name,
      onMessageReceived: onMessageReceived,
      messageSubject$: null
    };

    this.listeners.push(newListener);
  }

  listen() {
    // Step 2 - Listen for changes to the high-order connection observable.
    this.subscription$ = this.connection$
      .subscribe((connection: PushNotificationConnection) => {
        console.info('Push notification connection established');

        for (let listener of this.listeners) {
         listener.messageSubject$ = connection.subscribe(listener.name);
         listener.messageSubject$.subscribe((message: PushNotificationMessage) => {
           listener.onMessageReceived(message);
         }
        }
      },
      (error: any) => {
        console.warn('Push notification connection error', error);
      }
  }
}

Descubrí a través de un estudio cuidadoso del código interno que compone el núcleo de mi sistema de notificaciones push, que ya tenemos un Observable de orden superior. El código del websocket crea un observable (connectionManager.connect()), que necesita ser almacenado en caché en el servicio y suscribirse a él. Como ese código es específico del lugar donde trabajo, no puedo decir más al respecto.

¡Sin embargo, almacenar en caché a los oyentes también es importante! La llamada subscribe en .listen() simplemente itera a través de todos los oyentes adjuntos cada vez que la conexión cambia de estado, así que puedo agregar oyentes de manera espontánea mediante .addListener(), y debido a cómo funciona inherentemente el sistema de Observable de rxjs, Y al hecho de que estoy trabajando desde una lista de oyentes dentro del alcance, tengo un sistema mediante el cual puedo configurar oyentes dinámicamente, incluso si .connect() se llama antes de que se configuren oyentes.

Este código probablemente todavía pueda beneficiarse de un rediseño/refactorización, pero tengo algo que funciona, lo cual es el primer paso importante de cualquier buena programación. ¡Gracias a todos!

Respuesta #1

[Estoy editando mi respuesta porque la respuesta anterior era según el primer código compartido por el autor; como se menciona en el comentario, el autor ha cambiado/corregido el código] -

Dudo que el siguiente código afecte algo en la suscripción -

this.something.pipe(
  map((something) => {
    // ...Commands that I want to add to the subscription...
  })
);

Podrías intentar una función de orden superior mientras configuras inicialmente tu observable y si la función de orden superior está en el ámbito, puedes reasignarla. También dudo que funcione por las siguientes razones -

  1. Cuando se configura Observable, observable mantiene la referencia de la función pasada, la cual se invocará al suscribirse [https://medium.com/@benlesh/learning-observable-by-building-observable-d5da57405d87]. Ahora, si reasignas la función de orden superior, la función observable todavía apunta a la referencia antigua. Al reasignar la función de orden superior, no has cambiado la referencia de tu función original que se estableció cuando configuraste inicialmente el observable.

  2. Supongamos que por alguna razón, la reasignación de orden superior funciona. En ese caso, también existe una buena probabilidad de que antes de que se ejecute tu función de orden superior anterior, ya hayas reasignado tu función de orden superior (porque si el observable de origen hace una llamada asíncrona al backend, mientras el código está en espera, el bucle de eventos de JavaScript podría haber reasignado la función de orden superior y cuando la llamada asíncrona regrese, se ejecutará la nueva función de orden superior asignada). Tal vez este código aclare mi punto

let higherOrderFunc=map(x => x * 2);

this.something
    .pipe(
          mergeMap(_ => //call to backend; async call),
          higherOrderFunc,
         ).subscribe();
higherOrderFunc = map(x => x * 3); // this will execute before async call completes

Respuesta #2

Bueno, podrías hacerlo bastante fácilmente. Digamos que quieres un map diferido en tiempo de ejecución. Entonces haces algo como map(this.myMapper), donde myMapper es un campo privado visible dentro del ámbito apropiado. Al mutar ese campo privado, puedes añadir o eliminar comportamientos adicionales. Por ejemplo, map(x => x) significaría ausencia de cualquier mapeo.

Sin embargo, me parece que estás abusando de rxjs. Lo más probable es que lo que realmente necesites sea un observable de orden superior (observable que emite observables, "flujo de flujos"). Eso sería una solución mucho más rxjsica y limpia. Así que piénsalo dos veces.