bluetape4k Flow extensions: 복잡한 stream 조립을 이름 있는 운영 계약으로 바꾸기

Flow operator는 짧은 코드가 목적이 아니라, stream의 운영 계약을 읽히게 만드는 이름입니다.

2026-06-30 · bluetape4k-workshop · Kotlin Flow extensions

Kotlin Flow 코드는 처음에는 단순합니다. MutableSharedFlow 하나 만들고, launch로 모으고, 필요하면 Job을 취소합니다. 그런데 검색 자동완성, fallback source, callback bridge, metric sampling처럼 조금만 현실적인 요구가 들어오면 코드가 금방 “언제 누가 취소하는지”, “마지막 값만 필요한지”, “실패를 값으로 다룰지 예외로 다룰지”를 숨기기 시작합니다.

bluetape4k-workshop의 Flow extensions 예제는 이 숨은 결정을 operator 이름으로 끌어올립니다. 이번 글은 여섯 예제를 하나씩 소개하되, API 목록이 아니라 어떤 문제를 어떤 계약으로 바꾸는지에 초점을 둡니다.

예제	다루는 문제	핵심 Flow extensions
Search Pipeline	빠른 입력, 설정 snapshot, 최신 검색만 유지	bufferingDebounce, withLatestFrom, flatMapLatest, takeUntil
Race and Fallback	가장 빠른 정상 source, 순차 fallback, partial merge	race, amb, concat, concatArrayEager, concatMapEager, merge
Subject Bridge	callback을 event/state/history/work queue stream으로 변환	PublishSubject, BehaviorSubject, ReplaySubject, MulticastSubject, UnicastWorkSubject
Event Aggregation	주문 event replay, window, grouping, 누적 상태	chunked, windowed, groupBy, scanWith, bufferUntilChanged, zipWithNext
Parallel Enrichment	주문별 독립 enrichment를 parallel rail로 처리	parallel, sequential
Metrics Sampling	burst metric에서 빠른 preview와 안정된 dashboard 값을 분리	throttleLeading, throttleTrailing, pairwise, takeUntil, mapResultCatching

ReactiveX operator 문서는 buffer, groupBy, scan, window를 서로 다른 변환 operator로 분류하고, Reactor의 Three Sorts of Batching 문서는 grouping, windowing, buffering을 같이 비교합니다. 다만 출력 형태는 서로 다릅니다. scan 계열은 batching이 아니라 이전 상태와 현재 event로 다음 상태를 만드는 accumulator 쪽으로 봐야 합니다.

아래 그림들은 RxJava 문서와 튜토리얼에서 자주 쓰는 marble diagram 문법을 따릅니다. 기본적으로 위쪽 timeline은 입력 Flow, 가운데 box는 operator, 아래 timeline은 출력 Flow입니다. 병렬 rail처럼 timeline 하나로 설명하기 어려운 예제는 같은 visual language로 split/fold 지점을 드러냅니다.

입력값에서 출력값으로 먼저 보기

아래 코드는 workshop 테스트가 고정하는 입출력을 글에서 읽기 쉽게 줄인 형태입니다. 실제 예제는 각 module의 README와 test에 더 많은 실패/취소/검증 케이스를 포함합니다.

// 1. Search Pipeline
// input: "r", "re", "red" in one typing burst
val results = pipeline.search(
    queries = flowOf("r", "re", "red"),
    settings = flowOf(settings(tenantId = "tenant-a")),
    sessionClosed = flowOf(),
    debounce = 100.milliseconds,
).toList()
// output: one request, query == "red", tenantId == "tenant-a"

// 2. Race / Fallback
// input: cache(200ms), replica(20ms), remote(120ms)
val winner = catalog.fastestHealthy(listOf(cache, replica, remote)).take(1).toList()
// output: [CatalogSource.REPLICA], cache/remote collectors are cancelled

