NetFlow KMP

Overview

NetFlow is a typed request/response layer for Kotlin Multiplatform, built on the native HTTP client of each platform: OkHttp on Android, URLSession on iOS. Calls come back as a Flow with built-in loading, success and error states, as a suspending call for one-shot work, or as Jetpack Paging 3 pages via the optional netflow-paging module.

There are two ways in, and they share a client, a set of types, and a test double.

Annotated interfaces (netflow-annotations plus netflow-ksp) cover straightforward endpoints. You declare an interface, KSP generates the implementation, and the return type picks the response strategy. This is the on-ramp, not a separate library: the generated code is the DSL below, written out for you.

The call {} DSL covers everything an annotation cannot express, because annotations cannot hold a lambda: local cache reads, onNetworkSuccess side effects, remote-plus-local paging, and per-request retry policies.

Either way, a call separates the type coming off the wire (ApiType) from the type your UI consumes (DisplayType). In the DSL, when they differ the compiler requires a transform lambda, so forgetting to map a DTO to a domain model is a build error rather than a runtime surprise. Annotated methods return the ApiType and you map one layer out, in the repository.

Platforms: Android, iOS.

Installation

Core module

dependencies {
    implementation("io.github.kmpbits:netflow-core:<version>")
}

Paging module (optional)

Adds responsePaginated, backed by Jetpack Paging 3.

dependencies {
    implementation("io.github.kmpbits:netflow-core:<version>")
    implementation("io.github.kmpbits:netflow-paging:<version>")
}

Annotations module (optional)

Adds the annotation set and the KSP processor that generates implementations from your annotated interfaces.

// build.gradle.kts
plugins {
    id("com.google.devtools.ksp")
}

kotlin {
    sourceSets {
        commonMain {
            kotlin.srcDir("build/generated/ksp/metadata/commonMain/kotlin")
        }
        commonMain.dependencies {
            implementation("io.github.kmpbits:netflow-annotations:<version>")
        }
    }
}

dependencies {
    add("kspCommonMainMetadata", "io.github.kmpbits:netflow-ksp:<version>")
}

tasks.withType<org.jetbrains.kotlin.gradle.tasks.KotlinCompilationTask<*>>().configureEach {
    if (name != "kspCommonMainKotlinMetadata") {
        dependsOn("kspCommonMainKotlinMetadata")
    }
}

Both extra pieces of wiring are required. Without the srcDir line the generated file is written to disk but no source set can see it. Without the task dependency, target compilations can start before KSP has run, producing an unresolved reference to the generated factory that disappears on a second build.

netflow-ksp does not depend on netflow-paging. It emits responsePaginated as a fully qualified name and lets your classpath resolve it, so you only need the paging module if you actually return Flow<PagingData<T>>.

Replace <version> with the latest release on GitHub or Maven Central.

Getting started

Initialize the client

val client = netFlowClient {
    baseUrl = "https://api.example.com"

    header(Header(HttpHeader.custom("custom-header"), "value"))
    header(Header(HttpHeader.CONTENT_TYPE), "application/json")
}

Basic request

val response = client.call {
    path = "/users"
    method = HttpMethod.Get
}.response()

Deserialize to a model

val user: User = client.call {
    path = "/users/1"
}.responseToModel<User>()

responseToModel is the only extension that throws on failure. Every other extension below returns a sealed state instead.

Annotated interfaces

Annotate an interface with @NetFlowApi, and KSP generates an implementation plus a NetFlowClient.create<Name>() extension.

// commonMain
@NetFlowApi
interface TodoApi {

    @GET("todos")
    suspend fun getTodos(@Query completed: Boolean?): AsyncState<List<TodoDto>>

    @GET("todos/{id}")
    fun observeTodo(@Path id: Int): Flow<ResultState<TodoDto>>

    @Headers("Accept: application/json", "X-Client: netflow")
    @POST("todos")
    suspend fun create(@Body request: CreateTodoRequest): AsyncState<TodoDto>

    @DELETE("todos/{id}")
    suspend fun delete(@Path id: Int): AsyncState<Unit>

