Creating an input timer in RxJS; Time tracking

Question

This question is a continuation of my previous question, which you can find here:

How to use RxJS to display the "user enters" indicator?

After successfully checking if the user was typing, I needed to use this particular state as a trigger for the clock.

The logic is simple, in fact, I want the clock to start when the user types. But when the user stops typing, I need the watch to stop. When the user starts typing again, the clock should continue to accumulate.

I already managed to get it working, but it seems like a mess, and I need help refactoring it, so this is not a spaghetti ball. This is what the code looks like:

/*** Render Functions ***/ const showTyping = () => $('.typing').text('User is typing...'); const showIdle = () => $('.typing').text(''); const updateTimer = (x) => $('.timer').text(x); /*** Program Logic ***/ const typing$ = Rx.Observable .fromEvent($('#input'), 'input') .switchMapTo(Rx.Observable .never() .startWith('TYPING') .merge(Rx.Observable.timer(1000).mapTo('IDLE'))) .do(e => e === 'TYPING' ? showTyping() : showIdle()); const timer$ = Rx.Observable .interval(1000) .withLatestFrom(typing$) .map(x => x[1] === 'TYPING' ? 1 : 0) .scan((a, b) => a + b) .do(console.log) .subscribe(updateTimer) 

And here is the link to live JSBin: http://jsbin.com/lazeyov/edit?js,console,output

Maybe I'll walk you through the logic of the code:

• First, I create a stream to capture each input event.
• For each of these events, I will use switchMap to: (a) hide the original TIPING event so that we don’t lose it, and (b) disable the IDLE event, 1 second later. You can see that I create them as separate threads and then combine them together. Thus, I get a stream that will indicate the "input state" in the input field.
• I create a second thread that sends an event every second. Using withLatestFrom , I merge this stream with the previous input stream. Now that they are combined, I can check if the input state is "IDLE" or "TYPING". If they print, I give the event a value of 1 , otherwise 0 .
• Now I have stream 1 and 0 s, all I have to do is add them back using .scan() and map it to the DOM.

What is the RxJS way to write this function?

EDIT: Method 1 - Create a Change Event Flow

Based on @osln answer.

 /*** Helper Functions ***/ const showTyping = () => $('.typing').text('User is typing...'); const showIdle = () => $('.typing').text(''); const updateTimer = (x) => $('.timer').text(x); const handleTypingStateChange = state => state === 1 ? showTyping() : showIdle(); /*** Program Logic ***/ const inputEvents$ = Rx.Observable.fromEvent($('#input'), 'input').share(); // streams to indicate when user is typing or has become idle const typing$ = inputEvents$.mapTo(1); const isIdle$ = inputEvents$.debounceTime(1000).mapTo(0); // stream to emit "typing state" change-events const typingState$ = Rx.Observable.merge(typing$, isIdle$) .distinctUntilChanged() .share(); // every second, sample from typingState$ // if user is typing, add 1, otherwise 0 const timer$ = Rx.Observable .interval(1000) .withLatestFrom(typingState$, (tick, typingState) => typingState) .scan((a, b) => a + b, 0) // subscribe to streams timer$.subscribe(updateTimer); typingState$.subscribe(handleTypingStateChange); 

JSBin Live Demo

EDIT: Method 2 - Use the outputMap command to start the counter when the user starts typing

Based on Dorus's answer.

 /*** Helper Functions ***/ const showTyping = () => $('.typing').text('User is typing...'); const showIdle = () => $('.typing').text(''); const updateTimer = (x) => $('.timer').text(x); /*** Program Logic ***/ // declare shared streams const inputEvents$ = Rx.Observable.fromEvent($('#input'), 'input').share(); const idle$ = inputEvents$.debounceTime(1000).share(); // intermediate stream for counting until idle const countUntilIdle$ = Rx.Observable .interval(1000) .startWith('start counter') // first tick required so we start watching for idle events right away .takeUntil(idle$); // build clock stream const clock$ = inputEvents$ .exhaustMap(() => countUntilIdle$) .scan((acc) => acc + 1, 0) /*** Subscribe to Streams ***/ idle$.subscribe(showIdle); inputEvents$.subscribe(showTyping); clock$.subscribe(updateTimer); 

