reactive-core Is EventSource a Monad ?

Very interesting question. But let me turn it around: rather than analyzing how it behaves currently, let me ask you how it should behave?

On Tue, Dec 27, 2011 at 7:48 AM, Miguel Negrão <miguel.negrao-lists@friendlyvirus.org> wrote:

Hi all,
I'm trying to check if EventSource is a monad ( and in the process writing a type class instance for scalaz). It has a flatMap/bind method already defined. I'm trying to think of what the pure/return function should be. This function should take a value and lift it into the computational context of EventStreams in a minimal way (i.e. 2 => List(2), 2=>Some(2), etc). What I came up with is creating an EventSource and then firing it with the value, but this doesn't seem to follow the monad laws.  When I flatMap there is no recollection that the original value was fired...   My question is then, is EventSource a Monad ? If so what is the pure function ?
Here is my attempt at implementing pure:
object TestPure {
  def pure[A](a: =>A) = {    val x = new EventSource[A]    x.fire(a)    x  }
  def main(args: Array[String]) {    implicit val o = new Observing {}     val a = 2
    val x = pure(a)    x.foreach(i =>println("x fired " + i))
    val y = new EventSource[Int]    y.foreach(i=>println("y fired " + i))
    val f = (x:Int) => y.map( y => (x,y) )
    //Monad law: left identity    //pure(a).flatMap(f) = f(a)    val leftEquation = x.flatMap(f)     leftEquation.foreach(i => println("left fired " + i))
    val rightEquation = f(a)    rightEquation.foreach(i => println("right fired " + i))
    y.fire(4)  }
Running the program gives:y fired 4right fired (2,4)

best,Miguel Negrão
}

Yes, I really can't think of good way to define a pure/return function for EventStream, but it should be possible to do so because in Haskell the EventStreams are monads. I guess I should check how they do it in haskel. Signal on the other hand is probably a Monad with the functions it already has defined.
Btw, I'm having a problem with reactive-core, maybe you can help: It appears that if I have an EventSource inside a map inside one of my classes, firing the source doesn't cause any output  at the functions registered with forech. Calling.hasListeners gives false. Why are the listeners not added ? Could this be related with the Observing implicit ? This doesn't happen when I create an event source outside any class.
scala>  val y = new EventSource[Double]y: reactive.EventSource[Double] = reactive.EventSource@16374a56
scala>  y.foreach(println)
scala>  y.hasListenersres7: Boolean = true
//x is an instance of a class defined by me and sources is a Map[BCRControl,EventSource[Double]]scala>  x.sources.get(Kn(0,0)).get.hasListenersres8: Boolean = false
scala>  x.sources.get(Kn(0,0)).get.foreach(println)
scala>  x.sources.get(Kn(0,0)).get.hasListenersres10: Boolean = false
scala>  x.sources.get(Kn(0,0)).getres11: reactive.EventSource[Double] = reactive.EventSource@4e644b0d
best regards,Miguel Negrão

I don't know anything about EventSource, but C# reactive extensions
treats their observable sequences as monads, and their introductory
tutorial talks about the Return and SelectMany functions for these
sequences that allow them to be used monadically:
http://go.microsoft.com/fwlink/?LinkId=208528

On Dec 27 2011, 6:48 am, Miguel Negrão wrote:
> Hi all,
>
> I'm trying to check if EventSource is a monad ( and in the process writing
> a type class instance for scalaz). It has a flatMap/bind method already
> defined. I'm trying to think of what the pure/return function should be.
> This function should take a value and lift it into the computational
> context of EventStreams in a minimal way (i.e. 2 => List(2), 2=>Some(2),
> etc). What I came up with is creating an EventSource and then firing it
> with the value, but this doesn't seem to follow the monad laws.  When I
> flatMap there is no recollection that the original value was fired...
>   My question is then, is EventSource a Monad ? If so what is the pure
> function ?
>
> Here is my attempt at implementing pure:
>
> object TestPure {
>
>   def pure[A](a: =>A) = {
>     val x = new EventSource[A]
>     x.fire(a)
>     x
>   }
>
>   def main(args: Array[String]) {
>     implicit val o = new Observing {}
>     val a = 2
>
>     val x = pure(a)
>     x.foreach(i =>println("x fired " + i))
>
>     val y = new EventSource[Int]
>     y.foreach(i=>println("y fired " + i))
>
>     val f = (x:Int) => y.map( y => (x,y) )
>
>     //Monad law: left identity
>     //pure(a).flatMap(f) = f(a)
>     val leftEquation = x.flatMap(f)
>     leftEquation.foreach(i => println("left fired " + i))
>
>     val rightEquation = f(a)
>     rightEquation.foreach(i => println("right fired " + i))
>
>     y.fire(4)
>   }
>
> Running the program gives:
> y fired 4
> right fired (2,4)
>
> best,
> Miguel Negrão
>
>
>
>
>
>
>
> }