    @Wrapped
    @GET("todos/{id}")
    suspend fun getWrapped(@Path id: Int): AsyncState<TodoDto>
}

val api = client.createTodoApi()

Generation happens at compile time through KSP. There is no reflection, and the output is a readable .kt file under build/generated/ksp/metadata/commonMain/kotlin.

The return type picks the strategy

There is no strategy annotation. The processor reads the return type and generates the matching DSL call.

Adding @Wrapped routes any of these to the responseWrapped* equivalent, for APIs that return { "data": ... }.

Parameter and method annotations

@Path, @Query and @Header take their wire name from the parameter name. Pass a name to override it: @Query("is_done") completed: Boolean?.

Mapping to domain types

Annotations cannot carry a lambda, so there is no transform parameter on an annotated method. Annotated methods return the ApiType, and you map one layer out with the .map helpers in netflow-core. The mapping stays explicit and compiler-checked, it just lives in the repository.

// commonMain
class TodoRepository(client: NetFlowClient) {

    private val api = client.createTodoApi()

    suspend fun getTodos(): AsyncState<List<Todo>> =
        api.getTodos(completed = null).map { dtos -> dtos.map { it.toModel() } }

    fun observeTodo(id: Int): Flow<ResultState<Todo>> =
        api.observeTodo(id).map { state -> state.map { it.toModel() } }
}

Paged endpoints

A method returning Flow<PagingData<T>> generates a network-only paged call. T must extend PagingModel.

// commonMain
@GET("todos")
fun pagedTodos(): Flow<PagingData<TodoDto>>

@Paginated(pageQueryName = "page", pageSize = 15)
@GET("todos")
fun pagedTodosSmall(): Flow<PagingData<TodoDto>>

@Paginated is optional and defaults to page and a page size of 20. Annotated paged calls set onlyApiCall = true, so there is no local cache. Remote-plus-local paging needs a PagingSource factory, an insertAll block and a timestamp lambda, so it stays on the call {} DSL.

Composing the response yourself (NetFlowCall)

When an endpoint’s request is boilerplate but its response needs a cache, return NetFlowCall. The annotation captures path, method, query, headers and body; the caller owns the response side with the full DSL.

// commonMain
@GET("todos")
fun todosCall(): NetFlowCall

// repository
fun getTodos(): Flow<ResultState<List<Todo>>> =
    api.todosCall().responseListFlow<TodoDto, Todo>(transform = { it.toModel() }) {

        onNetworkSuccess { dtos ->
            database.todoQueries.transaction {
                dtos.forEach { database.todoQueries.insertTodo(it.toEntity()) }
            }
        }

        local(
            { observe { database.todoQueries.selectTodos() } },
            transform = { entities -> entities.map { it.toModel() } }
        )
    }

Because the response strategy now belongs to the caller, @Wrapped on a NetFlowCall method is a compile error. NetFlowClient.prepareCall { } gives the same request/response split to hand-written DSL code.

Compile-time validation

The processor fails the build rather than deferring to runtime:

Function 'getTodos' returns AsyncState and must be suspend.
Function 'observeTodo' returns Flow and must not be suspend.
Function 'observeTodo' path declares '{id}' but there is no @Path parameter named 'id'.
Function 'create' has more than one @Body; at most one @Body is allowed.
Function 'pagedTodos' — Flow<PagingData<T>> requires T to extend PagingModel.
@Paginated on 'getTodos' requires a Flow<PagingData<T>> return type.

Working with Flow

Same type

When the DTO and the domain model are the same type, pass a single type parameter and skip transform entirely:

val flow = client.call {
    path = "/users/1"
}.responseFlow<UserDto>()

Different types

When ApiType and DisplayType differ, transform is required as the first argument:

val flow = client.call {
    path = "/users/1"
}.responseFlow<UserDto, User>(transform = { it.toModel() })

With local cache

val usersFlow = client.call {
    path = "/users"
    method = HttpMethod.Get
}.responseFlow<UserDto, User>(transform = { it.toModel() }) {
    onNetworkSuccess { dto ->
        queries.insertUser(dto.toEntity())
    }

    local({ observe { queries.getUser() } }, transform = { it.toModel() })
}