JSBin Live Demo

Answer 1

I notice a few problems with your code. The bottom line is good, but if you use different operators, you can do the same even easier.

First you use switchMap , this is a good statement to start a new thread every time a new input arrives. However, what you really want is to continue the current timer while the user is typing. The best operator here is exhaustMap , because exhaustMap will hold an already active timer until it stops. Then we can stop the active timer if the user does not print for 1 second. This is easy to do with .takeUntil(input.debounceTime(1000)) . This will result in a very short query:

 input.exhaustMap(() => Rx.Observable.timer(1000).takeUntil(input.debounceTime(1000))) 

To this request, we can bind the displayed events that you want, showTyping , showIdle , etc. We also need to fix the index timers, as it will reset every time the user stops typing. This can be done using the second parameter of the project function in map , since this is the index of the value in the current thread.

 Rx.Observable.fromEvent($('#input'), 'input') .publish(input => input .exhaustMap(() => { showTyping(); return Rx.Observable.interval(1000) .takeUntil(input.startWith(0).debounceTime(1001)) .finally(showIdle); }) ).map((_, index) => index + 1) // zero based index, so we add one. .subscribe(updateTimer); 

Note that I used publish here, but this is not strictly necessary, as the source is hot. However, it is recommended because we use input twice, and now we do not need to think about whether it is hot or cold.

Live demo

 /*** Helper Functions ***/ const showTyping = () => $('.typing').text('User is typing...'); const showIdle = () => $('.typing').text(''); const updateTimer = (x) => $('.timer').text(x); /*** Program Logic ***/ Rx.Observable.fromEvent($('#input'), 'input') .publish(input => input .exhaustMap(() => { showTyping(); return Rx.Observable.interval(1000) .takeUntil(input.startWith(0).debounceTime(1001)) .finally(showIdle); }) ).map((_, index) => index + 1) // zero based index, so we add one. .subscribe(updateTimer); 

 <head> <script src="https://code.jquery.com/jquery-3.1.0.js"></script> <script src="https://unpkg.com/@reactivex/ rxjs@5.0.0-beta.12 /dist/global/Rx.js"></script> </head> <body> <div> <div>Seconds spent typing: <span class="timer">0</span></div> <input type="text" id="input"> <div class="typing"></div> </div> </body> 

Answer 2

If you want to constantly update the user interface, I don’t think there is any way to use the timer - I could write the stream a little differently by triggering the timer using change-events - but your current stream also looks good like this already there is:

 const inputEvents$ = Rx.Observable .fromEvent($('#input'), 'input'); const typing$ = Rx.Observable.merge( inputEvents$.mapTo('TYPING'), inputEvents$.debounceTime(1000).mapTo('IDLE') ) .distinctUntilChanged() .do(e => e === 'TYPING' ? showTyping() : showIdle()) .publishReplay(1) .refCount(); const isTyping$ = typing$ .map(e => e === "TYPING"); const timer$ = isTyping$ .switchMap(isTyping => isTyping ? Rx.Observable.interval(100) : Rx.Observable.never()) .scan(totalMs => (totalMs + 100), 0) .subscribe(updateTimer); 

Live here .

---

If you do not need to update the user interface and just want to record the duration of the input, you can use start and stop events and match them with timestamps like these, for example:

 const isTyping$ = typing$ .map(e => e === "TYPING"); const exactTimer$ = isTyping$ .map(() => +new Date()) .bufferCount(2) .map((times) => times[1] - times[0]) .do(updateTimer) .do(typedTime => console.log("User typed " + typedTime + "ms")) .subscribe(); 

Live here .