Reactive Extension part4

Reactive Extension part4

 

토요일 오후 커피숍에서 아이스 아메리카노를 빨며 한가롭게..딩굴딩굴..하고 싶었지만, 역시나 마음은 Rx 포스트를 언능 완료하고, ReactiveUI 포스트까지 쓰고, 실무에 적용을 해야하니..쿨럭..배보다 배꼽이 더 커진듯한 느낌은 이전이나 지금이나 변함이 없다.

 

part3에서 사용했던 소스에 추가로 코딩을 하기 때문에 이전 포스트에있는 소스를 받아서 사용하도록 한다.

 

"벌써 part4네 이제 포스트 쓰는 속도가 좀 빨라진 것 같네?"

 

"포스트 쓰는 페이스가 정상으로 회복됐다."

 

"오홍..그런데 페이스는 정상인데 페이스북 반응은 냉냉하던데?"

 

"좋아요랑 코맨드도 달려있다."

 

"누가 한건데?"

 

"나"

 

 

1. More query operators to tame the user’s input

 

            var inputSubscription = input.Subscribe(
                data =>
                {
                    Console.WriteLine(data);
                    lbl2.Text = data;
                });

 

input은 TextBox의 TextChanged 이벤트를 옵저버블(observable) 한 것으로 Text가 변경될 때 마다 Text를 push하고, 위의 Subscribe은 lbl2.Text에 변경된 텍스트를 출력하는 곳이다. 굵은 글씨 부분을 추가해서 각 단계마다 입력되는 Text를 확인해보자

 

 

Console에 출력된 내용은 위와 같다. 여기서 맨 아래 2줄은 동일한 결과를 표시하고 있는데, 그 이유는 reactive라는 글씨를 입력 한 후 커서를 t의 위치로 이동한 후 t를 선택하고, 다시 t를 입력했기 때문에 동일한 내용이지만 TextChanged 이벤트가 발생하여, 동일한 글씨가 출력된 것이다.

 

TextChanged 이벤트는 Text에 기초해서 발생하는 것이 아니라, 키보드 입력 상태를 체크해서 발생하는 이벤트이므로 동일한 Text라 하더라도 다시 발생하게 된다.

 

만약 Text를 web service의 검색어 조건으로 사용 한다면, 아마 동일한 Text를 2번 호출하는 경우가 발생할 것이다. 물론, 이전 입력값과 동일한지 아닌지를 이벤트 핸들러에서 비교해서 같지 않을 경우에만 호출하는 로직을 추가해 넣을 수 도 있지만, 이렇게 처리를 하더라도 비동기 이벤트 핸들러의 타이밍이 어긋나서 원하는 기능을 수행하지 못하는 경우가 발생할 수 있다.

 

 

"정말 그런 경우가 발생한다고?"

 

"가능하다고 생각한다."

 

"어떻게?"

 

"컴퓨터의 반응속도보다 빠르게 지우고, 쓰고를 반복하다 보면.."

 

(ㅡㅡ)a

 

 

Rx에서는 값의 변화를 추적하여, 이전 값과 서로 다른 경우에만 처리를 하도록 하는 강력한 기능이 존재한다.

 

            var input = (from evtPattern in Observable.FromEventPattern<EventArgs>(txt, "TextChanged")
                        select ((TextBox) evtPattern.Sender).Text)
                        .DistinctUntilChanged();

 

위에서 사용한 DistinctUntilChanged()가 바로 그 주인공으로, 동작 원리를 이해하기 위해서는 데이터의 흐름을 파악하는 것이 중요한데, 아래 그림과 설명을 참고한다.

 

* pdf에서는 FromEvent를 사용했지만 우리는 FromEventPattern을 사용하고 있다.

 

 

 

TextBox에 사용자가 내용을 입력하면, TextChanged 이벤트가 발생하며, 그 이벤트를 받은 Observer인 LINQ Select에서 TextBox.Text를 string으로 투영화(projection)해서 다음으로 전달하고, DistinctUntilChanged에서 이전 데이터와 같은지 다른지를 판단하여, 다른 경우에만 Subscribe에서 데이터를 전달하게 되는 것이다.

 

그리고, 아마 개발자라면 이런 일련의 과정들이 정말 일어나는지 아닌지 확인 하기를 원할 것이다. 그래서, Rx에서는 Do라는 오퍼레이터를 제공하고 있으며, 이 것을 사용해서 Log를 출력할 수 있다.

 

 

 

            var input = (from evtPattern in Observable.FromEventPattern<EventArgs>(txt, "TextChanged")
                        select ((TextBox) evtPattern.Sender).Text)
                        .Do(projectionText => Console.WriteLine("Before DistinctUntilChanged : {0}", projectionText))
                        .DistinctUntilChanged();

 

위와 같이 소스를 수정 후 실행한 결과이다.

 

 

