Cómo divertirse en RxGo

Answer

1. 

   davy.ai

   August 5, 2023 at 9:26 am

   En el código proporcionado, las funciones observable.DoOnNext() no se ejecutan porque la variable observable se reasigna a una nueva instancia con cada llamada a Map(). Para solucionar esto y lograr la ejecución paralela del paso final utilizando DoOnNext(), podemos usar el operador RxGo#Publish() que permite transmitir en múltiples suscriptores el Observable fuente. De esta manera, podemos crear Observables separados para cada una de las funciones DoOnNext() y suscribirnos a ellos por separado sin invocar la cadena múltiples veces.

   Aquí está el código modificado usando el operador RxGo#Publish():

   var thirdCounter int32
   
   func localRun(names ...string) {
       observable := rxgo.Just(names)().
           Map(func(_ context.Context, i interface{}) (interface{}, error) {
               s := i.(string)
               s = fmt.Sprintf("%s,%s", s, "one")
               return s, nil
           }).
           Map(func(_ context.Context, i interface{}) (interface{}, error) {
               s := i.(string)
               s = fmt.Sprintf("%s,%s", s, "two")
               return s, nil
           }).
           Map(func(_ context.Context, i interface{}) (interface{}, error) {
               atomic.AddInt32(&thirdCounter, 1)
               s := i.(string)
               s = fmt.Sprintf("%s,%s", s, "three")
               return s, nil
           })
   
       // Crear observables separados para cada una de las funciones DoOnNext
       obs1, obs2, obs3 := observable.Publish().RefCount(), observable.Publish().RefCount(), observable.Publish().RefCount()
   
       obs1.DoOnNext(func(i interface{}) {
           fmt.Println("primer DoOnNext", i)
       })
   
       obs2.DoOnNext(func(i interface{}) {
           fmt.Println("segundo DoOnNext", i)
       })
   
       obs3.DoOnNext(func(i interface{}) {
           fmt.Println("tercer DoOnNext", i)
       })
   
       // Suscribirse al último observable para activar la ejecución de la cadena
       for item := range observable.Last().Observe() {
           fmt.Println(item.V)
       }
       fmt.Printf("Tercer Contador = %d\n", thirdCounter)
   }
   
   func TestMocktFlow(t *testing.T) {
       cs := make([]string, 0)
       cs = append(cs, "Hola")
       cs = append(cs, "Hola")
       localRun(cs...)
   }
   

   En este código modificado, estamos utilizando el operador RxGo#Publish() para transmitir en múltiples Observables el Observable fuente, de los cuales nos podemos suscribir por separado para ejecutar las funciones DoOnNext() sin invocar la cadena múltiples veces. El operador RefCount() se utiliza para conectar automáticamente al Observable fuente solo cuando tenemos el primer suscriptor activo, y desconectarse del Observable fuente cuando el último suscriptor activo cancela su suscripción.

Comments are closed.