Introducing Bacon.js

In my previous posting I ranted a bit about the WTFs related to RxJs hot and cold Observables and didn't come up with a very good workaround.

This time I have one. Use Bacon.js. That would be my homemade reactive library for Javascript. It solves the hot/cold problem by replacing Observable with two distinct abstractions:

• EventStream: an observable stream of events. Never emits the same event twice. Always emits the same event to all subscribers.
• Property: like a stream with current value. Any new subscriber will immediately be called back with the current value if there is any.

I'll use my earlier game development examples again, converted into bacon.js, to show how the new library works. I assume some RxJs knowhow.

So if you have some events (say keyboard events) and you want to map them into some domain events (like game character movement), you can start with the source events and use map, filter and merge.
The following functions allow you to create a keyUp or keyDown stream for a given keyCode.

var allKeyUps = $(document).asEventStream("keyup")
var allKeyDowns = $(document).asEventStream("keydown")

function always(value) { return function(_) { return value } }
function keyCodeIs(keyCode) { return function(event) { return event.keyCode == keyCode} }
function keyUps(keyCode) { return allKeyUps.filter(keyCodeIs(keyCode)) }
function keyDowns(keyCode) { return allKeyDowns.filter(keyCodeIs(keyCode)) }

When you want to accumulate some state (like player position), you can use toProperty or scan to get a stateful Property. These things have a meaningful "current value", unlike streams like "keyup events".

The following creates keyState property from keyups and keydowns of each arrow key, then combines them into a single "direction" property usin toProperty:

function concat(a1, a2) {
  return a1.concat(a2)
}
function keyState(keyCode, value) {
  return keyDowns(keyCode).map(always([value])).
    merge(keyUps(keyCode).map(always([]))).toProperty([])
}
var direction = keyState(38, new Vector2D(0, -1))
  .combine(keyState(40, new Vector2D(0, 1)), concat)
  .combine(keyState(37, new Vector2D(-1, 0)), concat)
  .combine(keyState(39, new Vector2D(1, 0)), concat)
  .map(head)

The direction property will start with the empty array []. When you press an arrow key, the value will change to an array containing the corresponding direction vector. The combine method of Property is used to combine multiple properties into one.

The Property.sample method can be used to sample the current value of a Property each 50 milliseconds. The result will be an EventStream, because these are now distincts events (no current value, see?). So if you sample the direction each 50 milliseconds, you'll get a "movements" stream:

var startPos = new Vector2D(50, 50)

function head(array) { return array[0] }
function id(x) { return x }

var movements = direction.sample(50).filter(id)

Now we can further accumulate the movements into the current position:

var position = movements
  .scan(startPos, function(pos, move) { return pos.add(move) })

And voila, we are able to control our game character with the keyboard.

For a live demo of the same, see my slideshow, especially the last slide.

Bacon.js has quite good test coverage, it's open-source, it has at least some level of documentation and I'm a nice guy, so please try it out and let me know how the Bacon suits you.

All help is also appreciated also, so please contribute.

Things You Should Know About RX

As I've said before, RX for Javascript is a great library for event-rich Javascript apps. But as with many tools of great power, it allows you to do stupid things and get hurt. In this posting I'm focusing on those features that will probably provide you with the most WTFs.

Some examples

A counter that increases or decreases its value based on clicks to + and - buttons can be implemented like this:

  var incr = $('#incr').toObservable('click').Select(always(1))
  var decr = $('#decr').toObservable('click').Select(always(-1))
  var series = incr.Merge(decr)
    .Scan(0, function(total, x) { return total + x })

Is good? Well, as long as you subscribe at the right time. If you subscribe later, the later subscriber will get different results from the first one, because it's counter will start from 0 when it subscribes.

You could derive a position stream from a movements stream as follows:

  var position = movements
    .Scan(startPos, function(pos, move) { return pos.add(move) })
    .StartWith(startPos)

Is good? Well, the exact same problem. Position stream delivers different results depending when you subscribe.

Both examples seem perfectly ok, unless you know the details of RX. Let's take a walk.

Misconception #1 : Observable Is a Stream of Events

Well, I have said the above sentence countless times, regardless of how wrong it is. Actually Rx would make a lot more sense (with less WTFs) if that were true. A real stream, in my opinion, would provide all of its Subscribers with the same series of events after any moment of time t. Unfortunately with RX, the events that you receive after t depends on when you subscribed.

Why?

Hot and Cold Observables

In RX, there are "hot" and "cold" observables. As Bnaya Eshet described in his blog posting:

If a tree falls in a forest and no one is around to hear it, does it make a sound? if it do make a sound when nobody observed it, we should mark it as hot, otherwise it should be marked as cold.

So what? Well, simply put, hot observables are consistent among subscribers and cold ones are not. A hot observable can be seen as a list of (time, value) pairs. A hot observable won't produce the an event to subscribers that subscribed after the event occurred. The cold ones are, well, something that depends on when you subscribe.

