NetFlow KMP

Overview

NetFlow wraps Ktor with a typed request/response API for Kotlin Multiplatform. Calls come back as a Flow (with built-in loading/success/error states), as a direct suspending call for one-shot work, or as Jetpack Paging 3 pages via the optional netflow-paging module.

A NetFlowClient call always separates the type coming off the wire (ApiType) from the type your UI actually consumes (DisplayType). 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.

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>")
}

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.

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 implements PagingModel:

@Serializable
data class PostDto(
    val id: Int,
    val title: String,
    override var page: Int = 0,
    override var lastUpdatedTimestamp: Long = 0L
) : 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.

Related

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