아래에서 3번째 줄을 보면 reactive라는 글씨가 찍힌 것을 볼 수 있다. 하지만 그 내용이 바로 아랫줄에 출력이 되지 않고, 다음에 발생한 reactiv라는 글씨만 출력된 것을 볼 수 있다. 만약, Debug.WriteLine 같은 코드를 찍기 위해서는 이 Do 오퍼레이터를 활용하면 된다.

 

 

이제 다시 원 위치로 돌아가서 소스를 다시 살펴보자. 우리가 입력한 값을 web service를 사용해서 조회를 한다고 했을 때, 사용자의 모든 입력을 web service로 전송해야 하나? 즉, 조회를 하려는 단어는 reactive인데, 그 중간 단어들도 모두 호출을 한다면, 그 것 또한 문제가 아닐 수 없다. 그래서, 일반적으로는 Search 버튼을 두어 버튼을 누르거나, Enter 키를 누를 때 web service를 호출하도록 코딩을 한다.

 

 

"결과만 나오면 돼는 것 아닌가?"

 

"옳치않다."

 

"난 그냥 쓰는게 편하던데??"

 

"용납할 수 없다!"

 

"오~ 강하게 나오는데? 나한테 그렇게 나오다니... 어디 아프구나?"

 

"오전에 사과즙 먹은것이..쿨럭"

 

 

Rx에서는 observable sequence를 지연시키는 Throttle이라는 오퍼레이터를 가지고 있어서, 이런 문제를 쉽게 해결할 수 있다.

 

            var input = (from evtPattern in Observable.FromEventPattern<EventArgs>(txt, "TextChanged")
                        select ((TextBox) evtPattern.Sender).Text)
                        .Throttle(TimeSpan.FromSeconds(1))
                        .Do(projectionText => Console.WriteLine("Before DistinctUntilChanged : {0}", projectionText))
                        .DistinctUntilChanged();

 

위에 소스로 수정 후 실행을 해보자.

 

            var inputSubscription = input.Subscribe(
                data =>
                {
                    Console.WriteLine(data);
                    lbl2.Text = data;
                });

 

An unhandled exception of type 'System.InvalidOperationException' occurred in System.Windows.Forms.dll

Additional information: Cross-thread operation not valid: Control '' accessed from a thread other than the thread it was created on.

 

위의 굵은 글씨 위치에서 오류가 발생하는데, 이 Cross-thread operation not valid 오류에 대해서는 나중에 다시 설명하고 처리하도록 하고, 우선은 주석 처리를 한 후에 다시 실행한다.

 

            var inputSubscription = input.Subscribe(
                data =>
                {
                    Console.WriteLine(data);
                    //lbl2.Text = data;
                });

 

실행 후 'react'까지 빠르게 입력 후 Console에 'react'가 출력이 되면, 나머지 'ive'를 입력해 준다. 아마도 아래와 같은 결과가 나올 것이다.

 

 

Throttle()을 추가하기 전과 상당히 다른 결과가 출력되었다는 것을 알 수 있다.

 

Throttle는 사용자가 지정한 시간동안 push-notification이 발생하지 않으면, 그 결과를 다음으로 넘기는데, 중간에 한번이라도 발생하면, 타이머가 초기화 되어 처음부터 다시 카운트를 하게된다. 결과적으로, 사용자가 TextChanged 이벤트를 1초동안 발생시키지 않았을 경우에만 TextBox에 입력된 Text를 DistinctUntilChanged 오퍼레이터에서 전달하게 되는 것이다.

 

이 과정을 좀더 자세히 확인하기 위해서는 pdf에 있는 것 처럼 Timestamp(), RemoveTimestamp()를 이용해서 타이머를 데이터를 출력하면 되는데, 여기서는 다루지 않겠다.

 

 

지금까지 작업에 대해서 정리를 해보자.

 

TextBox의 TextChanged 이벤트를 이용해서, 사용자 입력에서 중복 방지 처리를 하고, 입력 지연 기능 또한 구현해 보았다. 위의 두가지 기능을 추가하기 위해서 이벤트 코딩이였다면 상당한 양의 코딩을 해야 했을 것이고, 정확한 동작이 되는지를 오랫동안 테스트를 해보아야 했을 것이다. 하지만, Rx에서는 단 두개의 오퍼레이터를 사용하는 것으로 만족스러운 결과를 만들 수 있는 것이다.

 

이제 Rx가 약간 좋아지려고 하나?? 하지만, 꼭 기억하고 있어야 하는 것이 있는데. 이 모든 처리가 가능한 근본적인 이유는 이벤트를 IObservable<T> object로 처리를 하면서, 발생 순서대로 push-notification을 해주기 때문에 가능하다는 것이다.

 

 

2. Rx’s concurrency model and synchronization support

 

이 곳에서 이야기할 내용이 Thread와 관련된 내용이기 때문에 잘 모르는 분을 위해 짦게 설명을 하고 넘어 가겠다.

 

