title: RXjs : How to create an operator on streams : scan operator where the accumulator state can be reset through an observable
body: <p>I need to create a new instance operator on streams with the following characteristics</p>
  
  <ul>
  <li><p><strong>Signature</strong>
  <code>
  Rx.Observable.prototype.scan_with_reset(accumulator, seed$)
  </code></p>
  
  <p>where :</p>
  
  <p><strong>Arguments</strong></p>
  
  <p><code>accumulator (Function)</code>: An accumulator function to be invoked on each     element.</p>
  
  <p><code>seed$ (Observable)</code> : An observable whose values will be used to restart the accumulator function. The accumulator function has the following signature <code>function accumulator_fn(accumulator_state, source_value)</code>. I want the value in <code>seed$</code> to reset <code>accumulator_state</code> to the <code>seed</code> value and emit the <code>seed</code> value.</p>
  
  <p><strong>Returns</strong>
  <em>(Observable)</em> : An observable sequence which results from the comonadic bind operation (whatever that means, I am copying <code>Rxjs</code> documentation here). Vs. the normal <code>scan</code> operator, what happens here is that when the accumulator function is 'restarted' from the seed value emitted by the <code>seed$</code> observable, that seed value is emitted, and the next value to be emitted by the <code>scan_with_reset</code> operator will be <code>accumulator_fn(seed, source_value)</code></p></li>
  <li><p><strong>Example of use :</strong></p>
  
  <p><code>var seed$ = Rx.Observable.fromEvent(document, 'keydown')
                           .map(function(ev){return ev.keyCode})
                           .startWith(0);
   var result$ = counter$.scan_with_reset(seed$, 
                            function accumulator_fn (acc, counter) {return acc+counter});</code></p>
  
  <p>The following diagrams should explain more in details the expected results:
  <code>
  seed              : 0---------13--------27------------
  counter           : -1--5--2----6---2-----4---1---3---
  result            : 0-1--6--8-13-19--21-27-31--32--35-</code></p></li>
  </ul>
  
  <p>My initial attempt to do this was to modify the <code>accumulator_fn</code> to have the <code>seed$</code> modify a variable that would in the scope of <code>accumulator_fn</code> so I can detect changes in the function itself.</p>
  
  <p>I pursue two goals here:</p>
  
  <ul>
  <li>have an implementation which is as stateless and closure-less as possible</li>
  <li>understand the mechanics behind defining one's own operators on
  streams, of which this would be hopefully a simple example</li>
  </ul>
  
  <p>I had a look at <code>scan</code> source code : <a href="https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/scan.js" rel="nofollow">https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/scan.js</a>
  but I am not sure where to go from there.</p>
  
  <p>Does anybody has any experience in creating Rxjs stream operators? What are the conventions to follow and traps to avoid? Are there any examples of custom-made operators that I could look at? How would you go about implementing this particular one?</p>
  
  <p>[UPDATE] : Some test code for the accepted answer</p>
  
  <pre><code> var seed$ = Rx.Observable.fromEvent(document, 'keydown')
                            .map(function(ev){return ev.keyCode})
                            .startWith(0);
   var counter$ = Rx.Observable.fromEvent(document, 'mousemove')
                               .map(function(ev){return 1});
   var result$ = counter$.scanWithReset(seed$,
                                        function accumulator_fn (acc, counter) {return acc+counter});
   var s = function (x) {console.log("value: ", x)};
   var disposable = result$.subscribe(s)
  </code></pre>
  
  <p>Moving the mouse should show a value increase by 1, and pressing a key should restart the counter with the value of the key pressed.</p>