[Android] Jetpack DataStoreのRead/Writeについて読んだので覚え書き

# はじめに
[Jetpack DataStore](https://developer.android.com/topic/libraries/architecture/datastore)がどうやってデータの管理をしているのかを知っておこうと思って、ちょっと[ソース](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/)を読んでみたのでその覚え書きです。
**自身のメモ的な要素が多分に含まれております。予めご了承いただければ幸いです。**

# ソースについて
今回読んだソースコードは2020/10/30時点のものとなります。
`DataStore`は現時点ではAlphaなのですぐに実装が様変わりすることも十分あるかと思います。

また、`DataStore`には`Preferences DataStore`と`Proto DataStore`がありますが、どちらもCoreの部分は同じコードで動いていて、データの扱い方がちょっと違う(`Preferences DataStore`では内部で汎用的なProtoファイルを定義して処理をしている)だけっぽかったので、今回は共通部分に関してのものになります。
**なのでPreferencesの方はあまり考慮せずに書いてます**

## Utils
今回特に詳細述べませんが出現するクラスは以下となります。

- DataAndHash: データ自体と、そのデータのHash値をセットにした`data class`意図しない改ざんなどが起こっていないことを判断するためのクラス
- UncloseableOutputStream: 書き込み時に、DataStore利用者がstreamをCloseできないようにするためのWrapperクラス
- ConflatedBroadcastChannel: 読み込まれたり更新されたデータが流れてくるChannel

# Read/Write実行の起点
そもそものDataStoreの初期化は[コレ](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore/src/main/java/androidx/datastore/DataStoreFactory.kt)

```kotlin:datastore/datastore/src/main/java/androidx/datastore/DataStoreFactory.kt
public fun <T> Context.createDataStore(...): DataStore<T> = DataStoreFactory.create(...)
```
そして`DataStoreFactory.create(...)`の定義は[ココ](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/DataStoreFactory.kt)

```kotlin:datastore/datastore-core/src/main/java/androidx/datastore/core/DataStoreFactory.kt
public fun <T> create(...): DataStore<T> = SingleProcessDataStore(...)
```

この`SingleProcessDataStore`が今回の目的物です。
すべての変更(read/write)はクラス内で生成する[`actor: SendChannel<Message<T>>`](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=142)内で処理されます。

```kotlin
  private val actor: SendChannel<Message<T>> = scope.actor(
    capacity = UNLIMITED
  ) {
    try {
      messageConsumer@ for (msg in channel) {
      ...

      try {
        readAndInitOnce(msg.dataChannel) // Read実行
      } catch (ex: Throwable) {
        resetDataChannel(ex)
        continue@messageConsumer
      }

      // We have successfully read data and sent it to downstreamChannel.

      if (msg is Message.Update) {
        msg.ack.completeWith(
          runCatching {
            transformAndWrite(msg.transform, downstreamChannel()) // Write実行
          }
        )
      }
    ...
  }
```

# Read
開始位置の定義は[ココ](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=183)。

```kotlin
  private suspend fun readAndInitOnce(dataChannel: ConflatedBroadcastChannel<DataAndHash<T>>) {
    if (dataChannel.valueOrNull != null) {
      // If we already have cached data, we don't try to read it again.
      return
    }

    val updateLock = Mutex()
    var initData = readDataOrHandleCorruption()

    ...

    dataChannel.offer(DataAndHash(initData, initData.hashCode()))
  }
```
読み込み実行は１度だけ行われ、すでに読み込みが行われている場合は関数最初のベースケースでEarly Returnして処理をスキップします。
実際に読み込むのは[readDataOrHandleCorruption()](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=190)関数内の[readData()](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=227)呼び出しです。

```kotlin
  private suspend fun readDataOrHandleCorruption(): T {
    try {
      return readData()
    } catch (ex: CorruptionException) {

      val newData: T = corruptionHandler.handleCorruption(ex)

      try {
        writeData(newData)
      } catch (writeEx: IOException) {
        // If we fail to write the handled data, add the new exception as a suppressed
        // exception.
        ex.addSuppressed(writeEx)
        throw ex
      }

      // If we reach this point, we've successfully replaced the data on disk with newData.
      return newData
    }
  }
```
データの読み込みに失敗した場合、Catchブロックが実行されることになり、defaultでは`corruptionHandler`はExceptionをそのまま返しますが、`SingleProcessDataStore`クラスのコンストラクタで、カスタムハンドラを渡すこともできます。
その場合はハンドラで取得できるデータを利用してデータ一旦の書き込みを行います。

```kotlin
  private suspend fun readData(): T {
    try {
      FileInputStream(file).use { stream ->
        return serializer.readFrom(stream)
      }
    } catch (ex: FileNotFoundException) {
      if (file.exists()) {
        throw ex
      }
      return serializer.defaultValue
    }
  }
```
ここでわかる通り、データファイルはserializerでシリアライズしたものをファイルに読み込み/書き出していて、後ろにRoomやSharedPreference等があるとかそういうものではないようですね。
またファイルがないとき、基本的には`serializer.defaultValue`が返却されますが、異常ケースの場合はExceptionが`readAndInitOnce`の呼び出し元まで飛び、処理自体が無効化されます。

# Write
書き込みに関しても、まずは読み込みの処理が走ります。
万が一読み込み前に書き込みが走ってもいいように配慮しているんしょうか。
開始位置の定義は[ココ](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=259)

```kotlin
  private suspend fun transformAndWrite(
    transform: suspend (t: T) -> T,
    /**
     * This is the channel that contains the data that will be used for the transformation.
     * It *must* already have a value -- otherwise this will throw IllegalStateException.
     * Once the transformation is completed and data is durably persisted to disk, and the new
     * value will be offered to this channel.
     */
    updateDataChannel: ConflatedBroadcastChannel<DataAndHash<T>>
  ): T {
    val curDataAndHash = updateDataChannel.value
    curDataAndHash.checkHashCode()
    val curData = curDataAndHash.value
    val newData = transform(curData)

    // Check that curData has not changed...
    curDataAndHash.checkHashCode()

    return if (curData == newData) {
      curData
    } else {
      writeData(newData)
      updateDataChannel.offer(DataAndHash(newData, newData.hashCode()))
      newData
    }
  }
```
まずはデータのtransform。
現在のDataのHashをチェックして、保持しているデータ自体が変更されていないことを確認します(データはImmutableであることが要求されている)
その後transform関数でデータを変換し、再度現在のデータのHashを確認します。これはtransform関数が副作用を持っていないことを保証するためのようです。
変換前と後でデータが変わっていなかったら何もせず、更新後のデータ書き込みは[writeData(...)](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=290)関数で行います。

```kotlin
  internal fun writeData(newData: T) {
    file.createParentDirectories()

    val scratchFile = File(file.absolutePath + SCRATCH_SUFFIX)
    try {
      FileOutputStream(scratchFile).use { stream ->
        serializer.writeTo(newData, UncloseableOutputStream(stream))
        stream.fd.sync()
        // TODO(b/151635324): fsync the directory, otherwise a badly timed crash could
        //  result in reverting to a previous state.
      }

      if (!scratchFile.renameTo(file)) {
        throw IOException("$scratchFile could not be renamed to $file")
      }
    } catch (ex: IOException) {
      if (scratchFile.exists()) {
        scratchFile.delete()
      }
      throw ex
    }
  }
```
書き込みは直接データファイルに行うのではなく、`SCRATCH_SUFFIX`(".tmp")に対して実行し、書き込みが完了してからrenameすることで行っているようです。
Linuxファイルシステムのやってるようなこと(データ破損対策)を表現してるんですかね。

# Read/Write実行の呼び出し
## Read
`DataStore`のデータはFlow型として扱うことができ、

```kotlin
  override val data: Flow<T> = flow {
    val curChannel = downstreamChannel()
    actor.offer(Message.Read(curChannel)) // Read実行呼び出し
    emitAll(curChannel.asFlow().map { it.value })
  }
```
と定義されています。
このFlowを作るときにactorにイベントを渡すことでReadの実行が呼び出されています。
そして、curChannel(ConflatedBroadcastChannel)から流れてきたデータを渡しています。
なので、データを取得したり変更したときにはこのChannelに対してOfferして検知できるようにしています。

## Write
データ更新は[以下の関数](https://cs.android.com/androidx/platform/frameworks/support/+/androidx-master-dev:datastore/datastore-core/src/main/java/androidx/datastore/core/SingleProcessDataStore.kt;l=82)から行われます。

```kotlin
  override suspend fun updateData(transform: suspend (t: T) -> T): T {
    val ack = CompletableDeferred<T>()
    val dataChannel = downstreamChannel()
    val updateMsg = Message.Update<T>(transform, ack, dataChannel)

    actor.send(updateMsg)

    // If no read has succeeded yet, we need to wait on the result of the next read so we can
    // bubble exceptions up to the caller. Read exceptions are not bubbled up through ack.
    if (dataChannel.valueOrNull == null) {
      dataChannel.asFlow().first()
    }

    // Wait with same scope as the actor, so we're not waiting on a cancelled actor.
    return withContext(scope.coroutineContext) { ack.await() }
  }
```
書き込みに関しても、読み込みとを同じくactorにイベントを渡して、`ack`がcompleteされるまで待っています。

# おわりに
ざっくりとではありますが、Read/Writeに関して書きました。
`scope.actor(...)`はobsoleteであったりしていますが、この辺もVersionが進めば変わってきそうですね。

また読んでいて思った点としては、
例えば`Activity A`でも`Activity B`でも同じDataを見ていて、　`A`->`B`->`A`(page back)と遷移したとき、`B`での変更は`A`では検知できず、データがズレてしまうのでは？
そもそも使い方として間違った使い方なのかもしれませんが、やむにやまれずそうせざるを得なくなったときどうするべきなんでしょ。