Buenas Soy nuevo en el foro y me pidieron hacer un tp para la univ. y estoy medio complicado. La idea es hacer una librería para crear y manipular Observables, capaces de emitir elementos, de modo que los que se suscriban a estos sean capaces de reaccionar a dicha secuencia de eventos. Un observable entiende la interfaz:

    trait Observable[T] {       def subscribe(observer : (T) => Unit) : Subscription     } 

NOTA: esta es la forma “corta” de subscribirse, donde sólo nos interesa recibir elementos generados. La forma “completa” sería

    trait Observable[T] {       def subscribe(subscriber: Subscriber[T]) : Subscription     } 

Subscriber permite recibir no sólo los elementos, sino tambin ser notificado cuando hay un error y cuando ya no hay más elementos para generar.

En ambos casos se retorna un objeto de tipo Subscription que luego podemos utilizar para dejar de escuchar (Esto no se pide en los siguientes puntos del TP, simplemente es para entender la idea)

     trait Subscription {         def unsubscribe() : Unit       } 

Creación de observables

Create

Queremos implementar la función create, que recibe como argumento una función de Subscriber => Unit, donde Subscriber tiene la interfaz:

    trait Subscriber[-T] {      /**       * Recibe un elemento nuevo      *        * @param t el elemento emitido       */     def onNext(t: T): Unit       /**      * Termina con error.      *       * @param t el throwable emitido      */     def onError(t: Throwable): Unit       /**     * Termina con success.     * No se enviarán más eventos     */     def onComplete(): Unit     } 

Por ejemplo, si quisiéramos crear un Observable que emitiese cierto rango de números:

     val o = Observable.create { subscriber =>         try {         Range(1,5).foreach { num =>         // se emite el elemento         subscriber.onNext(num)         }         // se completa exitosamente         subscriber.onCompleted()          } catch {         //si hubo un fallo se emite el error         case e:Exception => subscriber.onError(e)        }      }      o.subscribe { n =>         println(n)      }      // Output:     // 1     // 2     // 3     // 4 

Donde el subscribe que recibe un bloquedebería ser una manera corta de escribir:

      o.subscribe(new Subscriber[Int] {       def onNext(n: Int): Unit = {       println(n)       }       def onError(t: Throwable): Unit = {}        def onComplete(): Unit = { }        }) 

Se debe tener en cuenta que puede haber muchas subscripciones al mismo Observable. Los observables no comienzan a emitir elementos hasta no recibir su primer subscribe.

La firma del método subscribe debe poder incluir parámetros opcionales que reciban callbacks para los eventos de onCompleted y onError.

From

Observable que emite una secuencia de elementos

      val strings:Observable[String] = Observable.from("a", "b", "c")        val list = List(1,2,3)       val integers:Observable[Int] = Observable.from(list) 

Hasta acá, alguna ayuda para comenzar a implementar esto? Despúes tengo mas métodos del estilo de from como just, empty, never, etc. para crear Obs con distintos comportamientos, el tema es que estoy bastante erratico con Scala y me cuesta llevar a código lo que tengo en la cabeza. Si tiene exito el tema subo los otros métodos que tengo q implementar. Saludos