The transform inside local() maps the database entity to DisplayType; it drives what’s shown while the network call is still in flight. The transform on the function itself maps the network ApiType to DisplayType once the response lands.

Offline-only

local({
    onlyLocalCall = true
    call { queries.getAllUsers() }
}, transform = { it.toModel() })

Wrapped responses

For APIs that return { "data": { ... } } instead of a plain object:

responseWrappedFlow<UserDto>()
responseWrappedFlow<UserDto, User>(transform = { it.toModel() })

Or set wrappedResponse = true inside the builder when using responseFlow.

List variants

responseListFlow<UserDto>()
responseWrappedListFlow<UserDto>()
responseListFlow<UserDto, User>(transform = { it.toModel() })
responseWrappedListFlow<UserDto, User>(transform = { it.toModel() })

Observing

lifecycleScope.launch {
    usersFlow.collectLatest { state ->
        when (state) {
            is ResultState.Loading -> showLoading()
            is ResultState.Success -> showUsers(state.data)
            is ResultState.Error -> showError(state.error.message)
        }
    }
}

Working with Async

For one-shot suspending calls that don’t need observation.

suspend fun deleteUser(id: Int): AsyncState<Unit> {
    return client.call {
        path = "users/$id"
        method = HttpMethod.Delete
    }.responseAsync<Unit> {
        onNetworkSuccess { queries.deleteUser(id) }
    }
}

suspend fun getUser(id: Int): AsyncState<User> {
    return client.call {
        path = "users/$id"
    }.responseAsync<UserDto, User>(transform = { it.toModel() })
}

List and wrapped variants mirror the Flow API:

responseListAsync<UserDto>()
responseWrappedListAsync<UserDto>()
responseListAsync<UserDto, User>(transform = { it.toModel() })
responseWrappedListAsync<UserDto, User>(transform = { it.toModel() })
responseWrappedAsync<UserDto>()
responseWrappedAsync<UserDto, User>(transform = { it.toModel() })

Working with paging (netflow-paging)

responsePaginated integrates Jetpack Paging 3, supporting network-only and remote-plus-local strategies. The API response model extends PagingModel, an abstract class that carries the page and lastUpdatedTimestamp fields the pager needs:

@Serializable
data class PostDto(
    val id: Int,
    val title: String
) : PagingModel()

Network-only paging

fun getPosts(): Flow<PagingData<Post>> = client.call {
    path = "/posts"
}.responsePaginated<PostDto, Post> {
    onlyApiCall = true
    networkTransform { it.toModel() }
}

Remote + local paging with localQuery

The recommended option when no custom PagingSource is needed. Pass countQuery, itemsQuery, and an invalidation flow (SQLDelight users pass query.asFlow(), Room users pass their own Flow<List<T>>); NetFlow creates and manages the PagingSource internally:

fun getPosts(): Flow<PagingData<Post>> = client.call {
    path = "/posts"
}.responsePaginated<PostDto, Post> {
    localQuery(
        countQuery   = { database.postQueries.countPosts().executeAsOne() },
        itemsQuery   = { limit, offset -> database.postQueries.selectPosts(limit, offset).executeAsList() },
        invalidation = database.postQueries.selectAllPosts().asFlow(),
        transform    = { it.toModel() }
    )

    deleteOnRefresh = false
    insertAll(transform = { it.toEntity() }) { posts ->
        database.postQueries.transaction {
            database.postQueries.deleteAll()
            posts.forEach { database.postQueries.insertPost(it) }
        }
    }

    firstItemDatabase(
        itemDatabase = { database.postQueries.getFirstPost().executeAsOneOrNull() },
        timestamp    = { it.lastUpdatedTimestamp }
    )
}

Remote + local paging with a custom PagingSource

For full control over local loading, provide a PagingSource<Int, E> (or PagingSource<Long, E> via localSource’s localSourceLong counterpart, for SQLDelight sources keyed by Long) and wire it up with localSource(pagingSource = { ... }, transform = { it.toModel() }).