On Friday, January 6, 2012 9:09:39 PM UTC, sreque wrote:

I don't know anything about EventSource, but C# reactive extensions
treats their observable sequences as monads, and their introductory
tutorial talks about the Return and SelectMany functions for these
sequences that allow them to be used monadically:
http://go.microsoft.com/fwlink/?LinkId=208528

Interesting. It seems that in Rx the sequences replay all the values already given to new subscribers. Although I can't see how useful that would be in practice, for instance having all previous mouse moves replayed before the event was subscribed...
best,Miguel

On Friday, January 6, 2012 10:52:37 PM UTC, Miguel Negrão wrote:

On Friday, January 6, 2012 9:09:39 PM UTC, sreque wrote:

I don't know anything about EventSource, but C# reactive extensions
treats their observable sequences as monads, and their introductory
tutorial talks about the Return and SelectMany functions for these
sequences that allow them to be used monadically:
http://go.microsoft.com/fwlink/?LinkId=208528

Interesting. It seems that in Rx the sequences replay all the values already given to new subscribers. Although I can't see how useful that would be in practice, for instance having all previous mouse moves replayed before the event was subscribed...
best,Miguel

Actually, looking at the reactive-banana library, it seems events are only Functors and Signals are Applicative Functors:
http://apfelmus.nfshost.com/blog/2011/05/06-frp-why-functors.html

"Some of these combinators, namely the type classes instances, belong to very general abstractions, like functor, applicative functor and monoid. They are applicable (pun intended) far beyond FRP, so understanding will pay off for other things as well. This is no surprise: every higher-kinded type, i.e. a type with a parameter, is simply very likely to belong to some of the abstractions here. Monads would be another common example, but it just so happens that they don’t show up in reactive-banana."
best,Miguel 

On Fri, Jan 6, 2012 at 12:04 PM, Miguel Negrão <miguel.negrao-lists@friendlyvirus.org> wrote:

Yes, I really can't think of good way to define a pure/return function for EventStream, but it should be possible to do so because in Haskell the EventStreams are monads. I guess I should check how they do it in haskel. Signal on the other hand is probably a Monad with the functions it already has defined.

I can't think of any use for it, but what about an event stream that fires an event repeatedly in an infinite loop?

Btw, I'm having a problem with reactive-core, maybe you can help: It appears that if I have an EventSource inside a map inside one of my classes, firing the source doesn't cause any output  at the functions registered with forech. Calling.hasListeners gives false. Why are the listeners not added ? Could this be related with the Observing implicit ? This doesn't happen when I create an event source outside any class.
scala>  val y = new EventSource[Double]y: reactive.EventSource[Double] = reactive.EventSource@16374a56
scala>  y.foreach(println)
scala>  y.hasListeners res7: Boolean = true
//x is an instance of a class defined by me and sources is a Map[BCRControl,EventSource[Double]]scala>  x.sources.get(Kn(0,0)).get.hasListenersres8: Boolean = false
scala>  x.sources.get(Kn(0,0)).get.foreach(println)
scala>  x.sources.get(Kn(0,0)).get.hasListenersres10: Boolean = false
scala>  x.sources.get(Kn(0,0)).get res11: reactive.EventSource[Double] = reactive.EventSource@4e644b0d

