git.iko.soy/parseable-android
on 0da331f help sign in
parseable-android/app/src/main/java/com/parseable/android/ui/screens/streams/StreamsViewModel.kt
180 lines 6.5 kB history raw
1package com.parseable.android.ui.screens.streams
3import androidx.lifecycle.ViewModel
4import androidx.lifecycle.viewModelScope
5import com.parseable.android.data.local.FavoriteStream
6import com.parseable.android.data.local.FavoriteStreamDao
7import com.parseable.android.data.model.AboutInfo
8import com.parseable.android.data.model.ApiResult
9import com.parseable.android.data.model.LogStream
10import com.parseable.android.data.repository.ParseableRepository
11import com.parseable.android.data.repository.SettingsRepository
12import dagger.hilt.android.lifecycle.HiltViewModel
13import kotlinx.coroutines.CancellationException
14import kotlinx.coroutines.Job
15import kotlinx.coroutines.async
16import kotlinx.coroutines.coroutineScope
17import kotlinx.coroutines.channels.Channel
18import kotlinx.coroutines.flow.MutableStateFlow
19import kotlinx.coroutines.flow.StateFlow
20import kotlinx.coroutines.flow.asStateFlow
21import kotlinx.coroutines.flow.receiveAsFlow
22import kotlinx.coroutines.flow.update
23import kotlinx.coroutines.launch
24import kotlinx.coroutines.sync.Semaphore
25import kotlinx.coroutines.sync.withPermit
26import javax.inject.Inject
28data class StreamsState(
29 val streams: List<LogStream> = emptyList(),
30 val streamStats: Map<String, StreamsViewModel.StreamStatsUi> = emptyMap(),
31 val failedStats: Set<String> = emptySet(),
32 val aboutInfo: AboutInfo? = null,
33 val isLoading: Boolean = false,
34 val error: String? = null,
35 val favoriteNames: Set<String> = emptySet(),
38@HiltViewModel
39class StreamsViewModel @Inject constructor(
40 private val repository: ParseableRepository,
41 private val settingsRepository: SettingsRepository,
42 private val favoriteDao: FavoriteStreamDao,
43) : ViewModel() {
45 data class StreamStatsUi(
46 val eventCount: Long? = null,
47 val ingestionSize: String? = null,
48 val storageSize: String? = null,
49 )
51 private val _state = MutableStateFlow(StreamsState())
52 val state: StateFlow<StreamsState> = _state.asStateFlow()
54 private val _snackbarEvent = Channel<String>(Channel.BUFFERED)
55 val snackbarEvent = _snackbarEvent.receiveAsFlow()
57 init {
58 viewModelScope.launch {
59 favoriteDao.getAllNames().collect { names ->
60 _state.update { it.copy(favoriteNames = names.toSet()) }
61 }
62 }
63 }
65 fun toggleFavorite(streamName: String) {
66 viewModelScope.launch {
67 if (streamName in _state.value.favoriteNames) {
68 favoriteDao.deleteByName(streamName)
69 _snackbarEvent.send("Removed from favorites")
70 } else {
71 favoriteDao.insert(FavoriteStream(streamName = streamName))
72 _snackbarEvent.send("Added to favorites")
73 }
74 }
75 }
77 fun refresh() {
78 viewModelScope.launch {
79 _state.update { it.copy(isLoading = true, error = null) }
81 try {
82 // Fetch about info and streams in parallel
83 val (aboutResult, streamsResult) = coroutineScope {
84 val aboutDeferred = async { repository.getAbout() }
85 val streamsDeferred = async { repository.listStreams(forceRefresh = true) }
86 aboutDeferred.await() to streamsDeferred.await()
87 }
89 if (aboutResult is ApiResult.Success) {
90 _state.update { it.copy(aboutInfo = aboutResult.data) }
91 }
93 when (streamsResult) {
94 is ApiResult.Success -> {
95 _state.update {
96 it.copy(
97 streams = streamsResult.data,
98 isLoading = false,
99 )
100 }
101 // Load stats for each stream in parallel
102 loadStreamStats(streamsResult.data)
103 }
104 is ApiResult.Error -> {
105 _state.update {
106 it.copy(
107 isLoading = false,
108 error = streamsResult.userMessage,
109 )
110 }
111 }
112 }
113 } catch (e: CancellationException) {
114 _state.update { it.copy(isLoading = false) }
115 throw e
116 } catch (e: Exception) {
117 _state.update {
118 it.copy(isLoading = false, error = e.message ?: "Failed to load streams")
119 }
120 }
121 }
122 }
124 private val statsSemaphore = Semaphore(8)
125 private var statsJob: Job? = null
127 private fun loadStreamStats(streams: List<LogStream>) {
128 statsJob?.cancel()
129 statsJob = viewModelScope.launch {
130 streams.forEach { stream ->
131 launch {
132 statsSemaphore.withPermit {
133 loadSingleStreamStats(stream.name)
134 }
135 }
136 }
137 }
138 }
140 fun retryStats(streamName: String) {
141 viewModelScope.launch {
142 _state.update { it.copy(failedStats = it.failedStats - streamName) }
143 statsSemaphore.withPermit {
144 loadSingleStreamStats(streamName)
145 }
146 }
147 }
149 private suspend fun loadSingleStreamStats(streamName: String) {
150 when (val result = repository.getStreamStats(streamName)) {
151 is ApiResult.Success -> {
152 val stats = result.data
153 val ui = StreamStatsUi(
154 eventCount = stats.ingestion?.count
155 ?: stats.ingestion?.lifetimeCount,
156 ingestionSize = stats.ingestion?.size
157 ?: stats.ingestion?.lifetimeSize,
158 storageSize = stats.storage?.size
159 ?: stats.storage?.lifetimeSize,
160 )
161 _state.update {
162 it.copy(
163 streamStats = it.streamStats + (streamName to ui),
164 failedStats = it.failedStats - streamName,
165 )
166 }
167 }
168 is ApiResult.Error -> {
169 _state.update {
170 it.copy(failedStats = it.failedStats + streamName)
171 }
172 }
173 }
174 }
176 suspend fun logout() {
177 settingsRepository.clearConfig()
178 repository.clearCredentials()
179 }