If a query changes and no listener is registered on it, the PagingSource never invalidates and the UI won’t reflect the update after a network refresh or a local delete. Register a Query.Listener that calls invalidate() and removes itself, added in init, following the pattern the sample app’s TodoPagingSource uses.

PagingBuilder options

Consuming pages

In a ViewModel:

val posts = repository.getPosts().cachedIn(viewModelScope)

In Compose:

val posts = viewModel.posts.collectAsLazyPagingItems()

LazyColumn {
    items(count = posts.itemCount, key = posts.itemKey { it.id }) { index ->
        posts[index]?.let { PostItem(it) }
    }
}

On iOS, netflow-paging ships PagingCollectionViewController, a KMP class that bridges paging data to Swift. It’s designed for use with SKIE for async sequence support:

private let delegate = PagingCollectionViewController<Post>()

func loadNextPage() { delegate.loadNextPage() }

func observePagingData() {
    Task {
        for await pagingData in viewModel.posts {
            delegate.submitData(pagingData: pagingData)
        }
    }
}

func observeData() {
    Task {
        for await _ in delegate.onPagesUpdatedFlow {
            self.posts = delegate.getItems()
        }
    }
}

Testing with MockNetFlowClient

MockNetFlowClient implements NetFlowClient and intercepts every request instead of making real network calls, with support for response delays, request recording, and assertion helpers.

val mockClient = MockNetFlowClient { request ->
    when {
        request.path == "posts" && request.method == HttpMethod.Get ->
            NetFlowMockResponse.success("""[{"id":1,"title":"Hello","completed":false}]""")

        request.path.startsWith("posts/") && request.method == HttpMethod.Delete ->
            NetFlowMockResponse.success()

        else -> NetFlowMockResponse.notFound()
    }
}

NetFlowMockResponse helpers

All four accept an optional delay: Duration, for simulating slow networks.

Assertions

mockClient.assertCalled("posts", HttpMethod.Get)
mockClient.assertCalledTimes("posts/1", HttpMethod.Delete, times = 1)
mockClient.assertNotCalled("posts", HttpMethod.Post)

val request = mockClient.recordedRequests.first()
assertEquals(HttpMethod.Post, request.method)

mockClient.clearRecordedRequests()

Advanced configuration

Custom headers

client.call {
    path = "/secure-endpoint"
    header(Header(HttpHeader.custom("Authorization"), "Bearer $token"))
}.responseFlow<SecureDataDto, SecureData>(transform = { it.toModel() })

Query parameters

client.call {
    path = "/users"
    parameter("role" to "admin")
    parameter("active" to true)
}.responseFlow<UserDto, User>(transform = { it.toModel() })

Retry

client.call {
    path = "/unstable-endpoint"
    retry {
        times = RetryTimes.THREE
        delay = 1.seconds
        retryOn = { it is IOException }
    }
}.responseFlow<DataDto, Data>(transform = { it.toModel() })

Error handling

try {
    val response = client.call {
        path = "/might-fail"
    }.responseToModel<Data>()
} catch (e: NetFlowException) {
    when (e) {
        is NetworkException -> { /* handle network issues */ }
        is SerializationException -> { /* handle parsing errors */ }
        is HttpException -> {
            val code = e.code
            val errorBody = e.errorBody
        }
    }
}

Using with DI

single {
    netFlowClient {
        baseUrl = "https://api.example.com"
    }
}

Known limitations

• No multipart/form-data support: file uploads through client.call aren’t supported yet.
• No WebSocket support: NetFlow currently covers request/response and paged HTTP calls only.
• Annotations carry no transform: annotated methods return the ApiType by design. Map to domain types in the repository.
• Annotated paged calls are network-only: remote-plus-local paging stays on the call {} DSL.

Related

How the annotation layer and the DSL fit together, and where the annotations deliberately stop: Two Racing Lines: Annotations and the DSL in NetFlow 0.7.0.

The story behind NetFlow’s move from an Android-only library to Kotlin Multiplatform: Why I Took the Leap from Android-Only to Kotlin Multiplatform.