Revert "[trello.com/c/CdfJ7GTd] ChatProvider crash fix"

This reverts commit 9662f93.

Adamant/Modules/Chat/ViewModel/ChatViewModel.swift

@@ -6,7 +6,7 @@
 //  Copyright © 2022 Adamant. All rights reserved.
 //
 
-import Combine
+@preconcurrency import Combine
 import CoreData
 import MarkdownKit
 import UIKit
@@ -1122,8 +1122,8 @@ private extension ChatViewModel {
 
         Task {
             await chatsProvider.stateObserver
-                .makeSequence()
-                .sink { @MainActor [weak self] state in
+                .receive(on: DispatchQueue.main)
+                .sink { [weak self] state in
                     self?.isHeaderLoading = state == .updating ? true : false
                 }
                 .store(in: &subscriptions)
@@ -1198,7 +1198,7 @@ private extension ChatViewModel {
         updateMessages(
             resetLoadingProperty: performFetch,
             completion: isNewReaction
-                ? { @MainActor [commitVibro] in commitVibro.send() }
+                ? { @Sendable [commitVibro] in commitVibro.send() }
                 : {}
         )
     }
@@ -1468,11 +1468,22 @@ private extension ChatViewModel {
     }
 
     func waitForChatLoading(with address: String) async {
-        _ = await chatsProvider.chatLoadingStatus.makeSequence()
-            .filter { $0.contains { $0.key == address && $0.value == .loaded } }
-            .first
-
-        tempCancellables.removeAll()
+        await withUnsafeContinuation { continuation in
+            Task {
+                let publisher = await chatsProvider.chatLoadingStatusPublisher
+                publisher
+                    .filter { dict in
+                        dict.contains {
+                            $0.key == address && $0.value == .loaded
+                        }
+                    }
+                    .receive(on: DispatchQueue.main)
+                    .sink { [weak self] _ in
+                        self?.tempCancellables.removeAll()
+                        continuation.resume()
+                    }.store(in: &tempCancellables)
+            }
+        }
     }
 
     // TODO: Post process

Adamant/ServiceProtocols/DataProviders/ChatsProvider.swift

@@ -185,7 +185,7 @@ protocol ChatsProvider: DataProvider, Actor {
     var roomsMaxCount: Int? { get }
     var roomsLoadedCount: Int? { get }
 
-    var chatLoadingStatus: AnyAsyncStreamable<[String: ChatRoomLoadingStatus]> {
+    var chatLoadingStatusPublisher: AnyObservable<[String: ChatRoomLoadingStatus]> {
         get
     }
 

Adamant/ServiceProtocols/DataProviders/DataProvider.swift

@@ -18,7 +18,7 @@ enum State {
 
 protocol DataProvider: AnyObject, Actor {
     var state: State { get }
-    var stateObserver: AnyAsyncStreamable<State> { get }
+    var stateObserver: AnyObservable<State> { get }
     var isInitiallySynced: Bool { get }
 
     func reload() async

Adamant/Services/DataProviders/AdamantChatsProvider.swift

@@ -28,8 +28,8 @@ actor AdamantChatsProvider: ChatsProvider {
     let stack: CoreDataStack
 
     // MARK: Properties
-    private let stateNotifier = SendablePublisher(ObservableValue(State.empty))
-    var stateObserver: AnyAsyncStreamable<State> { stateNotifier.eraseToAnyAsyncStreamable() }
+    @ObservableValue private var stateNotifier: State = .empty
+    var stateObserver: AnyObservable<State> { $stateNotifier.eraseToAnyPublisher() }
 
     private(set) var state: State = .empty
     private(set) var receivedLastHeight: Int64?
@@ -43,12 +43,9 @@ actor AdamantChatsProvider: ChatsProvider {
     private(set) var blockList: [String] = []
     private(set) var removedMessages: [String] = []
 
-    private let chatLoadingStatusDictionary = SendablePublisher(ObservableValue(
-        [String: ChatRoomLoadingStatus]()
-    ))
-
-    var chatLoadingStatus: AnyAsyncStreamable<[String: ChatRoomLoadingStatus]> {
-        chatLoadingStatusDictionary.eraseToAnyAsyncStreamable()
+    @ObservableValue private var chatLoadingStatusDictionary: [String: ChatRoomLoadingStatus] = [:]
+    var chatLoadingStatusPublisher: AnyObservable<[String: ChatRoomLoadingStatus]> {
+        $chatLoadingStatusDictionary.eraseToAnyPublisher()
     }
 
     var chatMaxMessages: [String : Int] = [:]
@@ -246,10 +243,7 @@ actor AdamantChatsProvider: ChatsProvider {
                 ]
             )
 
-            // TODO: Remove Task.sync
-            Task.sync { [weak self] in
-                await self?.stateNotifier.isolated { $0.publisher.value = state }
-            }
+            stateNotifier = state
             return
         }
 
@@ -264,10 +258,7 @@ actor AdamantChatsProvider: ChatsProvider {
             ]
         )
 
-        // TODO: Remove Task.sync
-        Task.sync { [weak self] in
-            await self?.stateNotifier.isolated { $0.publisher.value = state }
-        }
+        stateNotifier = state
     }
 
     private func setupSecuredStore() {
@@ -306,13 +297,10 @@ extension AdamantChatsProvider {
         readedLastHeight = nil
         roomsMaxCount = nil
         roomsLoadedCount = nil
+        chatLoadingStatusDictionary.removeAll()
         chatMaxMessages.removeAll()
         chatLoadedMessages.removeAll()
 
-        Task.sync { [self] in
-            await chatLoadingStatusDictionary.isolated { $0.publisher.value.removeAll() }
-        }
-
         // Drop store
         securedStore.remove(StoreKey.chatProvider.address)
         securedStore.remove(StoreKey.chatProvider.receivedLastHeight)
@@ -522,7 +510,7 @@ extension AdamantChatsProvider {
                 limit: limit
             ).get()
             return chatrooms
-        } catch let error {
+        } catch let error as ApiServiceError {
             guard case .networkError = error else {
                 return nil
             }
@@ -668,27 +656,19 @@ extension AdamantChatsProvider {
     }
 
     func isChatLoading(with addressRecipient: String) -> Bool {
-        Task.sync { [self] in
-            await chatLoadingStatusDictionary.isolated { $0.publisher.value[addressRecipient] == .loading }
-        }
+        chatLoadingStatusDictionary[addressRecipient] == .loading
     }
 
     func isChatLoaded(with addressRecipient: String) -> Bool {
-        Task.sync { [self] in
-            await chatLoadingStatusDictionary.isolated { $0.publisher.value[addressRecipient] == .loaded }
-        }
+        chatLoadingStatusDictionary[addressRecipient] == .loaded
     }
 
     func getChatStatus(for recipient: String) -> ChatRoomLoadingStatus {
-        Task.sync { [self] in
-            await chatLoadingStatusDictionary.isolated { $0.publisher.value[recipient] ?? .none }
-        }
+        chatLoadingStatusDictionary[recipient] ?? .none
     }
 
     func setChatStatus(for recipient: String, status: ChatRoomLoadingStatus) {
-        Task.sync { [self] in
-            await chatLoadingStatusDictionary.isolated { $0.publisher.value[recipient] = status }
-        }
+        chatLoadingStatusDictionary[recipient] = status
     }
 }
 
@@ -839,7 +819,7 @@ extension AdamantChatsProvider {
             )
         }
 
-        _ = try await sendMessageToServer(
+        let transaction = try await sendMessageToServer(
             senderId: loggedAccount.address,
             recipientId: recipientId,
             transaction: transactionLocaly,
@@ -1332,7 +1312,7 @@ extension AdamantChatsProvider {
 
             return transaction
         } catch {
-            guard case let(apiError) = (error as ApiServiceError),
+            guard case let(apiError) = (error as? ApiServiceError),
                   case let(.serverError(text)) = apiError,
                   text.contains("Transaction is already confirmed")
                     || text.contains("Transaction is already processed")

Adamant/Services/DataProviders/AdamantTransfersProvider.swift

@@ -24,18 +24,9 @@ actor AdamantTransfersProvider: TransfersProvider {
     let securedStore: SecuredStore
     private let transactionService: ChatTransactionService
     weak var chatsProvider: ChatsProvider?
-    private let statePublisher = SendablePublisher(ObservableValue(State.empty))
-
-    var state: State {
-        Task.sync { [self] in
-            await statePublisher.isolated { $0.publisher.value }
-        }
-    }
-
-    var stateObserver: AnyAsyncStreamable<State> {
-        statePublisher.eraseToAnyAsyncStreamable()
-    }
 
+    @ObservableValue private(set) var state: State = .empty
+    var stateObserver: AnyObservable<State> { $state.eraseToAnyPublisher() }
     private(set) var isInitiallySynced: Bool = false
     private(set) var receivedLastHeight: Int64?
     private(set) var readedLastHeight: Int64?
@@ -51,9 +42,7 @@ actor AdamantTransfersProvider: TransfersProvider {
 
     /// Free stateSemaphore before calling this method, or you will deadlock.
     private func setState(_ state: State, previous prevState: State, notify: Bool = false) {
-        Task.sync { [weak self] in
-            await self?.statePublisher.isolated { $0.publisher.value = state }
-        }
+        self.state = state
 
         if notify {
             switch prevState {
@@ -199,9 +188,7 @@ extension AdamantTransfersProvider {
         }
 
         let prevState = state
-        Task.sync { [weak self] in
-            await self?.statePublisher.isolated { $0.publisher.value = .updating }
-        }
+        state = .updating
 
         guard let address = accountService.account?.address else {
             self.setState(.failedToUpdate(TransfersProviderError.notLogged), previous: prevState)

CommonKit/Sources/CommonKit/Helpers/Observation/ObservableValue.swift

@@ -10,15 +10,13 @@ import Combine
 
 /// `Published` changes its `wrappedValue` after calling `sink` or `assign`.
 /// But `ObservableValue` does it before.
-@propertyWrapper public final class ObservableValue<Output> {
+@propertyWrapper public final class ObservableValue<Output>: Publisher {
+    public typealias Output = Output
+    public typealias Failure = Never
+
     private let subject: CurrentValueSubject<Output, Failure>
 
     public var wrappedValue: Output {
-        get { value }
-        set { value = newValue }
-    }
-
-    public var value: Output {
         get { subject.value }
         set { subject.value = newValue }
     }
@@ -27,35 +25,15 @@ import Combine
         subject
     }
 
-    public init(_ value: Output) {
-        subject = .init(value)
-    }
-
-    public convenience init(wrappedValue: Output) {
-        self.init(wrappedValue)
-    }
-}
-
-extension ObservableValue: Subject {
-    public typealias Failure = Never
-
-    public func send(_ value: Output) {
-        subject.send(value)
-    }
-
-    public func send(completion: Subscribers.Completion<Never>) {
-        subject.send(completion: completion)
-    }
-
-    public func send(subscription: any Subscription) {
-        subject.send(subscription: subscription)
-    }
-
     public func receive<S>(
         subscriber: S
     ) where S: Subscriber, Never == S.Failure, Output == S.Input {
         subject.receive(subscriber: subscriber)
     }
+
+    public init(wrappedValue: Output) {
+        subject = .init(wrappedValue)
+    }
 }
 
 public extension Publisher where Failure == Never {

CommonKit/Sources/CommonKit/Helpers/Task+Extension.swift

@@ -30,18 +30,18 @@ public extension Task where Success == Never, Failure == Never {
 
     /// Avoid using it. It lowers performance due to changing threads.
     @discardableResult
-    static func sync<T: Sendable>(_ action: @Sendable @escaping () async throws -> T) rethrows -> T {
-        try _sync(action)
+    static func sync<T: Sendable>(_ action: @Sendable @escaping () async -> T) -> T {
+        _sync(action)
     }
 }
 
 @discardableResult
-private func _sync<T: Sendable>(_ action: @Sendable @escaping () async throws -> T) rethrows -> T {
+private func _sync<T: Sendable>(_ action: @Sendable @escaping () async -> T) -> T {
     var result: T?
     let semaphore = DispatchSemaphore(value: .zero)
 
     Task {
-        result = try await action()
+        result = await action()
         semaphore.signal()
     }
 

CommonKit/Sources/CommonKit/Models/ChatRoomLoadingStatus.swift

@@ -7,7 +7,7 @@
 
 import Foundation
 
-public enum ChatRoomLoadingStatus: Sendable {
+public enum ChatRoomLoadingStatus {
     case loaded
     case loading
     case none