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