All Observables that are directly derived from some real-world events are hot. For instance, mouse and keyboard events will be consistent among all subscribers. But, if you combine them with a cold observable, or apply a stateful combinator, such as Scan, you're not so safe anymore.

Observables from Arrays

Observable.FromArray serves you a cold Observable. It always spits out the same list of objects when you subscribe. Doesn't provide a consistent mapping of time to events. Period.

Mixing Hot and Cold

If you mix hot with cold, what do you get? Medium? So, if you

var hotness = $(document).toObservable("keyup")
var temperature = Observable.FromArray("coldness").Concat(hotness.Select(always("hotness")))

You might expect to get an observable starting with "coldness" and producing "hotness" at each keyup. However, StartWith made your new observable a bit colder in the sense that any new subscriber will always get "coldness" first.

No matter how you combine coldness with hotness, you won't get a hot Observable back. It won't be cold as in "tree falling in the woods", but inconsistent anyway. Lukewarm? No, more like Groundhog Day.

Stateful Streams Using Scan

As I showed in the counter-example (pun intended) above, Scan will give you a Groundhog Day Observable.

StartWith

Using StartWith won't save you. It's the same thing as concatenating with a cold stream of one event.

Workaround : Publish/Connect

This is a workaround I already presented in my hot-tempered Failing with RX-JS posting. You can make your Observable hot again by using Publish and Connect. For instance, you can fix the broken counter like this:

  var series = incr.Merge(decr)
    .Scan(0, function(total, x) { return total + x })
    .Publish()
  var dispose = series.Connect()

So first you call Publish on your Observable. That'll give you an Observable, that will have a single connection to the underlying Observable, no matter how many subscribers you add. Then you call Connect, which will start it, by calling the Subscribe method of the underlying stream. It will also give you a back a "dispose" function for disconnecting.

The obvious drawback of this method is that there's more clutter. Also, the connection to the underlying stream won't be automatically disconnected. You have to call dispose yourself.

Shake your Rump!

This time it's not about Rx for Javascript or even Javascript. Well, that's at least partly because I got hooked on Haskell and have spent most of my free time (not much though) on Haskell projects.

But lately, I've been working on a mobile (Android/iOS) app called Karma. The app will answer the ever-crucial question "who's turn is it to buy the next round?". To make it fun and easy to use, it needed an easy way of telling the system that two guys, say Bob and Mike, have just met for some beers. Thanks to location and acceleration info, it's possible to make this pairing happen just by shaking phones together.

The topic of this post is the "shake to pair" technology included. We started by trying the Bump API. However, we found it quite unreliable, even though the actual Bump application seemed solid. I guess they don't give their best performance for API users?

Now, what would a proper nerd do? What the heck, let's write it from scratch. And POW, here it is:

• RUMP Server
• RUMP Android Client

As you see, it's open-source. And, it's actually very simple:

1. When user shakes the phone, the RUMP Client sends a message to the server. Like this:

      { "userId" : "john", "displayName" : "John Kennedy", "location": { "latitude": 51.0, "longitude": -0.1 } }

1. Server matches incoming requests by time (3 second time window to match) and location (1 kilometer max distance).
2. After 3 seconds from the first received request, the server sends match info to all clients that were matched with the first one. The response is just an array of all the matching requests.
3. Rump Client delivers the match info to the client application.
4. Client application does something cool. For instance, the Karma application joins the matched users into the same table and calculates the next buyer.

Well, that's it. On the Github pages, you'll find instructions for building, running and integrating the Android Client into your app.

I think this would provide a nice basis for matching users in a multiplayer game, for instance. Shake your phones to start multi-player, anyone?

What else? We have an iPhone version of Karma in the works. Mayhaps we'll also extract the Rump client part from there and open-source that too? Would it make sense?

The server-side piece is written in Haskell. I'm on a horse.

Zipping With Arrays in RxJs

Have you ever tried to zip a hot observable with a cold one? That's what happens below.

   var former = function(a, b) { return a }
   var latter = function(a, b) { return b }
   var logger = function(a) { console.log(a) }
   var numbers = Rx.Observable.FromArray([1, 2, 3])
   var timer = Rx.Observable.Interval(100)
   var zipped = numbers.Zip(timer, former)
   zipped.Subscribe(logger)

The idea is to print out the contents of an array to the console one-by-one, each 100 milliseconds. It doesn't work though. What I've heard is that it fails because of a bug in Rx. Instead of printing out numbers 1,2,3 to the console, the code does ... nothing.

A simple workaround (Thanks Jussi) seems to work, though: Just concat your array observable with Rx.Observable.Never!

   // ...
   numbers.Concat(Rx.Observable.Never()).Zip(timer, former).Subscribe(logger)

1
2
3

So, now it prints the stuff to the console as you'd expect.

To make zipping with arrays a bit easier, you could add a ZipWithArray method to the Observable prototype like this.

  Rx.Observable.prototype.ZipWithArray = function(array, selector) {·
      return this.Zip(Rx.Observable.FromArray(array).Concat(Rx.Observable.Never()), 
  }

I've put this extension, as well a bunch of others on Github, in my rxjs-extensions project.