Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Little improvements, debug statements and cleanup #57

Merged
merged 1 commit into from
Aug 30, 2024
Merged

Little improvements, debug statements and cleanup #57

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions sync-core/src/main/java/eu/darken/octi/sync/core/SyncManager.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ class SyncManager @Inject constructor(
log(TAG) { "sync(options=$options)" }
val syncJobs = connectors.first().map {
scope.launch {
// TODO error handling
sync(it.identifier, options = options)
}
}
Expand Down
58 changes: 34 additions & 24 deletions syncs-gdrive/src/main/java/eu/darken/octi/syncs/gdrive/core/GDriveAppDataConnector.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import eu.darken.octi.sync.core.SyncConnector
import eu.darken.octi.sync.core.SyncConnectorState
import eu.darken.octi.sync.core.SyncOptions
import eu.darken.octi.sync.core.SyncRead
import eu.darken.octi.sync.core.SyncSettings
import eu.darken.octi.sync.core.SyncWrite
import eu.darken.octi.syncs.gdrive.core.GDriveEnvironment.Companion.APPDATAFOLDER
import kotlinx.coroutines.CoroutineScope
Expand All @@ -45,15 +46,17 @@ import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import okio.IOException
import java.time.Instant
import com.google.api.services.drive.model.File as GDriveFile


class GDriveAppDataConnector @AssistedInject constructor(
@Assisted private val client: GoogleClient,
@AppScope private val scope: CoroutineScope,
private val dispatcherProvider: DispatcherProvider,
dispatcherProvider: DispatcherProvider,
@ApplicationContext private val context: Context,
private val networkStateProvider: NetworkStateProvider,
private val supportedModuleIds: Set<@JvmSuppressWildcards ModuleId>,
private val syncSettings: SyncSettings,
) : GDriveBaseConnector(dispatcherProvider, context, client), SyncConnector {

data class State(
Expand Down Expand Up @@ -110,10 +113,12 @@ class GDriveAppDataConnector @AssistedInject constructor(
override suspend fun resetData(): Unit = withContext(NonCancellable) {
log(TAG, INFO) { "resetData()" }
runDriveAction("reset-data") {
appDataRoot
.listFiles()
.forEach { it.deleteAll() }
_state.updateBlocking { copy(isDead = true) }
appDataRoot.child(DEVICE_DATA_DIR_NAME)
?.listFiles()
?.forEach { file: GDriveFile ->
log(TAG, INFO) { "resetData(): Deleting $file" }
file.deleteAll()
}
}
}

Expand All @@ -124,9 +129,12 @@ class GDriveAppDataConnector @AssistedInject constructor(
?.listFiles()
?.onEach { log(TAG, DEBUG) { "deleteDevice(): Checking $it" } }
?.singleOrNull { it.name == deviceId.id }
?.onEach { log(TAG, WARN) { "deleteDevice(): Deleting $it" } }
?.onEach { log(TAG, INFO) { "deleteDevice(): Deleting $it" } }
?.deleteAll()
_state.updateBlocking { copy(isDead = true) }
if (deviceId == syncSettings.deviceId) {
log(TAG, WARN) { "We just deleted ourselves, this connector is dead now" }
_state.updateBlocking { copy(isDead = true) }
}
}
}

Expand All @@ -153,12 +161,11 @@ class GDriveAppDataConnector @AssistedInject constructor(

val deviceFetchJobs = validDeviceDirs.map { deviceDir ->
scope.async deviceFetch@{
log(TAG, DEBUG) { "readDrive(): Reading module data for device: $deviceDir" }
val moduleDirs = deviceDir.listFiles().filter { supportedModuleIds.contains(ModuleId(it.name)) }

log(TAG, VERBOSE) { "readDrive(): Reading module data for device: $deviceDir" }

val moduleFetchJobs = moduleDirs.map { moduleFile ->
scope.async moduleFetch@{
log(TAG, VERBOSE) { "readDrive(): Reading ${moduleFile.name} for ${deviceDir.name}" }
val payload = moduleFile.readData()

if (payload == null) {
Expand All @@ -172,11 +179,12 @@ class GDriveAppDataConnector @AssistedInject constructor(
moduleId = ModuleId(moduleFile.name),
modifiedAt = Instant.ofEpochMilli(moduleFile.modifiedTime.value),
payload = payload,
).also { log(TAG, VERBOSE) { "readDrive(): Module data: $it" } }
).also { log(TAG, VERBOSE) { "readDrive(): Got module data: $it" } }
}
}

val moduleData = moduleFetchJobs.awaitAll().filterNotNull()
log(TAG, DEBUG) { "readDrive(): Finished ${deviceDir.name}" }

GDriveDeviceData(
deviceId = DeviceId(deviceDir.name),
Expand All @@ -186,8 +194,8 @@ class GDriveAppDataConnector @AssistedInject constructor(
}

val devices = deviceFetchJobs.awaitAll()

log(TAG) { "readDrive() took ${System.currentTimeMillis() - start}ms" }

return GDriveData(
connectorId = identifier,
devices = devices,
Expand Down Expand Up @@ -246,7 +254,6 @@ class GDriveAppDataConnector @AssistedInject constructor(
}
} catch (e: Exception) {
log(TAG, ERROR) { "sync(): Failed to read: ${e.asLog()}" }
_state.updateBlocking { copy(lastError = e) }
}
}

Expand All @@ -257,7 +264,7 @@ class GDriveAppDataConnector @AssistedInject constructor(
}

private suspend fun GDriveEnvironment.writeDrive(data: SyncWrite) = withContext(NonCancellable) {
log(TAG, DEBUG) { "writeDrive(): $data)" }
log(TAG, DEBUG) { "writeDrive(): $data" }

// TODO cache write data for when we are online again?
if (!isInternetAvailable()) {
Expand All @@ -268,8 +275,9 @@ class GDriveAppDataConnector @AssistedInject constructor(
val userDir = appDataRoot.child(DEVICE_DATA_DIR_NAME)
?.also { if (!it.isDirectory) throw IllegalStateException("devices is not a directory: $it") }
?: run {
appDataRoot.createDir(folderName = DEVICE_DATA_DIR_NAME)
.also { log(TAG, INFO) { "write(): Created devices dir $it" } }
appDataRoot.createDir(folderName = DEVICE_DATA_DIR_NAME).also {
log(TAG, INFO) { "write(): Created devices dir $it" }
}
}

val deviceIdRaw = data.deviceId.id
Expand Down Expand Up @@ -323,16 +331,18 @@ class GDriveAppDataConnector @AssistedInject constructor(

return try {
_state.updateBlocking { copy(activeActions = activeActions + 1) }
try {
withDrive {
driveLock.withLock {
block()
}

withDrive {
driveLock.withLock {
block()
}
} catch (e: Exception) {
log(TAG, ERROR) { "runDriveAction($tag) failed: ${e.asLog()}" }
throw e
}.also {
_state.updateBlocking { copy(lastError = null) }
}
} catch (e: Exception) {
log(TAG, ERROR) { "runDriveAction($tag) failed: ${e.asLog()}" }
_state.updateBlocking { copy(lastError = e) }
throw e
} finally {
_state.updateBlocking {
log(TAG, VERBOSE) { "runDriveAction($tag) finished" }
Expand Down
19 changes: 6 additions & 13 deletions syncs-kserver/src/main/java/eu/darken/octi/syncs/kserver/core/KServerConnector.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,6 @@ class KServerConnector @AssistedInject constructor(
}
} catch (e: Exception) {
log(TAG, ERROR) { "Failed to read: ${e.asLog()}" }
_state.updateBlocking { copy(lastError = e) }
}
}
}
Expand Down Expand Up @@ -258,11 +257,6 @@ class KServerConnector @AssistedInject constructor(
log(TAG, VERBOSE) { "writeServer(): Done" }
}

private fun getStorageStats(): SyncConnectorState.Quota {
log(TAG, VERBOSE) { "getStorageStats()" }
return SyncConnectorState.Quota()
}

private suspend fun <R> runServerAction(
tag: String,
block: suspend () -> R,
Expand All @@ -273,14 +267,13 @@ class KServerConnector @AssistedInject constructor(
return try {
_state.updateBlocking { copy(activeActions = activeActions + 1) }

try {
serverLock.withLock {
withContext(NonCancellable) { block() }
}
} catch (e: Exception) {
log(TAG, ERROR) { "runServerAction($tag) failed: ${e.asLog()}" }
throw e
serverLock.withLock {
withContext(NonCancellable) { block() }
}
} catch (e: Exception) {
log(TAG, ERROR) { "runServerAction($tag) failed: ${e.asLog()}" }
_state.updateBlocking { copy(lastError = e) }
throw e
} finally {
_state.updateBlocking {
log(TAG, VERBOSE) { "runServerAction($tag) finished" }
Expand Down
Loading