Как загружать данные в Kotlin с корутинами (Первая часть)
Retrofit с корутинами - это сочетание, которое изменило мой взгляд на работу с HTTP API. Помню времена, когда каждый сетевой запрос превращался в танцы с колбэками, обработкой потоков и ручным парсингом JSON. С появлением поддержки suspend-функций в Retrofit 2.6 всё стало настолько просто, что первое время я не верил - неужели больше не нужны адаптеры, обёртки и сложные цепочки преобразований?
Загрузка из REST API через Retrofit
Основа работы с Retrofit и корутинами Kotlin - suspend методы в интерфейсе REST API. Никаких Call<T> объектов, никаких enqueue() вызовов:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
| interface UserApiService {
@GET("users/{id}")
suspend fun getUser(@Path("id") userId: String): User
@POST("users")
suspend fun createUser(@Body user: CreateUserRequest): User
@PUT("users/{id}")
suspend fun updateUser(@Path("id") userId: String, @Body user: User): User
@DELETE("users/{id}")
suspend fun deleteUser(@Path("id") userId: String): Response<Unit>
} |
|
Для обработки ошибок HTTP я создаю wrapper функцию, которая преобразует исключения в удобный для работы формат:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| class ApiClient(private val userApi: UserApiService) {
suspend fun <T> safeApiCall(apiCall: suspend () -> T): ApiResult<T> {
return try {
val result = apiCall()
ApiResult.Success(result)
} catch (e: HttpException) {
when (e.code()) {
404 -> ApiResult.Error.NotFound
401 -> ApiResult.Error.Unauthorized
in 500..599 -> ApiResult.Error.ServerError(e.message())
else -> ApiResult.Error.Unknown(e.message())
}
} catch (e: IOException) {
ApiResult.Error.NetworkError(e.message ?: "Network unavailable")
} catch (e: Exception) {
ApiResult.Error.Unknown(e.message ?: "Unknown error")
}
}
suspend fun loadUser(userId: String): ApiResult<User> {
return safeApiCall { userApi.getUser(userId) }
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
| suspend fun loadActiveUsers(): List<User> = withContext(Dispatchers.IO) {
val allUsers = userApi.getAllUsers()
allUsers
.filter { it.isActive }
.sortedBy { it.lastLoginDate }
.map { user ->
// Дополнительная обработка каждого пользователя
user.copy(displayName = formatDisplayName(user.firstName, user.lastName))
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
| suspend fun loadUserDashboard(userId: String): DashboardData = coroutineScope {
val userDeferred = async { userApi.getUser(userId) }
val postsDeferred = async { userApi.getUserPosts(userId) }
val friendsDeferred = async { userApi.getUserFriends(userId) }
val statisticsDeferred = async { userApi.getUserStatistics(userId) }
DashboardData(
user = userDeferred.await(),
posts = postsDeferred.await().take(10), // последние 10 постов
friends = friendsDeferred.await(),
statistics = statisticsDeferred.await()
)
} |
|
Retrofit отлично интегрируется с Flow для создания реактивных потоков данных. Можно легко создать поток, который периодически обновляет данные с сервера:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| class NewsRepository(private val newsApi: NewsApiService) {
fun observeLatestNews(): Flow<List<NewsArticle>> = flow {
while (true) {
try {
val news = newsApi.getLatestNews()
emit(news)
} catch (e: Exception) {
// В случае ошибки не останавливаем поток
emit(emptyList())
}
delay(TimeUnit.MINUTES.toMillis(5)) // обновляем каждые 5 минут
}
}.flowOn(Dispatchers.IO)
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| suspend fun completeUserRegistration(
email: String,
password: String,
profile: UserProfile
): RegistrationResult = coroutineScope {
// Шаг 1: Создаём аккаунт
val account = userApi.createAccount(CreateAccountRequest(email, password))
// Шаг 2: Загружаем аватар параллельно с созданием профиля
val uploadAvatarJob = async {
if (profile.avatarFile != null) {
userApi.uploadAvatar(account.id, profile.avatarFile)
} else null
}
// Шаг 3: Создаём профиль пользователя
val userProfile = userApi.createProfile(
account.id,
profile.copy(avatarUrl = uploadAvatarJob.await()?.url)
)
// Шаг 4: Отправляем welcome email
userApi.sendWelcomeEmail(account.email)
RegistrationResult.Success(account, userProfile)
} |
|
Каким самым простым методом (Android приложение на Kotlin ) можно забрать данные с сервера?
Добрый день.
Проект состоит из:
- Контроллера, который отправляет данные на сервер GET... Как при появлении Wifi загружать данные
Есть приложение. В нем есть много фрагментов которые подгружают данные из сервера, но юзер зашел в... Kotlin "it" как перевести на java?
есть пример кода на котлине:
override fun one(id: Long): Observable<Banner> {
return... Как исправить утечку памяти Android Studio Kotlin
Я обращаюсь к базе данных получаю ответ в json и процессе того как обрабатывается возникает ошибка....
GraphQL запросы с корутинами
GraphQL с корутинами открывает совершенно новые возможности для работы с API. После двух лет использования Apollo GraphQL в проектах на Kotlin могу сказать: это сочетание решает многие проблемы традиционного REST подхода, особенно когда речь идёт о сложных запросах данных и real-time обновлениях. Основное преимущество GraphQL - возможность запросить именно те данные, которые нужны для конкретного экрана. Никакого overfetching, никаких лишних полей. Apollo Kotlin автоматически генерирует suspend-функции из GraphQL схем:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| class UserRepository(private val apolloClient: ApolloClient) {
suspend fun getUserProfile(userId: String): ApolloResponse<GetUserProfileQuery.Data> {
return apolloClient.query(
GetUserProfileQuery(userId = userId)
).execute()
}
suspend fun updateUserName(userId: String, newName: String): ApolloResponse<UpdateUserMutation.Data> {
return apolloClient.mutation(
UpdateUserMutation(userId = userId, name = newName)
).execute()
}
} |
|
Фрагменты GraphQL особенно мощны в сочетании с корутинами. Можно создавать переиспользуемые части запросов и комбинировать их для разных экранов:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| suspend fun loadDashboardData(userId: String): DashboardData = coroutineScope {
val profileDeferred = async {
apolloClient.query(GetUserProfileQuery(userId)).execute()
}
val postsDeferred = async {
apolloClient.query(GetUserPostsQuery(userId, first = 10)).execute()
}
val friendsDeferred = async {
apolloClient.query(GetUserFriendsQuery(userId)).execute()
}
val profileResponse = profileDeferred.await()
val postsResponse = postsDeferred.await()
val friendsResponse = friendsDeferred.await()
DashboardData(
profile = profileResponse.data?.user?.userProfile,
posts = postsResponse.data?.user?.posts?.nodes ?: emptyList(),
friends = friendsResponse.data?.user?.friends?.nodes ?: emptyList()
)
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| class ChatRepository(private val apolloClient: ApolloClient) {
fun observeMessages(chatId: String): Flow<ChatMessage> =
apolloClient
.subscription(MessageSubscription(chatId = chatId))
.toFlow()
.mapNotNull { response ->
response.data?.messageAdded?.let { message ->
ChatMessage(
id = message.id,
text = message.text,
author = message.author.name,
timestamp = message.createdAt
)
}
}
.catch { exception ->
when (exception) {
is ApolloNetworkException -> emit(ChatMessage.connectionError())
else -> throw exception
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| suspend fun loadUserWithCaching(userId: String, forceRefresh: Boolean = false): User? {
val fetchPolicy = if (forceRefresh) {
FetchPolicy.NetworkOnly
} else {
FetchPolicy.CacheFirst
}
val response = apolloClient.query(
GetUserQuery(userId = userId)
).fetchPolicy(fetchPolicy).execute()
return response.data?.user?.let { userData ->
User(
id = userData.id,
name = userData.name,
email = userData.email,
avatarUrl = userData.avatar?.url
)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| suspend fun batchUpdateUsers(updates: List<UserUpdate>): List<User> = coroutineScope {
val mutations = updates.map { update ->
async {
apolloClient.mutation(
UpdateUserMutation(
userId = update.userId,
name = update.name,
email = update.email
)
).execute()
}
}
mutations.awaitAll().mapNotNull { response ->
response.data?.updateUser?.user?.let { userData ->
User(userData.id, userData.name, userData.email)
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| suspend fun safeGraphQLCall<T>(
operation: suspend () -> ApolloResponse<T>
): GraphQLResult<T> {
return try {
val response = operation()
when {
response.hasErrors() -> {
val errorMessages = response.errors?.map { it.message } ?: emptyList()
GraphQLResult.Error(errorMessages.joinToString(", "))
}
response.data != null -> GraphQLResult.Success(response.data!!)
else -> GraphQLResult.Error("Empty response")
}
} catch (e: ApolloException) {
GraphQLResult.Error(e.message ?: "GraphQL operation failed")
}
} |
|
Кэширование и Room базы данных
Room с корутинами - это союз, который превратил работу с локальными данными из мучения в удовольствие. Помню, как раньше приходилось городить конструкции с AsyncTask или ThreadPoolExecutor для каждого запроса к SQLite. Теперь всё выглядит так элегантно, что иногда забываешь о том, что под капотом происходят сложные операции с базой данных.
Основа любого Room DAO с корутинами - suspend методы. Они автоматически выполняются в background потоке, не блокируя UI:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| @Dao
interface UserDao {
@Query("SELECT * FROM users WHERE id = :userId")
suspend fun getUser(userId: String): User?
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertUser(user: User): Long
@Update
suspend fun updateUser(user: User)
@Delete
suspend fun deleteUser(user: User)
@Query("SELECT * FROM users WHERE last_sync < :timestamp")
suspend fun getStaleUsers(timestamp: Long): List<User>
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
| @Dao
interface ArticleDao {
@Query("SELECT * FROM articles ORDER BY published_date DESC")
fun observeAllArticles(): Flow<List<Article>>
@Query("SELECT * FROM articles WHERE is_favorite = 1")
fun observeFavoriteArticles(): Flow<List<Article>>
@Query("SELECT * FROM articles WHERE category = :category")
fun observeArticlesByCategory(category: String): Flow<List<Article>>
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| class ArticleRepository(
private val articleDao: ArticleDao,
private val articleApi: ArticleApiService
) {
suspend fun getArticle(id: String): Article? {
// Сначала ищем в кэше
val cached = articleDao.getArticle(id)
if (cached != null && !cached.isStale()) {
return cached
}
// Загружаем с сервера
return try {
val fresh = articleApi.getArticle(id)
articleDao.insertArticle(fresh.copy(cachedAt = System.currentTimeMillis()))
fresh
} catch (e: Exception) {
// Fallback на устаревший кэш
cached
}
}
fun observeArticles(): Flow<List<Article>> = articleDao.observeAllArticles()
.map { articles ->
if (articles.isEmpty() || articles.any { it.isStale() }) {
refreshArticlesInBackground()
}
articles
}
private fun refreshArticlesInBackground() {
CoroutineScope(Dispatchers.IO).launch {
try {
val fresh = articleApi.getAllArticles()
articleDao.insertAll(fresh)
} catch (e: Exception) {
// Логируем, но не ломаем UI
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| @Dao
abstract class UserDao {
@Insert
abstract suspend fun insertUser(user: User): Long
@Insert
abstract suspend fun insertUserPreferences(preferences: UserPreferences)
@Transaction
open suspend fun insertUserWithPreferences(user: User, preferences: UserPreferences) {
val userId = insertUser(user)
insertUserPreferences(preferences.copy(userId = userId))
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| data class UserWithPosts(
@Embedded val user: User,
@Relation(
parentColumn = "id",
entityColumn = "user_id"
)
val posts: List<Post>
)
@Dao
interface UserDao {
@Transaction
@Query("SELECT * FROM users WHERE id = :userId")
suspend fun getUserWithPosts(userId: String): UserWithPosts?
@Transaction
@Query("SELECT * FROM users WHERE is_active = 1")
fun observeActiveUsersWithPosts(): Flow<List<UserWithPosts>>
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| val MIGRATION_1_2 = object : Migration(1, 2) {
override fun migrate(database: SupportSQLiteDatabase) {
// Создаём новую таблицу
database.execSQL("CREATE TABLE user_settings (id INTEGER PRIMARY KEY, user_id TEXT, theme TEXT)")
// Мигрируем данные (в реальности это может быть очень долго)
database.execSQL("INSERT INTO user_settings SELECT id, id, 'dark' FROM users")
}
}
@Database(
entities = [User::class, UserSettings::class],
version = 2
)
abstract class AppDatabase : RoomDatabase() {
abstract fun userDao(): UserDao
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| class DashboardRepository(
private val userDao: UserDao,
private val notificationDao: NotificationDao,
private val settingsDao: SettingsDao
) {
fun observeDashboardData(userId: String): Flow<DashboardData> =
combine(
userDao.observeUser(userId),
notificationDao.observeUnreadCount(userId),
settingsDao.observeUserSettings(userId)
) { user, unreadCount, settings ->
DashboardData(
user = user,
unreadNotifications = unreadCount,
isDarkTheme = settings.theme == "dark",
showNotifications = settings.notificationsEnabled
)
}
} |
|
Комбинирование нескольких источников
Реальные приложения редко работают с одним источником данных. Обычно нужно объединить информацию из API, локальной базы, кэша, настроек пользователя и других источников. Корутины предоставляют элегантные способы комбинирования этих потоков без превращения кода в спагетти. Самый простой случай - параллельная загрузка из нескольких API с последующим объединением результатов:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| class ProfileRepository(
private val userApi: UserApiService,
private val avatarApi: AvatarApiService,
private val settingsApi: SettingsApiService
) {
suspend fun loadCompleteProfile(userId: String): CompleteProfile = coroutineScope {
val userInfoDeferred = async { userApi.getBasicInfo(userId) }
val avatarDeferred = async { avatarApi.getUserAvatar(userId) }
val settingsDeferred = async { settingsApi.getUserSettings(userId) }
val userInfo = userInfoDeferred.await()
val avatar = avatarDeferred.await()
val settings = settingsDeferred.await()
CompleteProfile(
basicInfo = userInfo,
avatarUrl = avatar.highResUrl,
preferences = settings,
isComplete = userInfo.isVerified && avatar.isUploaded
)
}
} |
|
combine и combineLatest решают эту задачу изящно:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| class HomeRepository(
private val newsDao: NewsDao,
private val weatherApi: WeatherApi,
private val notificationManager: NotificationManager
) {
fun observeHomeScreen(): Flow<HomeScreenData> = combine(
newsDao.observeLatestNews().map { it.take(5) },
flow {
while (true) {
emit(weatherApi.getCurrentWeather())
delay(TimeUnit.MINUTES.toMillis(15))
}
},
notificationManager.observeUnreadCount()
) { news, weather, unreadCount ->
HomeScreenData(
topNews = news,
currentWeather = weather,
hasUnreadNotifications = unreadCount > 0,
lastUpdated = System.currentTimeMillis()
)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| class ContentRepository(
private val settingsDao: SettingsDao,
private val contentApi: ContentApiService,
private val contentDao: ContentDao
) {
fun observeLocalizedContent(): Flow<LocalizedContent> =
settingsDao.observeLanguageSettings()
.distinctUntilChanged()
.switchMap { settings ->
combine(
contentDao.observeCachedContent(settings.languageCode),
refreshContentForLanguage(settings.languageCode).asFlow()
) { cached, fresh ->
if (fresh != null && fresh.version > cached.version) {
contentDao.insertContent(fresh)
fresh
} else {
cached
}
}
}
private suspend fun refreshContentForLanguage(lang: String): LocalizedContent? {
return try {
contentApi.getContentForLanguage(lang)
} catch (e: Exception) {
null
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| class RobustDataRepository(
private val primaryApi: PrimaryApiService,
private val backupApi: BackupApiService,
private val localCache: LocalCacheDao
) {
suspend fun loadCriticalData(id: String): CriticalData? {
return try {
// Первый приоритет - основное API
primaryApi.getData(id).also { data ->
localCache.saveData(data) // сохраняем в кэш
}
} catch (primaryError: Exception) {
try {
// Второй приоритет - backup API
backupApi.getData(id).also { data ->
localCache.saveData(data)
}
} catch (backupError: Exception) {
// Последняя надежда - локальный кэш
localCache.getData(id).also { cachedData ->
if (cachedData == null) {
throw DataUnavailableException(
"All sources failed",
listOf(primaryError, backupError)
)
}
}
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| class DashboardAggregator(
private val repositories: List<DashboardDataSource>
) {
fun observeAggregatedDashboard(): Flow<DashboardState> =
repositories
.map { source -> source.observeData() }
.let { flows ->
combine(flows) { dataArray ->
DashboardState(
sections = dataArray.mapNotNull { it },
loadingCount = dataArray.count { it == null },
lastUpdate = System.currentTimeMillis()
)
}
}
} |
|
WebSocket соединения с корутинами
WebSocket с корутинами - это область, где я наделал много ошибок в первые месяцы работы. Казалось бы, что сложного? Открыл соединение, слушай сообщения, отправляй ответы. Но реальность оказалась куда хитрее: разрывы соединения, переподключения, обработка больших объёмов данных и управление жизненным циклом соединения в контексте Android lifecycle. Основа работы с WebSocket в корутинах - преобразование callback-based API в Flow. OkHttp WebSocket работает через listener паттерн, а нам нужен реактивный поток данных:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| class WebSocketManager(private val client: OkHttpClient) {
private val _messages = MutableSharedFlow<String>()
val messages: SharedFlow<String> = _messages.asSharedFlow()
private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected)
val connectionState: StateFlow<ConnectionState> = _connectionState.asStateFlow()
private var webSocket: WebSocket? = null
fun connect(url: String): Flow<WebSocketEvent> = channelFlow {
val listener = object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
_connectionState.value = ConnectionState.Connected
trySend(WebSocketEvent.Connected)
}
override fun onMessage(webSocket: WebSocket, text: String) {
_messages.tryEmit(text)
trySend(WebSocketEvent.MessageReceived(text))
}
override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
_connectionState.value = ConnectionState.Disconnecting
}
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
_connectionState.value = ConnectionState.Error(t.message ?: "Unknown error")
trySend(WebSocketEvent.Error(t))
}
}
val request = Request.Builder().url(url).build()
webSocket = client.newWebSocket(request, listener)
awaitClose { webSocket?.close(1000, "Connection closed") }
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| class ReconnectingWebSocket(
private val webSocketManager: WebSocketManager
) {
private val reconnectionScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
fun connectWithRetry(url: String): Flow<String> = flow {
var retryCount = 0
val maxRetries = 5
while (true) {
try {
webSocketManager.connect(url)
.collect { event ->
when (event) {
is WebSocketEvent.Connected -> {
retryCount = 0 // сбрасываем счётчик при успешном подключении
}
is WebSocketEvent.MessageReceived -> {
emit(event.message)
}
is WebSocketEvent.Error -> {
if (retryCount < maxRetries) {
val delay = (2.0.pow(retryCount) * 1000).toLong()
delay(delay)
retryCount++
} else {
throw ConnectionException("Max retries exceeded")
}
}
}
}
} catch (e: Exception) {
if (retryCount < maxRetries) {
delay(5000) // базовая задержка между попытками
retryCount++
} else {
throw e
}
}
}
}.distinctUntilChanged()
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| class BufferedWebSocketSender {
private val messageQueue = Channel<String>(Channel.UNLIMITED)
private val connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected)
init {
CoroutineScope(Dispatchers.IO + SupervisorJob()).launch {
connectionState
.filter { it is ConnectionState.Connected }
.collect {
// Отправляем все накопленные сообщения при подключении
while (!messageQueue.isEmpty) {
val message = messageQueue.tryReceive().getOrNull()
message?.let { webSocket?.send(it) }
}
}
}
}
suspend fun sendMessage(message: String) {
if (connectionState.value is ConnectionState.Connected) {
webSocket?.send(message)
} else {
messageQueue.trySend(message) // буферизуем для отправки позже
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| class StreamingWebSocketProcessor {
private val messageBuilder = StringBuilder()
private val completeMessages = MutableSharedFlow<ProcessedMessage>()
suspend fun processIncomingData(data: String) {
messageBuilder.append(data)
// Предполагаем, что сообщения разделены специальным маркером
val delimiter = "\n##END##\n"
var delimiterIndex = messageBuilder.indexOf(delimiter)
while (delimiterIndex != -1) {
val completeMessage = messageBuilder.substring(0, delimiterIndex)
messageBuilder.delete(0, delimiterIndex + delimiter.length)
val processed = processMessage(completeMessage)
completeMessages.emit(processed)
delimiterIndex = messageBuilder.indexOf(delimiter)
}
}
private suspend fun processMessage(raw: String): ProcessedMessage = withContext(Dispatchers.Default) {
// Тяжёлая обработка сообщения в background потоке
val parsed = parseJsonMessage(raw)
val validated = validateMessage(parsed)
ProcessedMessage(validated, System.currentTimeMillis())
}
} |
|
Миграция легаси-кода с Thread и AsyncTask
Миграция legacy кода на корутины - это процесс, который я прошёл в десятках проектов. В каждом старом приложении обязательно найдутся остатки AsyncTask, Thread или даже Handler с Runnable. Избавляться от этого наследия нужно аккуратно, понимая, что каждый кусок старого кода когда-то решал конкретную проблему.
AsyncTask был основным инструментом для фоновых операций в Android до появления корутин. Типичный код выглядел ужасающе:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| // Старый подход с AsyncTask
private inner class LoadUserDataTask : AsyncTask<String, Void, User?>() {
override fun doInBackground(vararg userIds: String): User? {
return try {
userApi.getUser(userIds[0])
} catch (e: Exception) {
null
}
}
override fun onPostExecute(result: User?) {
if (result != null) {
displayUser(result)
} else {
showError("Failed to load user")
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
| // Новый подход с корутинами
fun loadUser(userId: String) {
viewModelScope.launch {
try {
val user = userRepository.getUser(userId)
_userState.value = UserState.Success(user)
} catch (e: Exception) {
_userState.value = UserState.Error(e.message ?: "Unknown error")
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
| // Legacy Thread код
private val backgroundThread = HandlerThread("DataProcessor").apply { start() }
private val backgroundHandler = Handler(backgroundThread.looper)
private val mainHandler = Handler(Looper.getMainLooper())
fun processData(data: List<String>) {
backgroundHandler.post {
val processed = data.map { processItem(it) }
mainHandler.post {
updateUI(processed)
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
| // Миграция на корутины
suspend fun processData(data: List<String>) {
val processed = withContext(Dispatchers.Default) {
data.map { processItem(it) }
}
withContext(Dispatchers.Main) {
updateUI(processed)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| class MigratedRepository {
private val repositoryScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
// Заменяет ExecutorService
fun startBackgroundWork() {
repositoryScope.launch {
while (isActive) {
processBackgroundTasks()
delay(5000)
}
}
}
fun cleanup() {
repositoryScope.cancel() // отменяет все активные операции
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| // Этап 1: обёртка legacy кода
suspend fun legacyDataLoad(): List<Data> = withContext(Dispatchers.IO) {
// Временно используем старый код
val completableFuture = CompletableFuture<List<Data>>()
legacyDataLoader.loadData { result ->
completableFuture.complete(result)
}
completableFuture.await()
}
// Этап 2: полная замена на корутины
suspend fun modernDataLoad(): List<Data> = withContext(Dispatchers.IO) {
dataRepository.loadData()
} |
|
Пагинация и бесконечная прокрутка через Flow
Пагинация - это та область, где корутины показывают себя во всей красе. До их появления реализация бесконечной прокрутки превращалась в кошмар из колбэков, состояний загрузки и ручного управления потоками. Сейчас, после трёх лет работы с Flow-based пагинацией, могу сказать: это один из самых элегантных способов работы с большими наборами данных.
Основа любой пагинации - состояние загрузки и данные. Flow позволяет объединить их в единый реактивный поток:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
| sealed class PaginationState<T> {
data class Loading<T>(val currentData: List<T> = emptyList()) : PaginationState<T>()
data class Success<T>(val data: List<T>, val hasMore: Boolean) : PaginationState<T>()
data class Error<T>(val currentData: List<T>, val error: Throwable) : PaginationState<T>()
}
class PaginatedRepository<T>(
private val pageSize: Int = 20,
private val dataLoader: suspend (page: Int, size: Int) -> List<T>
) {
private val _paginationState = MutableStateFlow<PaginationState<T>>(PaginationState.Loading())
val paginationState: StateFlow<PaginationState<T>> = _paginationState.asStateFlow()
private var currentPage = 0
private var isLoading = false
suspend fun loadNextPage() {
if (isLoading) return
val currentState = _paginationState.value
val existingData = when (currentState) {
is PaginationState.Success -> currentState.data
is PaginationState.Error -> currentState.currentData
is PaginationState.Loading -> currentState.currentData
}
isLoading = true
_paginationState.value = PaginationState.Loading(existingData)
try {
val newData = dataLoader(currentPage, pageSize)
val allData = existingData + newData
_paginationState.value = PaginationState.Success(
data = allData,
hasMore = newData.size >= pageSize
)
if (newData.isNotEmpty()) {
currentPage++
}
} catch (e: Exception) {
_paginationState.value = PaginationState.Error(existingData, e)
} finally {
isLoading = false
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| class OfflinePaginatedRepository(
private val api: ArticleApiService,
private val dao: ArticleDao
) {
fun observePaginatedArticles(): Flow<PaginationState<Article>> =
dao.observeAllArticles()
.map<List<Article>, PaginationState<Article>> { localArticles ->
if (localArticles.isEmpty()) {
PaginationState.Loading()
} else {
PaginationState.Success(localArticles, hasMore = true)
}
}
.onStart { loadMoreIfNeeded() }
private suspend fun loadMoreIfNeeded() {
val localCount = dao.getArticleCount()
if (localCount == 0 || shouldRefresh()) {
try {
val page = (localCount / 20) + 1
val newArticles = api.getArticles(page = page, size = 20)
dao.insertArticles(newArticles)
} catch (e: Exception) {
// Логируем ошибку, но не прерываем поток данных
}
}
}
suspend fun loadNextPage() {
val currentCount = dao.getArticleCount()
val nextPage = (currentCount / 20) + 1
try {
val newArticles = api.getArticles(page = nextPage, size = 20)
if (newArticles.isNotEmpty()) {
dao.insertArticles(newArticles)
}
} catch (e: Exception) {
throw PaginationException("Failed to load page $nextPage", e)
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| class PreloadingPaginator<T> {
private val preloadThreshold = 5 // загружать когда остается 5 элементов до конца
fun shouldPreload(currentPosition: Int, totalItems: Int): Boolean {
return currentPosition >= totalItems - preloadThreshold
}
fun observePaginatedData(): Flow<List<T>> = combine(
paginationRepository.paginationState,
scrollPositionFlow
) { paginationState, scrollPosition ->
val data = when (paginationState) {
is PaginationState.Success -> paginationState.data
is PaginationState.Error -> paginationState.currentData
is PaginationState.Loading -> paginationState.currentData
}
if (shouldPreload(scrollPosition, data.size) &&
paginationState is PaginationState.Success &&
paginationState.hasMore) {
// Запускаем предварительную загрузку в фоне
GlobalScope.launch { paginationRepository.loadNextPage() }
}
data
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| class ErrorAwarePaginator<T> {
suspend fun loadWithRetry(
maxRetries: Int = 3,
retryDelay: Long = 1000
) {
var attempt = 0
while (attempt < maxRetries) {
try {
loadNextPage()
return
} catch (e: Exception) {
attempt++
if (attempt >= maxRetries) {
val currentData = getCurrentData()
_paginationState.value = if (currentData.isEmpty()) {
PaginationState.Error(emptyList(), e)
} else {
// Для догрузки показываем snackbar, но данные сохраняем
showRetrySnackbar("Failed to load more items")
PaginationState.Success(currentData, hasMore = true)
}
return
}
delay(retryDelay * attempt) // экспоненциальная задержка
}
}
}
} |
|
Retry-механизмы и экспоненциальная задержка
Retry логика - это та область, где я научился не доверять сети и внешним сервисам. За пять лет разработки мобильных приложений понял: если что-то может упасть, оно обязательно упадёт, причём в самый неподходящий момент. Пользователь зайдёт в лифт прямо во время загрузки критичных данных, или сервер ляжет на секунду именно тогда, когда нужно сохранить важную информацию.
Простейший retry механизм выглядит обманчиво просто, но дьявол кроется в деталях. Наивный подход с фиксированными интервалами быстро приводит к проблемам:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
| // Плохой подход - фиксированные интервалы
suspend fun naiveRetry(maxAttempts: Int = 3) {
repeat(maxAttempts) { attempt ->
try {
return apiService.getData()
} catch (e: Exception) {
if (attempt == maxAttempts - 1) throw e
delay(1000) // всегда одна секунда
}
}
} |
|
Экспоненциальная задержка с jitter решает эту проблему элегантно. Каждая следующая попытка происходит через значительно больший интервал, а случайная компонента распределяет нагрузку:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
| class RetryPolicy(
private val maxAttempts: Int = 5,
private val baseDelayMs: Long = 1000,
private val maxDelayMs: Long = 30000,
private val jitterFactor: Double = 0.1
) {
suspend fun <T> executeWithRetry(operation: suspend () -> T): T {
var lastException: Exception? = null
repeat(maxAttempts) { attempt ->
try {
return operation()
} catch (e: CancellationException) {
throw e // не ретраим отмены корутин
} catch (e: Exception) {
lastException = e
if (attempt == maxAttempts - 1) {
throw RetryExhaustedException("All $maxAttempts attempts failed", e)
}
if (!shouldRetry(e)) {
throw e // некоторые ошибки не стоит ретраить
}
val delayMs = calculateDelay(attempt)
delay(delayMs)
}
}
throw lastException ?: IllegalStateException("This should never happen")
}
private fun calculateDelay(attempt: Int): Long {
val exponentialDelay = (baseDelayMs * (2.0.pow(attempt))).toLong()
val cappedDelay = minOf(exponentialDelay, maxDelayMs)
val jitter = (cappedDelay * jitterFactor * Random.nextDouble()).toLong()
return cappedDelay + jitter
}
private fun shouldRetry(exception: Exception): Boolean {
return when (exception) {
is UnknownHostException -> true // проблемы с сетью
is SocketTimeoutException -> true
is ConnectException -> true
is HttpException -> when (exception.code()) {
in 500..599 -> true // серверные ошибки
429 -> true // rate limiting
408 -> true // request timeout
else -> false // клиентские ошибки не ретраим
}
else -> false
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| suspend fun <T> suspend (() -> T).withRetry(
maxAttempts: Int = 3,
delayMs: Long = 1000
): T {
var lastException: Exception? = null
repeat(maxAttempts) { attempt ->
try {
return this()
} catch (e: Exception) {
lastException = e
if (attempt < maxAttempts - 1) {
val delay = delayMs * (attempt + 1)
delay(delay)
}
}
}
throw lastException!!
}
// Использование
suspend fun loadUserData(userId: String): User {
return { userApi.getUser(userId) }.withRetry(maxAttempts = 5)
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| class NetworkRetryManager {
private val quickRetry = RetryPolicy(maxAttempts = 3, baseDelayMs = 500)
private val slowRetry = RetryPolicy(maxAttempts = 5, baseDelayMs = 2000)
private val criticalRetry = RetryPolicy(maxAttempts = 10, baseDelayMs = 1000, maxDelayMs = 60000)
suspend fun <T> executeQuickOperation(operation: suspend () -> T): T {
return quickRetry.executeWithRetry(operation)
}
suspend fun <T> executeSlowOperation(operation: suspend () -> T): T {
return slowRetry.executeWithRetry(operation)
}
suspend fun <T> executeCriticalOperation(operation: suspend () -> T): T {
return criticalRetry.executeWithRetry(operation)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
| class CircuitBreakerRetry(
private val failureThreshold: Int = 5,
private val recoveryTimeMs: Long = 30000
) {
private var failureCount = 0
private var lastFailureTime = 0L
private var state = CircuitState.CLOSED
suspend fun <T> execute(operation: suspend () -> T): T {
when (state) {
CircuitState.OPEN -> {
if (System.currentTimeMillis() - lastFailureTime > recoveryTimeMs) {
state = CircuitState.HALF_OPEN
} else {
throw CircuitBreakerException("Circuit breaker is OPEN")
}
}
CircuitState.HALF_OPEN -> {
return try {
val result = operation()
reset()
result
} catch (e: Exception) {
recordFailure()
throw e
}
}
CircuitState.CLOSED -> {
return try {
operation()
} catch (e: Exception) {
recordFailure()
throw e
}
}
}
return operation()
}
private fun recordFailure() {
failureCount++
lastFailureTime = System.currentTimeMillis()
if (failureCount >= failureThreshold) {
state = CircuitState.OPEN
}
}
private fun reset() {
failureCount = 0
state = CircuitState.CLOSED
}
} |
|
Параллельная загрузка данных с async/await
Параллельная загрузка - это место, где корутины показывают свою настоящую силу. Помню, как раньше пытался организовать одновременные HTTP-запросы через ExecutorService, жонглировал Future объектами и молился, чтобы все потоки завершились без deadlock. С появлением async/await всё изменилось кардинально - параллелизм стал таким же простым, как последовательный код.
Основная идея async/await проста: async запускает корутину и сразу возвращает Deferred объект, await блокирует выполнение до получения результата. Но дьявол, как всегда, в деталях:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| class DashboardService(
private val userApi: UserApiService,
private val newsApi: NewsApiService,
private val weatherApi: WeatherApiService
) {
suspend fun loadDashboard(userId: String): DashboardData = coroutineScope {
val userDeferred = async { userApi.getProfile(userId) }
val newsDeferred = async { newsApi.getLatestNews(userId) }
val weatherDeferred = async { weatherApi.getCurrentWeather(userId) }
// Все три запроса выполняются параллельно
// await блокирует только до завершения конкретной операции
DashboardData(
user = userDeferred.await(),
news = newsDeferred.await(),
weather = weatherDeferred.await()
)
}
} |
|
Ключевое отличие async от launch - async возвращает результат. Если результат не нужен, используйте launch. Если нужен - async с последующим await:
| Kotlin | 1
2
3
4
5
6
| // Неправильно - launch не возвращает результат
val job = launch { userApi.getProfile(userId) } // результат потерян
// Правильно - async возвращает Deferred<User>
val deferred = async { userApi.getProfile(userId) }
val user = deferred.await() |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| suspend fun loadWithErrorHandling(): DashboardData = coroutineScope {
val userResult = async {
try { userApi.getProfile(userId) }
catch (e: Exception) { User.empty() }
}
val newsResult = async {
try { newsApi.getLatestNews() }
catch (e: Exception) { emptyList<News>() }
}
DashboardData(
user = userResult.await(),
news = newsResult.await()
)
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
| suspend fun loadIndependentData(): PartialDashboard = supervisorScope {
val userJob = async { loadUserData() }
val newsJob = async { loadNewsData() }
val weatherJob = async { loadWeatherData() }
PartialDashboard(
user = try { userJob.await() } catch (e: Exception) { null },
news = try { newsJob.await() } catch (e: Exception) { emptyList() },
weather = try { weatherJob.await() } catch (e: Exception) { null }
)
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| suspend fun loadMultipleUsers(userIds: List<String>): List<User> = coroutineScope {
userIds.map { userId ->
async { userApi.getUser(userId) }
}.awaitAll()
}
// Или с фильтрацией результатов
suspend fun loadValidUsers(userIds: List<String>): List<User> = coroutineScope {
userIds.map { userId ->
async {
try { userApi.getUser(userId) }
catch (e: Exception) { null }
}
}.awaitAll().filterNotNull()
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
| class ConcurrencyLimitedLoader {
private val semaphore = Semaphore(5) // максимум 5 одновременных запросов
suspend fun loadManyUsers(userIds: List<String>): List<User> = coroutineScope {
userIds.map { userId ->
async {
semaphore.withPermit {
userApi.getUser(userId)
}
}
}.awaitAll()
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
| suspend fun loadWithTimeout(): DashboardData? = withTimeoutOrNull(5000) {
coroutineScope {
val user = async { userApi.getProfile() }
val news = async { newsApi.getLatestNews() }
DashboardData(user.await(), news.await())
}
} |
|
Обработка файловых операций и потоков данных
Файловые операции с корутинами - это область, где я сталкивался с самыми коварными проблемами производительности. Блокирование UI потока чтением больших файлов, утечки памяти при работе с потоками, падения приложений при обработке поврежденных данных. После четырёх лет оптимизации файловых операций в продакшене могу сказать: корутины превращают эту сложную область в управляемый процесс.
Основная ошибка новичков - попытка читать файлы целиком в память. Для небольших конфигурационных файлов это работает, но стоит пользователю выбрать видео на несколько гигабайт, и приложение рухнет с OutOfMemoryError:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| // Плохо - загружает весь файл в память
suspend fun readFileWrong(file: File): String = withContext(Dispatchers.IO) {
file.readText() // может съесть всю память
}
// Правильно - потоковое чтение с буферизацией
suspend fun readFileSafely(file: File): Flow<String> = flow {
file.bufferedReader().use { reader ->
var line: String?
while (reader.readLine().also { line = it } != null) {
emit(line!!)
}
}
}.flowOn(Dispatchers.IO) |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| class LogFileProcessor {
suspend fun processLargeLogFile(logFile: File): ProcessingResult =
logFile.bufferedReader().use { reader ->
reader.lineSequence()
.asFlow()
.map { line -> parseLogEntry(line) }
.filter { entry -> entry.isError() }
.chunked(100) // обрабатываем пакетами по 100 записей
.collect { errorBatch ->
database.insertErrors(errorBatch)
yield() // даём другим корутинам возможность выполниться
}
}
private suspend fun parseLogEntry(line: String): LogEntry = withContext(Dispatchers.Default) {
// CPU-интенсивный парсинг в Default диспетчере
LogEntry.parse(line)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| class FileManager {
suspend fun copyFileWithProgress(
source: File,
destination: File,
onProgress: (bytesTransferred: Long, totalBytes: Long) -> Unit
) = withContext(Dispatchers.IO) {
source.inputStream().use { input ->
destination.outputStream().use { output ->
val totalBytes = source.length()
var bytesTransferred = 0L
val buffer = ByteArray(8192)
while (isActive) {
val bytesRead = input.read(buffer)
if (bytesRead == -1) break
output.write(buffer, 0, bytesRead)
bytesTransferred += bytesRead
withContext(Dispatchers.Main) {
onProgress(bytesTransferred, totalBytes)
}
if (bytesTransferred % (1024 * 1024) == 0L) {
yield() // проверяем отмену каждый мегабайт
}
}
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| class JsonStreamProcessor {
suspend fun processLargeJsonArray(file: File): Flow<ProcessedItem> = flow {
file.inputStream().use { input ->
val reader = JsonReader(InputStreamReader(input))
reader.beginArray()
while (reader.hasNext()) {
ensureActive() // проверяем отмену
val item = gson.fromJson<RawItem>(reader, RawItem::class.java)
val processed = processItem(item)
emit(processed)
}
reader.endArray()
}
}.flowOn(Dispatchers.IO)
private suspend fun processItem(item: RawItem): ProcessedItem = withContext(Dispatchers.Default) {
// Тяжёлая обработка в background потоке
ProcessedItem(item.transform())
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| class BinaryFileProcessor {
suspend fun processVideoFile(videoFile: File) = coroutineScope {
val frameChannel = Channel<VideoFrame>(capacity = 10)
// Producer корутина - читает фреймы
val readerJob = launch(Dispatchers.IO) {
videoFile.inputStream().use { input ->
val buffer = ByteArray(frameSize)
while (input.read(buffer) != -1) {
val frame = VideoFrame.fromBytes(buffer)
frameChannel.send(frame)
}
}
frameChannel.close()
}
// Consumer корутина - обрабатывает фреймы
launch(Dispatchers.Default) {
for (frame in frameChannel) {
val processed = applyVideoFilter(frame)
saveProcessedFrame(processed)
}
}
readerJob.join()
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| suspend fun extractAndProcess(zipFile: File) = withContext(Dispatchers.IO) {
ZipInputStream(zipFile.inputStream()).use { zipStream ->
generateSequence { zipStream.nextEntry }
.filterNot { it.isDirectory }
.asFlow()
.map { entry ->
async(Dispatchers.Default) {
val content = zipStream.readBytes()
ProcessedEntry(entry.name, processContent(content))
}
}
.buffer(5) // ограничиваем параллелизм
.map { it.await() }
.collect { processedEntry ->
saveToDatabase(processedEntry)
}
}
} |
|
Интеграция с Firebase и облачными сервисами
Firebase с корутинами - это сочетание, которое кардинально изменило мой подход к разработке мобильных приложений. Когда Google добавил поддержку корутин в Firebase SDK, исчезла необходимость в бесконечных колбэках и сложных цепочках асинхронных операций. Теперь работа с облачными сервисами выглядит так же просто, как обычные функции. Firestore Database предоставляет extension функции для корутин, которые превращают callback-based API в элегантные suspend методы:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| class FirestoreRepository {
private val firestore = FirebaseFirestore.getInstance()
suspend fun saveUser(user: User): String =
firestore.collection("users")
.add(user)
.await()
.id
suspend fun getUser(userId: String): User? =
firestore.collection("users")
.document(userId)
.get()
.await()
.toObject<User>()
suspend fun updateUserField(userId: String, field: String, value: Any) {
firestore.collection("users")
.document(userId)
.update(field, value)
.await()
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| fun observeUserPosts(userId: String): Flow<List<Post>> = callbackFlow {
val listener = firestore.collection("posts")
.whereEqualTo("authorId", userId)
.orderBy("timestamp", Query.Direction.DESCENDING)
.addSnapshotListener { snapshot, error ->
if (error != null) {
close(error)
return@addSnapshotListener
}
val posts = snapshot?.documents?.mapNotNull {
it.toObject<Post>()
} ?: emptyList()
trySend(posts)
}
awaitClose { listener.remove() }
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| class AuthRepository {
private val auth = FirebaseAuth.getInstance()
suspend fun signInWithEmail(email: String, password: String): AuthResult =
auth.signInWithEmailAndPassword(email, password).await()
suspend fun createAccount(email: String, password: String): AuthResult =
auth.createUserWithEmailAndPassword(email, password).await()
fun observeAuthState(): Flow<FirebaseUser?> = callbackFlow {
val listener = auth.addAuthStateListener { auth ->
trySend(auth.currentUser)
}
awaitClose { auth.removeAuthStateListener(listener) }
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| class CloudFunctionsService {
private val functions = FirebaseFunctions.getInstance()
suspend fun processPayment(paymentData: PaymentData): PaymentResult {
val data = hashMapOf("payment" to paymentData.toMap())
return functions
.getHttpsCallable("processPayment")
.call(data)
.await()
.data as PaymentResult
}
suspend fun sendNotification(userId: String, message: String) {
val data = mapOf(
"userId" to userId,
"message" to message
)
functions
.getHttpsCallable("sendNotification")
.call(data)
.await()
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| class StorageRepository {
private val storage = FirebaseStorage.getInstance()
suspend fun uploadFileWithProgress(
file: File,
path: String,
onProgress: (bytesTransferred: Long, totalBytes: Long) -> Unit
): String = suspendCancellableCoroutine { continuation ->
val ref = storage.reference.child(path)
val uploadTask = ref.putFile(file.toUri())
uploadTask
.addOnProgressListener { snapshot ->
onProgress(snapshot.bytesTransferred, snapshot.totalByteCount)
}
.addOnSuccessListener {
continuation.resume(ref.path)
}
.addOnFailureListener { exception ->
continuation.resumeWithException(exception)
}
continuation.invokeOnCancellation {
uploadTask.cancel()
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| suspend fun createUserProfile(
email: String,
password: String,
profileData: ProfileData,
avatarFile: File?
): UserProfile = coroutineScope {
// Параллельно создаём аккаунт и загружаем аватар
val authResult = async { authRepository.createAccount(email, password) }
val avatarUrl = avatarFile?.let {
async { storageRepository.uploadFile(it, "avatars/${UUID.randomUUID()}") }
}
val user = authResult.await().user!!
val profile = profileData.copy(
id = user.uid,
avatarUrl = avatarUrl?.await()
)
// Сохраняем профиль в Firestore
firestoreRepository.saveUser(profile)
profile
} |
|
Подводные камни и решения
После пяти лет активного использования корутин в продакшене я накопил внушительную коллекцию "граблей", на которые наступал сам и которые регулярно встречаю в code review коллег. Корутины кажутся простыми на поверхности, но под капотом скрывается множество тонкостей, которые могут превратить элегантный асинхронный код в источник трудноотлавливаемых багов. Самая коварная ловушка - это ложное ощущение безопасности. Когда переходишь с Thread и AsyncTask на корутины, кажется, что все проблемы concurrency волшебным образом исчезли. На самом деле они просто стали менее очевидными. Race conditions, deadlock и memory leaks никуда не делись - они просто приобрели новые формы.
Классический пример - неправильное использование GlobalScope. Я видел проекты, где разработчики бездумно запускали корутины через GlobalScope.launch, не понимая последствий. Такие корутины живут весь жизненный цикл приложения и могут продолжать работу даже после закрытия экранов, потребляя ресурсы и вызывая крэши.
| Kotlin | 1
2
3
4
5
6
7
8
9
| // Плохо - создаёт неконтролируемые корутины
class BadViewModel : ViewModel() {
fun loadData() {
GlobalScope.launch { // неправильно!
val data = repository.getData()
updateUI(data) // может вызвать крэш если UI уничтожен
}
}
} |
|
Ещё одна распространённая ошибка - попытка вызывать suspend функции из неподходящих мест. Новички часто пытаются использовать suspend методы в конструкторах, init блоках или обычных функциях, получая ошибки компиляции и не понимая, почему "простой" код не работает.
Memory leaks в ViewModels
Утечки памяти в ViewModels с корутинами - это проблема, которая преследовала меня первые два года работы с новой архитектурой. Самое коварное в этих утечках то, что они проявляются не сразу. Приложение может работать месяцами без видимых проблем, а потом внезапно начать падать с OutOfMemoryError в самых неожиданных местах.
Корень зла - неправильное понимание жизненного цикла ViewModel и способов запуска корутин. Когда пользователь поворачивает экран или переходит между фрагментами, Android создаёт новые экземпляры UI компонентов, но ViewModel может оставаться в памяти. Если в старой ViewModel продолжают работать корутины, держащие ссылки на View или Context, получается классическая утечка памяти.
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| // Опасный код - создаёт утечку памяти
class LeakyViewModel(private val context: Context) : ViewModel() {
fun startBackgroundWork() {
// Корутина держит ссылку на context через замыкание
GlobalScope.launch {
while (true) {
val data = heavyComputation(context) // утечка!
delay(1000)
}
}
}
private fun heavyComputation(context: Context): String {
// Использование context в долгоживущей корутине
return context.getString(R.string.computed_value)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| class SafeViewModel(
private val repository: DataRepository
) : ViewModel() {
fun startBackgroundWork() {
viewModelScope.launch {
while (isActive) {
val data = repository.computeData() // безопасно
_dataState.value = data
delay(1000)
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| // Плохо - ViewModel держит ссылку на View
class BadViewModel : ViewModel() {
private var currentView: View? = null
fun setView(view: View) {
currentView = view // потенциальная утечка
}
fun updateView() {
viewModelScope.launch {
val result = repository.getData()
currentView?.let { view ->
// View может быть уничтожен, но ссылка остаётся
updateUI(view, result)
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| class FlowLeakViewModel : ViewModel() {
// Проблемный подход
fun getDataStream(): Flow<Data> = flow {
while (true) {
emit(repository.getCurrentData())
delay(5000)
}
}.flowOn(Dispatchers.IO) // каждый подписчик создаёт новый поток
// Правильный подход
private val dataStream = repository.observeData()
.shareIn(
scope = viewModelScope,
started = SharingStarted.WhileSubscribed(5000),
replay = 1
)
fun getDataStream(): SharedFlow<Data> = dataStream
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| class TaskManagerViewModel : ViewModel() {
private val runningTasks = mutableListOf<Job>()
fun startTask() {
val job = viewModelScope.launch {
processLongRunningTask()
}
runningTasks.add(job) // потенциальная утечка
}
override fun onCleared() {
super.onCleared()
runningTasks.forEach { it.cancel() }
runningTasks.clear()
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
| class BetterTaskManagerViewModel : ViewModel() {
fun startTask() {
viewModelScope.launch {
try {
processLongRunningTask()
} finally {
// cleanup происходит автоматически при onCleared()
}
}
}
} |
|
Отмена операций при закрытии экрана
Отмена операций при закрытии экрана - это область, где я научился не доверять автоматике Android системы. Пользователи имеют привычку закрывать экраны в самые неподходящие моменты: во время загрузки критичных данных, в процессе отправки форм, при выполнении тяжёлых вычислений. Если не отменить эти операции корректно, они продолжат выполняться в фоне, потребляя ресурсы и иногда вызывая крэши при попытке обновить несуществующий UI.
Самая распространённая ошибка - использование applicationContext или GlobalScope для операций, связанных с конкретным экраном. Такие корутины живут независимо от UI lifecycle и могут работать даже после уничтожения Activity:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
| // Плохой подход - операция не отменится при закрытии экрана
class BadActivity : AppCompatActivity() {
fun loadData() {
GlobalScope.launch {
val data = repository.loadHeavyData()
runOnUiThread {
updateUI(data) // может вызвать крэш если Activity уничтожена
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
| class SafeActivity : AppCompatActivity() {
fun loadData() {
lifecycleScope.launch {
try {
val data = repository.loadHeavyData()
updateUI(data) // безопасно - корутина отменится с Activity
} catch (e: CancellationException) {
// Корутина была отменена - это нормально
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
| class DataFragment : Fragment() {
override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
super.onViewCreated(view, savedInstanceState)
// Привязываем к lifecycle View, а не Fragment
viewLifecycleOwner.lifecycleScope.launch {
repository.observeData().collect { data ->
updateRecyclerView(data) // безопасно - отменится при onDestroyView
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| class DataViewModel : ViewModel() {
private val _uiState = MutableStateFlow<UiState>(UiState.Loading)
val uiState = _uiState.asStateFlow()
fun loadData() {
viewModelScope.launch {
try {
val data = repository.getData()
_uiState.value = UiState.Success(data)
} catch (e: Exception) {
_uiState.value = UiState.Error(e.message)
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| class FileUploadManager {
private val uploadScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
fun uploadFile(file: File): Job {
return uploadScope.launch {
try {
val result = fileService.upload(file)
notificationManager.showUploadSuccess(result)
} catch (e: CancellationException) {
// Отмена загрузки - удаляем частично загруженный файл
cleanupPartialUpload(file)
throw e
}
}
}
fun cancelAllUploads() {
uploadScope.cancel()
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| class ComplexDataScreen : Fragment() {
private val screenScope = CoroutineScope(SupervisorJob())
override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
super.onViewCreated(view, savedInstanceState)
// Привязано к View lifecycle
viewLifecycleOwner.lifecycleScope.launch {
viewModel.uiState.collect { state ->
updateUI(state)
}
}
// Независимо от View, но завершается при onDestroy Fragment
screenScope.launch {
backgroundDataSync()
}
}
override fun onDestroy() {
super.onDestroy()
screenScope.cancel()
}
} |
|
Тестирование асинхронного кода
Тестирование корутин - это кошмар, который я переживал первые полгода работы с новой архитектурой. Привычные unit тесты превращались в flaky кошмар, где результат зависел от скорости выполнения машины и фазы луны. Асинхронные операции завершались в непредсказуемом порядке, тесты падали с timeout ошибками, а mock объекты не успевали срабатывать до проверки результатов.
Основная проблема - обычные тесты не умеют ждать завершения корутин. JUnit запускает тест, видит что функция вернула результат, и считает тест завершенным. А корутина может продолжать работать в фоне:
| Kotlin | 1
2
3
4
5
6
| // Неправильно - тест завершится до окончания корутины
@Test
fun badAsyncTest() {
viewModel.loadData() // запускает корутину и сразу возвращается
assertEquals(expectedData, viewModel.data.value) // падает - данные ещё не загружены
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
| @Test
fun correctAsyncTest() = runTest {
val repository = FakeRepository()
val viewModel = DataViewModel(repository)
viewModel.loadData()
// Дожидаемся завершения всех корутин
assertEquals(expectedData, viewModel.data.value)
} |
|
StandardTestDispatcher выполняет корутины только при явном вызове, UnconfinedTestDispatcher выполняет их немедленно:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| class ViewModelTest {
private val testDispatcher = StandardTestDispatcher()
private val testScope = TestScope(testDispatcher)
@Test
fun testDataLoading() = testScope.runTest {
val viewModel = DataViewModel(testDispatcher)
viewModel.loadData()
// Корутина ещё не выполнилась
assertEquals(LoadingState, viewModel.state.value)
// Принудительно выполняем все pending корутины
advanceUntilIdle()
assertEquals(LoadedState, viewModel.state.value)
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| @Test
fun testFlowOperations() = runTest {
val repository = FakeRepository()
val results = mutableListOf<DataState>()
val job = launch {
repository.observeData()
.take(3) // ограничиваем количество значений
.collect { results.add(it) }
}
repository.emitData(firstData)
repository.emitData(secondData)
repository.emitData(thirdData)
job.join()
assertEquals(3, results.size)
assertEquals(firstData, results[0])
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| class RepositoryTest {
@Mock
private lateinit var apiService: ApiService
@Test
fun testWithDelay() = runTest {
// Мокаем с задержкой
whenever(apiService.getData()).thenSuspend {
delay(1000)
testData
}
val repository = DataRepository(apiService)
val startTime = currentTime
val result = repository.getData()
assertEquals(testData, result)
assertEquals(1000, currentTime - startTime) // проверяем что delay сработал
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| @Test
fun testErrorHandling() = runTest {
val failingRepository = object : Repository {
override suspend fun getData() = throw NetworkException("Network failed")
}
val viewModel = DataViewModel(failingRepository)
viewModel.loadData()
advanceUntilIdle()
val state = viewModel.state.value
assertTrue(state is ErrorState)
assertEquals("Network failed", (state as ErrorState).message)
} |
|
TestScope.backgroundScope для разделения основных и фоновых операций:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| @Test
fun testComplexScenario() = runTest {
val viewModel = ComplexViewModel()
// Основная операция
viewModel.startMainTask()
// Фоновые операции
backgroundScope.launch {
repeat(10) {
viewModel.backgroundUpdate()
delay(100)
}
}
advanceTimeBy(1000) // прокручиваем время на секунду
assertEquals(10, viewModel.backgroundUpdateCount)
assertTrue(viewModel.isMainTaskCompleted)
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
| @Test
fun testCancellation() = runTest {
val repository = spy(TestRepository())
val job = launch {
repository.longRunningOperation()
}
advanceTimeBy(500)
job.cancel()
verify(repository).cleanup() // проверяем что cleanup вызван
} |
|
Проблемы с GlobalScope и их альтернативы
GlobalScope - это самая опасная ловушка в мире корутин, в которую попадают 90% разработчиков на начальном этапе изучения. Помню, как сам активно использовал его первые месяцы, наивно полагая, что раз Google предоставил такой удобный API, значит им можно пользоваться везде. Результат был предсказуем - приложения начали течь память, корутины работали после закрытия экранов, а отладка превратилась в кошмар.
Основная проблема GlobalScope в том, что он живёт весь жизненный цикл приложения. Корутины, запущенные через GlobalScope.launch, продолжают выполняться даже после уничтожения Activity, Fragment или ViewModel. Это приводит к множеству критичных проблем:
| Kotlin | 1
2
3
4
5
6
7
8
9
| // Катастрофически неправильный код
class BadViewModel : ViewModel() {
fun loadUserData() {
GlobalScope.launch { // зло!
val userData = userRepository.getData()
_userState.value = UserState.Success(userData) // может вызвать крэш
}
}
} |
|
Ещё хуже становится, когда в GlobalScope корутинах используются ссылки на Android компоненты. Context, View, Fragment - все эти объекты могут остаться в памяти гораздо дольше их естественного жизненного цикла:
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| // Утечка памяти гарантирована
class LeakyActivity : AppCompatActivity() {
fun startBackgroundWork() {
GlobalScope.launch {
while (true) {
val result = heavyComputation()
runOnUiThread {
updateUI(result) // Activity может быть уничтожена!
}
delay(5000)
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| class SafeViewModel : ViewModel() {
fun loadUserData() {
viewModelScope.launch {
try {
val userData = userRepository.getData()
_userState.value = UserState.Success(userData)
} catch (e: CancellationException) {
// Корректная отмена - ничего не делаем
throw e
} catch (e: Exception) {
_userState.value = UserState.Error(e.message)
}
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
| class SafeActivity : AppCompatActivity() {
fun loadData() {
lifecycleScope.launch {
val data = repository.getData()
updateUI(data) // безопасно - корутина отменится с Activity
}
}
} |
|
| Kotlin | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| class BackgroundSyncManager {
private val syncScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
fun startPeriodicSync() {
syncScope.launch {
while (isActive) {
try {
syncData()
delay(TimeUnit.HOURS.toMillis(1))
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
// Логируем и продолжаем
}
}
}
}
fun stopSync() {
syncScope.cancel() // явно останавливаем все операции
}
} |
|
| Kotlin | 1
2
3
4
5
| suspend fun processMultipleFiles(files: List<File>) = coroutineScope {
files.map { file ->
async { processFile(file) }
}.awaitAll()
} |
|
GlobalScope имеет очень ограниченную область применения - только для операций, которые должны продолжаться весь жизненный цикл приложения и не держат ссылки на UI компоненты. В 99% случаев существует лучшая альтернатива.
Как перетащить мусор в корзину? Drag 'n drop в Kotlin
Здравствуйте.
Есть мини-игра, где надо перетащить клочок бумаги в корзину. Есть код, который вроде... [Kotlin] Как узнать и сравнить положение двух imageView?
Есть две картинки: клочок бумаги и корзина. Как сделать так, чтобы бумага исчезала, когда ее... Как учить Kotlin и стоит ли?
Добрый день, вообще я фрилансер пишу сайты на таких технологиях HTML, CSS, jQuery, WordPress,... Как последовательно выполнять методы используя kotlin coroutines?
У меня есть картинки в recyclerview по нажатию на которые выполняется запрос к апи, при чём они... firebase database как получать сообщения и выводить их на экран на kotlin
Не понимаю что нужно писать в onClickListener, чтобы при нажатии на кнопку брался текст с базы... Как запускать параллельные таймеры (kotlin)?
Программа для раздельного старта спортсменов. Есть EditText для количества участников, по нему... [Kotlin] Как заставить Service работать, даже если приложение выгружено из списка задач?
Здравствуйте. Пишу приложение для напоминаний. Пользователь вводит текст, который хочет вывести,... Как встроить admob в kotlin проект?
Есть проект созданный в android studio на kotlin, при встраивание admob проект собирается но при... Как использовать RxJava2, Retrofit2 в Android Kotlin
Здравствуйте. Пишу приложение, где нужно прочитать JSON файл из интернета и вывести на экран.... Как исправить ошибку Can't create handler inside thread that has not called Looper.prepare() на Kotlin?
У меня есть activity в котором я подключаюсь к веб-сокету. Он должен присылать сообщения с помощью... Kotlin приложение для просмотра картинок. Как зациклить этот адаптер?
Здравствуйте.
Не хватает знаний/мозгов как в идеале сделать "карусель" из картинок (т.е. чтобы... Как преобразовать Нейросеть в Kotlin из Python?
Код на Python
import numpy as np
def sigmoid(x): return 1 / (1 + np.exp(-x)) # создаем...
|
