Running histogram stream with Rx

Question

I have the following single letter stream

A 
B 
C 
A
D 
B 
A 
C 
D

And from this stream, I would like a stream of running count per letter

(A,1)
(A,1), (B,1)
(A,1), (B,1), (C,1)
(A,2), (B,1), (C,1)
(A,2), (B,1), (C,1), (D,1)
(A,2), (B,2), (C,1), (D,1)
(A,3), (B,2), (C,1), (D,1)     
(A,3), (B,2), (C,2), (D,1) 
(A,3), (B,2), (C,2), (D,2)     

, i.e at each new letter, the totals are updated and emitted.

I guess this problem is fairly language agnostic, so don't hesitate to propose a solution in your language of choice.

Answer 1

This is how it can be done using RxJava:

final Observable<String> observable = Observable.just("A", "B", "C", "A", "D", "B", "A", "C", "D");
final Observable<LinkedHashMap<String, Integer>> histogram = observable.scan(new LinkedHashMap<>(), (state, value) -> {
  if (state.containsKey(value)) {
    state.put(value, state.get(value) + 1);
  } else {
    state.put(value, 1);
  }

  return state;
});

histogram.subscribe(state -> {
  System.out.println(state);
});

Output:

{}
{A=1}
{A=1, B=1}
{A=1, B=1, C=1}
{A=2, B=1, C=1}
{A=2, B=1, C=1, D=1}
{A=2, B=2, C=1, D=1}
{A=3, B=2, C=1, D=1}
{A=3, B=2, C=2, D=1}
{A=3, B=2, C=2, D=2}

Answer 2

In RxJS it could be something like:

var letters   = Rx.Observable.of('A', 'B', 'C', 'A', 'D', 'B', 'A', 'C', 'D'),
    histogram = letters.scan(countL, Object.create(null));

histogram.subscribe(console.log.bind(console));

function countL(ls, l) {
    if (!ls[l]) ls[l] = 0;
    ls[l]++;
    return ls;
}