어플리케이션에는 크게 2가지의 스레드가 존재하며, 각각을 Main Thread(UI Thread), Worker Thread(Background Thread)라고 이야기 한다. Main Thread는 화면에 데이터를 출력하는 것을 관리하기 때문에 화면에 보이는 내용들은 모두 Main Thread에 소속되어 있다. 반면, 시간이 오래 걸리는 작업이나 화면에 출력할 필요가 없는 작업들은 Worker Thread에서 처리를 하게된다. 이렇게 작업 영역을 서로 나누어 놓은 것은 1개의 Thread만 사용하게 되면, 화면 멈춤 현상이 발생하기 때문이다. 과거에 Visual Basic 6.0이 이와 같이 1개의 Thread만 사용한 프로그래밍 언어였다.

 

이벤트는 기본적으로 Main Thread에서 발생한다. 그렇기 때문에 이벤트 핸들러 내부에서 처리되는 모든 내용도 기본적으로는 Main Thread에서 처리가 되도록 되어있다. (하지만, .NET Framework 4.5 이상이고, 이벤트 핸들러에 async를 사용 했다면, 이 이벤트 핸들러는 Worker Thread에서 실행될 것이다.)

 

* Microsoft 김명신 부장님의 제보로 알게된 사항으로 async, await를 사용하더라도 이벤트 핸들러는 모두 Main Thread에서 동작하는 것을 확인하였다.

 

 

 

 

 

 

하지만, 바로 전 소스에서 Throttle() 오퍼레이터를 사용하자마자, Cross-thread operation not valid 오류를 만나게 되었다. 이 오류가 발생한 이유는 지정한 시간동안 실행을 막아야하는데, 이 작업을 Main Thread에서 실행하면, 화면이 멈추어 버리기 때문에, Worker Thread로 처리 하도록 Rx가 변경을 했기 때문이다.

 

 

위의 위치에 Breakpoint를 걸고, 실행을 시켜본 결과이다. Threads 창으로 보면 제일 아래 노란색 화살표가 있는 부분이 현재 실행 중인 Thread인데, Category가 Worker Thread로 표시 되는 것을 알 수 있다. 이렇게 Worker Thread에서 실행 중에 화면 UI에 어떤 내용을 출력하려고 하면 발생하는 오류가 바로 Cross-thread operation not valid 오류이며, 이 것을 가르켜 동시성(concurrency) 문제라고 한다.

 

Rx에서는 이런 동시성 문제를 Scheduler와 동기화 처리 오퍼레이터를 사용하여 해결 할 수 있는데, 이 부분에 대해서 살펴 보도록 하겠다.

 

여기서 알아볼 동기화 처리 오퍼레이터는 ObserveOn() 이라는 것으로, 이 것을 사용하면 Worker Therad의 context와 Main Thread의 context를 동기화 시킬 수 있으며, IScheduler 인터페이스를 반드시 파라메터로 넘겨 주어야 한다.

 

            var inputSubscription = input
                    .ObserveOn(SynchronizationContext.Current)
                    .Subscribe(
                    data =>
                    {
                        Console.WriteLine(data);
                        lbl2.Text = data;
                    });

 

위의 굵은 글씨 부분을 추가하고, 아래 주석을 푼 후에 다시 실행해 보자.

 

 

위에서 보듯이 이제 노란 화살표는 Main Thread에 위치해 있는 것을 볼 수 있다. ObserveOn은 파라메터로 받은 IScheduler와 동기화를 시켜주는데, WinForm일 때와 WPF, Silverlight, WinRT일때 IScheduler를 사용하는 방법에 차이가 있다는 것을 알고 있어야 할 것이다.

 

* pdf에서 사용하는 control자체를 넘기는 방식은 더이상 지원하지 않고 있으며, 검색으로 찾아본 해결 방법 중에서ControlScheduler도 object browser에서 찾을 수 가 없다. 혹시라도 다른 방법을 발견하면 수정하도록 하겠다.

 

ObserOn과 비슷한 동기화 연산자로 SubscribeOn이 존재하는데, 이는 IScheduler와 Subscribe간의 동기화를 하기 위해 사용된다.

 

이제 최종 결과를 확인해 보자.

 

 

Console의 내용과 WinForm상에 출력되는 내용도 모두 정상적으로 출력이 된다.

 

 

3. End

이번 포스트는 여기서 마무리 하고 다음에는 위의 소스를 이용해서 진짜 web service를 호출하는 포스트를 올리도록 하겠다. 그러니, 버리지 말고 잘 간직하자!

 

일요일 저녁에 조용히 ReactiveUI 코딩을 만지작 만지작 하려고 했는데, 이전 포스트에 진심이 묻어나는 멋진 리플을 발견하여 의욕이 1000% 급상승하여 부랴부랴 마무리를 하게 되었다. 좋은 리플한개는 고래도 춤추게 한다는 이야기는 전설이 아니였었다. 하하하

 

 

"배꼽 큰 놈에서 고래로 변신이냐?"

 

"불만인가?"

 

"의욕과 전투력은 비례하지 않을텐데??"

 

"으흠...그렇군..잠시.."

 

 

4. 소스

 

RxSample.part4.zip

다운로드