// 3. Subject Bridge
// input: early event before subscription, then active event after subscription
bridge.publishEvent(DeviceEvent("event-01", "device-01", CONNECTED, "early"))
val received = async { bridge.events.take(1).toList() }
bridge.awaitEventSubscribers()
bridge.publishEvent(DeviceEvent("event-02", "device-01", TELEMETRY, "temperature=22"))
// output: only event-02 is received by PublishSubject subscribers

// 4. Event Aggregation
// input: sampleEvents() == five order events
val chunked = pipeline.chunkedActivity(sampleEvents().asFlow(), chunkSize = 2).toList()
// output chunk sizes: [2, 2, 1]

val windows = pipeline.rollingActivity(sampleEvents().asFlow(), size = 3, step = 2).toList()
// output window sizes: [3, 3, 1]

val groups = pipeline.groupedByOrder(sampleEvents().asFlow()).toList().associateBy { it.key }
// output groups:
//   order-1 -> [OrderCreated, LineAdded, PaymentAuthorized]
//   order-2 -> [OrderCreated, ShipmentStarted]

val states = pipeline.readModels(sampleEvents().asFlow()).toList()
// output: state snapshots; final order-1 == PAID, final order-2 == SHIPPED

// 5. Parallel Enrichment
// input: O-1001 customer-1001, item suit-01 x2
val enriched = pipeline.enrichInParallel(flowOf(order), parallelism = 3) { dispatchers[it] }
    .toList()
// output: O-1001 -> loyaltyGrade=GOLD, discountPercent=10, fulfillable=true

// 6. Metrics Sampling
// input: cpu.usage values 10,20,30,...,100 every 200ms
val leading = pipeline.leadingPreview(highFrequencyCpuSamples(), 501.milliseconds).toList()
val trailing = pipeline.dashboardSamples(highFrequencyCpuSamples(), 501.milliseconds).toList()
// output leading values:  [10.0, 40.0, 70.0, 100.0]
// output trailing values: [30.0, 60.0, 90.0, 100.0]

Search pipeline: 입력 burst와 최신 조건을 분리한다

검색 자동완성은 생각보다 많은 결정을 포함합니다. 빈 문자열은 버려야 하고, 사용자가 빠르게 타이핑하면 마지막 query만 의미가 있으며, tenant 설정이나 feature flag는 검색이 실행되는 시점의 최신 값을 써야 합니다. 그리고 session이 끝나면 대기 중인 검색도 멈춰야 합니다.

예제의 핵심 흐름은 다음과 같습니다.

검색 pipeline은 입력 burst에서 마지막 query만 남기고, 그 순간의 설정 snapshot을 붙인 뒤, 오래된 검색을 취소합니다.

queries
    .bufferingDebounce(debounce)
    .filter { it.isNotBlank() }
    .withLatestFrom(settings) { query, latestSettings ->
        SearchRequest(query, latestSettings)
    }
    .flatMapLatest { request -> adapter.search(request) }
    .takeUntil(sessionClosed)

bufferingDebounce는 “burst 안에서 마지막 query만 남긴다”는 계약입니다. withLatestFrom은 “검색 요청이 만들어지는 순간의 설정을 snapshot으로 붙인다”는 계약입니다. flatMapLatest는 이전 검색이 아직 끝나지 않았더라도 더 최신 query가 들어오면 이전 검색을 취소합니다.

여기서 중요한 것은 cancellation입니다. search adapter는 CancellationException을 삼키면 안 됩니다. 예제 테스트는 cancellation이 정상적으로 밖으로 전파되는지 고정합니다. stream operator는 cancellation boundary를 숨기려고 쓰는 것이 아니라, 어디서 취소되어야 하는지 더 잘 보이게 하려고 쓰는 것입니다.

Race/fallback: 빠른 source와 우선순위 source는 다르다

cache, local replica, remote API, backup endpoint가 있을 때 “먼저 오는 값”이 항상 정답은 아닙니다. 어떤 화면은 가장 빠른 정상 응답 하나면 충분하고, 어떤 작업은 우선순위 순서대로 fallback해야 하며, 어떤 enrichment는 여러 source의 partial result를 합쳐야 합니다.

flow-extensions-race-fallback 예제는 이 차이를 operator 선택으로 드러냅니다.