Hmm, that's pretty strange. Observings, like any implicit parameter, are pulled in via static scope of the call site, so it should not depend on where the event stream is instantiated or "owned." Can you show the code that defines and populates the map? (For instance, if you use 'withDefault' it will create a new event stream every time.)Does it have the same System.identityHashCode every time you call get for the same key? What does <eventStream>.debugString show?What if you assign it to a val, and call foreach and hasListeners on the val?Also, use reflection to watch the Observing that you're using. Try something like classOf[Observing].getMethod("refs").invoke(observing) (if that's the correct API usage).

best regards,Miguel Negrão

On Sunday, January 8, 2012 2:43:40 AM UTC, nafg wrote:

On Fri, Jan 6, 2012 at 12:04 PM, Miguel Negrão <miguel.ne...@friendlyvirus.org> wrote:

Yes, I really can't think of good way to define a pure/return function for EventStream, but it should be possible to do so because in Haskell the EventStreams are monads. I guess I should check how they do it in haskel. Signal on the other hand is probably a Monad with the functions it already has defined.

I can't think of any use for it, but what about an event stream that fires an event repeatedly in an infinite loop?

Actually I think I've hit a practical case where I could use something like the "pure" function, but I might be misunderstanding my problem, and the solution might be something else:
object ExampleReactive3 extends Observing{
  def linlin( in: Double, srcLo: Double, srcHi: Double, dstLo: Double, dstHi: Double ) : Double = {    (in - srcLo) / (srcHi - srcLo) * (dstHi - dstLo) + dstLo  }
  def main(args: Array[String]) {
    val bcr = BCR.reactiveKtlMult()    //bcr.DUMP_IN = true  //to check controller is working
    //either 0.0 or 1.0    val button = bcr.sources( Bt(0, 0) )    //values from 0.0 to 1.0    val knob = bcr.sources( Kn(0, 0) )    val empty = new EventSource[Double]        // If button1 is activated we are in calibration mode. We detect the lowest and highest value and store them    // If button1 is deactivated then we clip and scale the values coming from knob
    val x = button      .flatMap{ x => x match {        case 0.0 => empty        case 1.0 => knob      }}      .foldLeft( (0.5,0.5) ){ case ((lo,hi), x) => ( x.min(lo), x.max(hi)) }
    val result = x.flatMap{ _ match {        case (lo,hi) => knob.map( z => linlin(z.min(hi).max(lo), lo, hi, 0.0, 1.0) )      }      }
    result.foreach(println)
    button.fire(0.0)
  }
}
This example shows how to calibrate values. Let's assume knob was some kind of device that outputs values but we don't know the range, and we would like to normalize the values to 0.0, 1.0. Then while button1 is activated we check the range of the knob and then when the button is deactivated we output the normalized values. The code above works fine, but it will only output anything after the first calibration. I would prefer that it assumed a default calibration of [0.0,1.0] but when the button is deactivated at start, we get an empty in the flatmap, and therefore no output down the line. Instead of "empty" I would like to have a EventSource with value (0.0,1.0), since EventSource has no memory, this will not work. Maybe I need to do this using Signals, and then convert to Events in the end ?
I guess maybe what I'm looking for is Events that keep the last value, such that flatMap over it would still output values even if the foreach was added after the event had fired. Maybe signals can help ?

Btw, I'm having a problem with reactive-core, maybe you can help: It appears that if I have an EventSource inside a map inside one of my classes, firing the source doesn't cause any output  at the functions registered with forech. Calling.hasListeners gives false. Why are the listeners not added ? Could this be related with the Observing implicit ? This doesn't happen when I create an event source outside any class.

scala>  val y = new EventSource[Double]y: reactive.EventSource[Double] = reactive.EventSource@16374a56
scala>  y.foreach(println)
scala>  y.hasListeners res7: Boolean = true
//x is an instance of a class defined by me and sources is a Map[BCRControl,EventSource[Double]]scala>  x.sources.get(Kn(0,0)).get.hasListenersres8: Boolean = false
scala>  x.sources.get(Kn(0,0)).get.foreach(println)
scala>  x.sources.get(Kn(0,0)).get.hasListenersres10: Boolean = false
scala>  x.sources.get(Kn(0,0)).get res11: reactive.EventSource[Double] = reactive.EventSource@4e644b0d

Hmm, that's pretty strange. Observings, like any implicit parameter, are pulled in via static scope of the call site, so it should not depend on where the event stream is instantiated or "owned." Can you show the code that defines and populates the map? (For instance, if you use 'withDefault' it will create a new event stream every time.)

I was creating the map using the "mapValues" method on another Map which actually runs the function every time get is called, so I was generating a new EventSource every time I called get. Mistery solved.
thanks !Miguel 

On Monday, January 9, 2012 10:04:08 PM UTC, Miguel Negrão wrote:

On Sunday, January 8, 2012 2:43:40 AM UTC, nafg wrote:

On Fri, Jan 6, 2012 at 12:04 PM, Miguel Negrão <miguel.ne...@friendlyvirus.org> wrote:

Yes, I really can't think of good way to define a pure/return function for EventStream, but it should be possible to do so because in Haskell the EventStreams are monads. I guess I should check how they do it in haskel. Signal on the other hand is probably a Monad with the functions it already has defined.

I can't think of any use for it, but what about an event stream that fires an event repeatedly in an infinite loop?

Indeed Signal solves the problem. Sorry for the noise:
class RichEventStream(val a: EventStream[Double]) {
  private def linlin( in: Double, srcLo: Double, srcHi: Double, dstLo: Double, dstHi: Double ) : Double = {    (in - srcLo) / (srcHi - srcLo) * (dstHi - dstLo) + dstLo  }
  def calibrate(calibrate:EventStream[Boolean]) = {
    val empty = new EventSource[Double]
    val x = calibrate      .flatMap{ x => x match {        case false => empty        case true => a      }}      .foldLeft( (0.5,0.5) ){ case ((lo,hi), x) => ( x.min(lo), x.max(hi) ) }.hold( (0.0,1.0) )
    calibrate.hold(false).flatMap{      case true => Val(0.0)      case false => x.flatMap{        case (lo,hi) => a.hold(0.5).map( z => linlin(z.min(hi).max(lo), lo, hi, 0.0, 1.0) )      }    }.change
  }}
best,Miguel

Right, Signal is what you were looking for. Glad you solved your issue!

On Tue, Jan 10, 2012 at 10:11 AM, Miguel Negrão <miguel.negrao-lists@friendlyvirus.org> wrote:

On Monday, January 9, 2012 10:04:08 PM UTC, Miguel Negrão wrote:

On Sunday, January 8, 2012 2:43:40 AM UTC, nafg wrote:

On Fri, Jan 6, 2012 at 12:04 PM, Miguel Negrão <miguel.ne...@friendlyvirus.org> wrote:

Yes, I really can't think of good way to define a pure/return function for EventStream, but it should be possible to do so because in Haskell the EventStreams are monads. I guess I should check how they do it in haskel. Signal on the other hand is probably a Monad with the functions it already has defined.

I can't think of any use for it, but what about an event stream that fires an event repeatedly in an infinite loop?

Indeed Signal solves the problem. Sorry for the noise:
class RichEventStream(val a: EventStream[Double]) {
  private def linlin( in: Double, srcLo: Double, srcHi: Double, dstLo: Double, dstHi: Double ) : Double = {     (in - srcLo) / (srcHi - srcLo) * (dstHi - dstLo) + dstLo  }
  def calibrate(calibrate:EventStream[Boolean]) = {
    val empty = new EventSource[Double]
    val x = calibrate      .flatMap{ x => x match {        case false => empty        case true => a      }}      .foldLeft( (0.5,0.5) ){ case ((lo,hi), x) => ( x.min(lo), x.max(hi) ) }.hold( (0.0,1.0) )
    calibrate.hold(false).flatMap{      case true => Val(0.0)      case false => x.flatMap{        case (lo,hi) => a.hold(0.5).map( z => linlin(z.min(hi).max(lo), lo, hi, 0.0, 1.0) )       }    }.change
  }}
best,Miguel