FS2로 구현하는 debounce 함수

Calico는 cats effect와 fs2를 사용하여 UI를 구현한 라이브러리입니다. Stream 개념으로 UI를 사용한다는 것이 재밌어보여 써보고 있는데 아직 fs2 개념이 익숙하지 않고 모르는 함수들이 많아 어렵지만 재밌기는 합니다.

Input 필드에서 사용자가 글자를 입력하다 멈추고 있으면 어떤 동작을 하게 만들고 싶어 클로드에 물어보았습니다. 이런 함수를 debounce라고 한다고 하네요. 다음은 클로드가 알려준 코드를 약간 수정한 것입니다.

def debounce[A](duration: FiniteDuration): Pipe[IO, A, A] = 
        (stream: Stream[IO, A]) => 
            Stream.eval(SignallingRef[IO, Option[A]](None)).flatMap: signal =>
                val producer = stream.evalMap(a => signal.set(Some(a))).drain
                val consumer = signal.discrete.switchMap:
                    case Some(a) =>
                        Stream(a).delayBy[IO](duration)
                    case None =>
                        Stream.empty
                producer.mergeHaltBoth(consumer)

기간 duration을 인자로 주면 Stream[IO, A]를 입력으로 받고 역시 Stream[IO, A]를 반환하는, 즉 fs.Pipe[IO, A, A]를 만드는 함수입니다. 한참을 들여다보다 다시 클로드에게 소스를 분석해 달라고 물어보니 설명까지 해 주네요.

1. 2번째 줄에서 SignallingRef형 변수 signal을 만들면서 시작합니다. 이때 eval까지의 형은 Stream[IO, SignallingRef[IO, Option[A]]]가 됩니다. flatMap 함수를 사용하니 내부에서 stream 값을 만들것임을 알수 있습니다.
2. producer는 입력으로 받은 stream변수에 값이 들어오면 signal에 그 값을 설정하는 간단한 Stream[IO, Nothing]형 변수입니다.
3. consumer는 먼저 Signal[IO, A]형 변수 signal에 discrete함수를 호출하여 Stream[IO, Option[A]]형으로 변환한 다음 이를 switchMap 함수를 통해 값이 주어지는 경우(case Some(a)) duration 만큼 지난 다음 값을 만들어내고 stream이 끝나는 경우 역시 빈 stream을 반환합니다. 그 다음줄에서 mergeHaltBoth 함수를 사용하여 producer와 consumer를 동시에 실행합니다. 인자로 주어진 두 stream중 어느 하나라도 종료되면 나머지도 종료됩니다. 이상으로 위의 flatMap 함수내에 주어질 stream이 완성되었습니다.

결론적으로는 producer는 들어온 값을 signal 변수값으로 설정하고 consumer는 signal의 stream에 들어온 값을 늦게 반환하는 역할을 하며 두 stream을 합쳤으나 producer는 drain을 사용했으므로 값을 생성하지 못하며 consumer에서만 stream 값이 만들어 집니다. 각각 독립된 일을 하는 stream이 signal 변수를 통해 서로 합쳐져 원하는 기능이 완성되었습니다.

값을 늦게 생성하되 시간내에 새로운 값이 들어오면 기존의 값을 취소시키는, debounce 동작에 핵심함수는 switchMap이라 할수 있겠네요. API 문서에 의하면 새로운 값이 들어오면 내부의 stream을 정지시킨다고 되어 있습니다 (Like Stream.flatMap but interrupts the inner stream when new elements arrive in the outer stream). 즉, 값이 주어지면 내부에 늦게 값을 생성하는 stream을 만들었으나 시간이 지나기 전에 새로운 값이 들어오면 기존의 stream은 중단되고 새로운 stream이 만들어지게 됩니다.

위에서 만든 함수는 아래와 같이 calico에서 사용할 수 있습니다.

input.withSelf: self =>
(
    placeholder := "내용 입력",
    onInput --> (debounce(2.seconds)).andThen(_.evalTap(_ => self.value.get.flatMap(IO.println)).drain)
)

input 칸에 쭉 입력하다 멈추면 2초후에 콘솔에 내용이 출력됩니다.