race/amb는 먼저 정상 값을 낸 source를 고르고 나머지를 취소합니다. concat은 우선순위를 지키고, merge는 여러 source의 partial result를 함께 씁니다.

선택	의미
race / amb	먼저 정상 값을 내는 source가 이기고, 나머지는 취소
concat	strict fallback 순서를 유지
concatArrayEager	source는 미리 시작하되 output 순서는 유지
concatMapEager	동적 source를 eager하게 시작하고 outer order를 보존
merge	partial enrichment처럼 여러 source의 기여를 함께 사용
materialize / dematerialize	실패를 값으로 관찰한 뒤 마지막에 terminal error로 변환

이 예제는 manual select나 async 조립을 없애자는 이야기가 아닙니다. 더 정확히는 “latency 우선인가, priority 우선인가, partial contribution인가”를 코드 표면에서 읽히게 만들자는 이야기입니다.

Subject bridge: callback stream도 종류가 있다

외부 SDK callback을 Flow로 바꿀 때 callbackFlow 하나로 모든 것을 처리하면 늦게 붙은 subscriber의 의미가 흐려집니다. 과거 event를 받을 수 있는지, 최신 state 하나만 받으면 되는지, fan-out을 기다려야 하는지, work item이 한 번만 소비되어야 하는지가 서로 다르기 때문입니다.

Subject bridge 예제는 callback을 다섯 가지 stream 성격으로 나눕니다.

callback을 Flow로 바꾸는 것보다 중요한 것은 subscription 계약입니다. 늦게 붙은 subscriber가 과거 event, 최신 state, replay history, work item 중 무엇을 받는지가 달라집니다.

Subject	쓰임
PublishSubject	현재 붙어 있는 subscriber에게만 event 전달
BehaviorSubject	늦게 붙은 subscriber에게 최신 state를 먼저 전달
ReplaySubject	제한된 history를 replay
MulticastSubject	기대한 subscriber 수가 모일 때까지 fan-out을 조율
UnicastWorkSubject	work item을 단일 consumer queue처럼 한 번만 소비

이 구분은 테스트에서도 중요합니다. 예제는 late subscriber가 과거 event를 받는지, latest state를 받는지, work item이 중복 소비되지 않는지를 따로 확인합니다. callback bridge의 핵심은 “Flow로 바꿨다”가 아니라 “어떤 구독 계약으로 바꿨는가”입니다.

Aggregation과 sampling: window는 메모리 계약이다

event aggregation 예제는 replay stream을 chunked, windowed, groupBy, scanWith로 다룹니다. metric sampling 예제는 burst metric을 throttleLeading과 throttleTrailing로 나눕니다.

둘의 공통점은 window가 단지 편의 API가 아니라 메모리와 지연 시간의 계약이라는 점입니다.

chunked: event를 concrete batch로 묶는다

chunked(2)는 timeline을 두 개씩 잘라 List 같은 concrete batch를 내보냅니다. 마지막 partial batch를 허용하면 [e5]도 출력됩니다.

events
    .chunked(chunkSize, partialWindow = true)
    .map { chunk -> summarize(chunk) }

위 예제 입력이 e1, e2, e3, e4, e5라면 출력 batch 크기는 [2, 2, 1]입니다. 이 operator를 고를 때는 batch 하나가 메모리에 얼마나 쌓일 수 있는지 같이 설명해야 합니다.

windowed: 겹치는 window를 만든다

windowed(size = 3, step = 2)는 겹치는 window를 만들 수 있습니다. 같은 event가 둘 이상의 window에 들어갈 수 있다는 점이 chunked와 다릅니다.

sampleEvents() 기준으로 첫 window는 [e1, e2, e3], 두 번째 window는 [e3, e4, e5], 마지막 partial window는 [e5]입니다. 출력 크기는 [3, 3, 1]이지만, 이는 batch size가 아니라 window size입니다.

groupBy: 시간이나 크기가 아니라 key로 stream을 나눈다

