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.

A Nice Java HashMap Antipattern

Don't you just love how simple it is to instantiate maps in Java? Like, could it be any uglier than

Map<String, String> map = new HashMap<String, String>();
map.put("fruit", "banana");
map.put("vegetable", "carrot");

I saw someone using a pattern that made it look a bit nicer, like

Map<String, String> map = new HashMap<String, String>() {
  put("fruit", "banana");
  put("vegetable", "carrot");
}

It kinda puts a bit more structure to this mess. However, you'll get some unwanted runtime side-effects.

When you use this "syntax", you'll actually create a new anonymous inner class. Each anonymous inner class instace will have a runtime reference to thethis object for the context where you define the anonymous inner class in. Thus, if your this happens to have references to 100 megabytes of porn, that stuff will not be GC'd as long as this Map or yours is alive. Also, should you try to serialize your map for sending or storing it somewhere, you'll also be sending your 100 megabytes of porn.

Enjoy your cup of Java!