Merge pull request #2 from juyan/memory-store

Add InMemoryStore and simplify concurrency

.github/workflows/swift.yml

@@ -11,10 +11,11 @@ on:
 
 jobs:
   build:
-
     runs-on: macos-latest
-
     steps:
+    - uses: swift-actions/setup-swift@v1
+      with:
+        swift-version: '5.5'
     - uses: actions/checkout@v3
     - name: Build
       run: swift build -v

Sources/SwiftFileStore/FileObjectStore.swift

@@ -28,47 +28,41 @@ public final class FileObjectStore: ObjectStore {
   }
 
   public func read<T>(key: String, namespace: String, objectType _: T.Type) async throws -> T? where T: DataRepresentable {
-    let readTask = Task { () -> T? in
-      let fileURL = rootDir.appendingPathComponent(namespace).appendingPathComponent(key)
-      if FileManager.default.fileExists(atPath: fileURL.path) {
-        return try T.from(data: Data(contentsOf: fileURL))
-      } else {
-        return nil
-      }
+    let fileURL = rootDir.appendingPathComponent(namespace).appendingPathComponent(key)
+    if FileManager.default.fileExists(atPath: fileURL.path) {
+      return try T.from(data: Data(contentsOf: fileURL))
+    } else {
+      return nil
     }
-    return try await readTask.value
   }
 
   public func write<T>(key: String, namespace: String, object: T) async throws where T: DataRepresentable {
-    let writeTask = Task { () in
-      let dirURL = rootDir.appendingPathComponent(namespace)
-      let fileURL = dirURL.appendingPathComponent(key)
-      try FileManager.default.createDirIfNotExist(url: dirURL)
-      try object.serialize().write(to: fileURL)
+    let dirURL = rootDir.appendingPathComponent(namespace)
+    let fileURL = dirURL.appendingPathComponent(key)
+    try FileManager.default.createDirIfNotExist(url: dirURL)
+    try object.serialize().write(to: fileURL)
+    Task {
       await observerManager.publishValue(key: key, namespace: namespace, value: object)
     }
-    return try await writeTask.value
   }
 
   public func remove(key: String, namespace: String) async throws {
-    let removeTask = Task {
-      let dirURL = rootDir.appendingPathComponent(namespace)
-      let fileURL = dirURL.appendingPathComponent(key)
-      if FileManager.default.fileExists(atPath: fileURL.path) {
-        try FileManager.default.removeItem(at: fileURL)
-      }
+    let dirURL = rootDir.appendingPathComponent(namespace)
+    let fileURL = dirURL.appendingPathComponent(key)
+    if FileManager.default.fileExists(atPath: fileURL.path) {
+      try FileManager.default.removeItem(at: fileURL)
+    }
+    Task {
       await observerManager.publishRemoval(namespace: namespace, key: key)
     }
-    return try await removeTask.value
   }
 
   public func removeAll(namespace: String) async throws {
-    let removeAllTask = Task {
-      let dirURL = rootDir.appendingPathComponent(namespace)
-      try FileManager.default.removeItem(at: dirURL)
+    let dirURL = rootDir.appendingPathComponent(namespace)
+    try FileManager.default.removeItem(at: dirURL)
+    Task {
       await observerManager.publishRemoval(namespace: namespace)
     }
-    return try await removeAllTask.value
   }
 
   public func observe<T>(key: String, namespace: String, objectType: T.Type) async -> AsyncThrowingStream<T?, Error> where T: DataRepresentable {

Sources/SwiftFileStore/MemoryObjectStore.swift

@@ -0,0 +1,64 @@
+//
+//  File.swift
+//  
+//
+//  Created by Jun Yan on 11/21/23.
+//
+
+import Foundation
+
+
+/// A fallback `ObjectStore` in case file operation fails. Can also be used for unit tests.
+public actor MemoryObjectStore: ObjectStore {
+
+    private var objects: [String: [String: Data]] = [:]
+    private let observerManager = ObserverManager()
+
+
+    public func read<T>(key: String, namespace: String, objectType: T.Type) async throws -> T? where T : DataRepresentable {
+        return objects[namespace]?[key].flatMap { try? T.from(data: $0) }
+    }
+
+    public func write<T>(key: String, namespace: String, object: T) async throws where T : DataRepresentable {
+        let data = try object.serialize()
+        objects[namespace, default: [:]][key] = data
+        await observerManager.publishValue(key: key, namespace: namespace, value: object)
+    }
+
+    public func remove(key: String, namespace: String) async throws {
+        objects[namespace, default: [:]][key] = nil
+        await observerManager.publishRemoval(namespace: namespace, key: key)
+    }
+
+    public func removeAll(namespace: String) async throws {
+        objects[namespace] = nil
+        await observerManager.publishRemoval(namespace: namespace)
+    }
+
+    public func observe<T>(key: String, namespace: String, objectType: T.Type) async -> AsyncThrowingStream<T?, Error> where T : DataRepresentable {
+        let observer = await observerManager.getObserver(key: key, namespace: namespace)
+        do {
+          let existingValue = try await read(key: key, namespace: namespace, objectType: objectType)
+          return AsyncThrowingStream { continuation in
+            continuation.yield(existingValue)
+            let callbackID = UUID().uuidString
+            observer.registerCallback(id: callbackID) { data in
+              if let d = data, let typed = d as? T {
+                continuation.yield(typed)
+              } else if data == nil {
+                continuation.yield(nil)
+              } else {
+                continuation.finish(throwing: "invalid data type")
+              }
+            }
+            continuation.onTermination = { @Sendable _ in
+              observer.callbacks.removeValue(forKey: callbackID)
+            }
+          }
+        } catch {
+          return AsyncThrowingStream { continuation in
+            continuation.finish(throwing: error)
+          }
+        }
+    }
+}

Sources/SwiftFileStore/ObserverManager.swift

@@ -36,32 +36,26 @@ actor ObserverManager {
   }
 
   func publishValue<T>(key: String, namespace: String, value: T) where T: DataRepresentable {
-    Task.detached(priority: .background) {
-      if let observer = await self.observers[namespace]?[key] {
-        observer.callbacks.values.forEach { callback in
-          callback(value)
-        }
+    if let observer = self.observers[namespace]?[key] {
+      observer.callbacks.values.forEach { callback in
+        callback(value)
       }
     }
   }
 
   func publishRemoval(namespace: String, key: String) {
-    Task.detached(priority: .background) {
-      if let observer = await self.observers[namespace]?[key] {
-        observer.callbacks.values.forEach { callback in
-          callback(nil)
-        }
+    if let observer = self.observers[namespace]?[key] {
+      observer.callbacks.values.forEach { callback in
+        callback(nil)
       }
     }
   }
 
   func publishRemoval(namespace: String) {
-    Task.detached(priority: .background) {
-      if let namespaceObservers = await self.observers[namespace]?.values {
-        namespaceObservers.forEach { observer in
-          observer.callbacks.values.forEach { callback in
-            callback(nil)
-          }
+    if let namespaceObservers = self.observers[namespace]?.values {
+      namespaceObservers.forEach { observer in
+        observer.callbacks.values.forEach { callback in
+          callback(nil)
         }
       }
     }

Tests/SwiftFileStoreTests/FileObjectStoreTests.swift

@@ -0,0 +1,72 @@
+import Combine
+import XCTest
+@testable import SwiftFileStore
+
+final class FileObjectStoreTests: XCTestCase {
+
+  var store: FileObjectStore!
+
+  override func setUp() {
+    super.setUp()
+    store = try! FileObjectStore.create()
+  }
+
+  override func tearDown() {
+    super.tearDown()
+    try! FileManager.default.removeItem(at: store.rootDir)
+  }
+
+  func test_readWrite() async throws {
+    let object = TestObject(value: 2)
+    try await store.write(key: "test", namespace: "test", object: object)
+    let readResult = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    XCTAssertEqual(readResult, object)
+  }
+
+  func test_deletetNamespace() async throws {
+    let object = TestObject(value: 1)
+    let object2 = TestObject(value: 2)
+    try await store.write(key: "test", namespace: "test", object: object)
+    try await store.write(key: "test2", namespace: "test", object: object2)
+    try await store.removeAll(namespace: "test")
+
+    let readResult = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    let readResult2 = try await store.read(key: "test2", namespace: "test", objectType: TestObject.self)
+    XCTAssertNil(readResult)
+    XCTAssertNil(readResult2)
+  }
+
+  func test_deleteObject() async throws {
+    let object = TestObject(value: 1)
+    try await store.write(key: "test", namespace: "test", object: object)
+    let readResult = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    XCTAssertNotNil(readResult)
+    try await store.remove(key: "test", namespace: "test")
+    let readResult2 = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    XCTAssertNil(readResult2)
+  }
+
+  func test_observer() async throws {
+    let object = TestObject(value: 1)
+    let object2 = TestObject(value: 2)
+    let expectation = XCTestExpectation(description: "stream subscription")
+    let expectation2 = XCTestExpectation(description: "stream breaks")
+    Task {
+      var values: [TestObject?] = []
+      let stream = await store.observe(key: "test", namespace: "test", objectType: TestObject.self)
+      expectation.fulfill()
+      for try await value in stream {
+        values.append(value)
+        if values.count == 3 {
+          break
+        }
+      }
+      XCTAssertEqual(values, [nil, object, object2])
+      expectation2.fulfill()
+    }
+    await fulfillment(of: [expectation], timeout: 1)
+    try await store.write(key: "test", namespace: "test", object: object)
+    try await store.write(key: "test", namespace: "test", object: object2)
+    await fulfillment(of: [expectation2], timeout: 1)
+  }
+}

Tests/SwiftFileStoreTests/MemoryObjectStoreTests.swift

@@ -0,0 +1,62 @@
+import Combine
+import XCTest
+@testable import SwiftFileStore
+
+final class MemoryObjectStoreTests: XCTestCase {
+
+  let store = MemoryObjectStore()
+
+  func test_readWrite() async throws {
+    let object = TestObject(value: 2)
+    try await store.write(key: "test", namespace: "test", object: object)
+    let readResult = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    XCTAssertEqual(readResult, object)
+  }
+
+  func test_deletetNamespace() async throws {
+    let object = TestObject(value: 1)
+    let object2 = TestObject(value: 2)
+    try await store.write(key: "test", namespace: "test", object: object)
+    try await store.write(key: "test2", namespace: "test", object: object2)
+    try await store.removeAll(namespace: "test")
+
+    let readResult = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    let readResult2 = try await store.read(key: "test2", namespace: "test", objectType: TestObject.self)
+    XCTAssertNil(readResult)
+    XCTAssertNil(readResult2)
+  }
+
+  func test_deleteObject() async throws {
+    let object = TestObject(value: 1)
+    try await store.write(key: "test", namespace: "test", object: object)
+    let readResult = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    XCTAssertNotNil(readResult)
+    try await store.remove(key: "test", namespace: "test")
+    let readResult2 = try await store.read(key: "test", namespace: "test", objectType: TestObject.self)
+    XCTAssertNil(readResult2)
+  }
+
+  func test_observer() async throws {
+    let object = TestObject(value: 1)
+    let object2 = TestObject(value: 2)
+    let expectation = XCTestExpectation(description: "stream subscription")
+    let expectation2 = XCTestExpectation(description: "stream breaks")
+    Task {
+      var values: [TestObject?] = []
+      let stream = await store.observe(key: "test", namespace: "test", objectType: TestObject.self)
+      expectation.fulfill()
+      for try await value in stream {
+        values.append(value)
+        if values.count == 3 {
+          break
+        }
+      }
+      XCTAssertEqual(values, [nil, object, object2])
+      expectation2.fulfill()
+    }
+    await fulfillment(of: [expectation], timeout: 1)
+    try await store.write(key: "test", namespace: "test", object: object)
+    try await store.write(key: "test", namespace: "test", object: object2)
+    await fulfillment(of: [expectation2], timeout: 1)
+  }
+}

Tests/SwiftFileStoreTests/TestObject.swift

@@ -0,0 +1,13 @@
+//
+//  File.swift
+//  
+//
+//  Created by Jun Yan on 11/21/23.
+//
+
+import Foundation
+@testable import SwiftFileStore
+
+struct TestObject: Codable, JSONDataRepresentable, Equatable {
+    let value: Int
+}