groupBy(orderId)는 timeline을 자르지 않습니다. 같은 key를 가진 event가 자기 stream으로 흘러가도록 partition합니다.

order-1은 Created, LineAdded, PaymentAuthorized stream이 되고, order-2는 Created, ShipmentStarted stream이 됩니다. 그래서 groupBy는 “batch 몇 개가 나왔는가”보다 “key별 stream lifecycle을 어떻게 닫을 것인가”가 더 중요합니다.

scanWith: batch가 아니라 상태 snapshot을 누적한다

scanWith(seed, accumulator)는 event를 묶지 않습니다. 이전 상태와 현재 event로 다음 read model snapshot을 만듭니다.

readModels(sampleEvents())는 batch stream이 아니라 state snapshot stream입니다. 마지막 상태를 보면 order-1은 PAID, order-2는 SHIPPED가 됩니다.

metrics
    .throttleLeading(window)
    .filter { it.value >= alertThreshold }

throttleLeading은 window의 첫 값을 빠르게 내보내고, throttleTrailing은 window가 안정된 뒤 마지막 값을 내보냅니다.

throttleLeading은 alert preview에 어울립니다. window 안의 첫 값을 빨리 보여주기 때문입니다. 반대로 throttleTrailing은 dashboard tile에 어울립니다. window가 안정된 뒤 마지막 값을 보여주기 때문입니다.

예제들은 입력 검증도 stream 바깥으로 밀어내지 않습니다. invalid id, 음수 금액, 제어 문자, non-finite metric 같은 값은 collect 전에 실패해야 합니다. 로그용 toString()과 Flow<T>.log()도 민감한 값을 그대로 흘리지 않도록 redaction을 전제로 둡니다.

Parallel enrichment: 병렬 rail은 독립 작업에만 쓴다

flow-extensions-parallel-enrichment는 주문별 customer lookup, inventory check, discount 계산을 parallel rail로 나눕니다. 각 주문의 enrichment가 독립적이기 때문에 Flow<T>.parallel(...)로 rail을 나누고, 마지막에 sequential()로 다시 단일 stream으로 접습니다.

parallel(3)은 독립적인 enrichment를 rail로 나누고, sequential()은 결과를 다시 단일 output stream으로 접습니다.

orders
    .parallel(parallelism, runOn)
    .map { rail ->
        enrich(rail)
    }
    .sequential()

여기서 병렬화의 기준은 “빨라질 것 같다”가 아닙니다. 각 rail이 공유 mutable state 없이 독립적으로 처리될 수 있고, 마지막에 다시 순서를 접는 지점이 명확해야 합니다. 이 조건이 없으면 operator가 깔끔해 보여도 race condition을 감춘 코드가 됩니다.

운영에서 조심할 점

Flow extensions를 쓴다고 coroutine 운영 계약이 바뀌지는 않습니다. 오히려 더 분명히 드러나야 합니다.

• suspend call 주변에서 CancellationException을 삼키지 않는다.
• lifecycle 종료 신호는 takeUntil 같은 명시적인 boundary로 둔다.
• fallback과 race operator를 섞을 때 latency 우선, priority 우선, partial merge 중 무엇을 원하는지 먼저 정한다.
• window/chunk/grouping은 메모리와 지연 시간의 상한을 함께 설명해야 한다.
• 로그와 debug rendering은 secret, token, password, raw payload를 그대로 내보내지 않는다.
• parallel rail은 독립성이 검증된 작업에만 적용하고, 다시 단일 stream으로 접는 위치를 명확히 둔다.

이 예제들이 좋은 이유는 Flow를 “더 멋진 chain”으로 보여주지 않는다는 점입니다. 각 operator가 어떤 failure, cancellation, subscriber, window, backpressure 계약을 담당하는지 테스트와 README에서 같이 드러냅니다.

소스에서 이어 읽기

• flow-extensions-search-pipeline
• flow-extensions-race-fallback
• flow-extensions-subject-bridge
• flow-extensions-event-aggregation
• flow-extensions-parallel-enrichment
• flow-extensions-metrics-sampling