|
|
|
@ -17,24 +17,24 @@ public protocol ActivitiesServiceType: AnyObject { |
|
|
|
|
var activitiesPublisher: AnyPublisher<[ActivityCollection.MappedToDateActivityOrTransaction], Never> { get } |
|
|
|
|
var didUpdateActivityPublisher: AnyPublisher<Activity, Never> { get } |
|
|
|
|
|
|
|
|
|
func start() |
|
|
|
|
func stop() |
|
|
|
|
func reinject(activity: Activity) |
|
|
|
|
func copy(activitiesFilterStrategy: ActivitiesFilterStrategy, transactionsFilterStrategy: TransactionsFilterStrategy) -> ActivitiesServiceType |
|
|
|
|
func start() async |
|
|
|
|
func stop() async |
|
|
|
|
func reinject(activity: Activity) async |
|
|
|
|
func copy(activitiesFilterStrategy: ActivitiesFilterStrategy, transactionsFilterStrategy: TransactionsFilterStrategy) async -> ActivitiesServiceType |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
typealias ContractsAndCards = [(contract: AlphaWallet.Address, server: RPCServer, card: TokenScriptCard, interpolatedFilter: String)] |
|
|
|
|
typealias ActivityTokenObjectTokenHolder = (activity: Activity, tokenObject: Token, tokenHolder: TokenHolder) |
|
|
|
|
typealias TokenObjectsAndXMLHandlers = [(contract: AlphaWallet.Address, server: RPCServer, xmlHandler: XMLHandler)] |
|
|
|
|
|
|
|
|
|
public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
public actor ActivitiesService: ActivitiesServiceType { |
|
|
|
|
let sessionsProvider: SessionsProvider |
|
|
|
|
private let tokensService: TokensService |
|
|
|
|
private let eventsActivityDataStore: EventsActivityDataStoreProtocol |
|
|
|
|
//Dictionary for lookup. Using `.firstIndex` too many times is too slow (60s for 10k events) |
|
|
|
|
private var activitiesIndexLookup: AtomicDictionary<Int, (index: Int, activity: Activity)> = .init() |
|
|
|
|
private var cancellableSet: AtomicDictionary<AddressAndRPCServer, AnyCancellable> = .init() |
|
|
|
|
private var activities: AtomicArray<Activity> = .init() |
|
|
|
|
private var activitiesIndexLookup: [Int: (index: Int, activity: Activity)] = [:] |
|
|
|
|
private var cancellableSet: [AddressAndRPCServer: AnyCancellable] = [:] |
|
|
|
|
private var activities: [Activity] = [] |
|
|
|
|
private let didUpdateActivitySubject: PassthroughSubject<Activity, Never> = .init() |
|
|
|
|
private let activitiesSubject: CurrentValueSubject<[ActivityCollection.MappedToDateActivityOrTransaction], Never> = .init([]) |
|
|
|
|
private var cancellable = Set<AnyCancellable>() |
|
|
|
@ -47,11 +47,11 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
private let transactionDataStore: TransactionDataStore |
|
|
|
|
private let transactionsFilterStrategy: TransactionsFilterStrategy |
|
|
|
|
|
|
|
|
|
public var activitiesPublisher: AnyPublisher<[ActivityCollection.MappedToDateActivityOrTransaction], Never> { |
|
|
|
|
public nonisolated var activitiesPublisher: AnyPublisher<[ActivityCollection.MappedToDateActivityOrTransaction], Never> { |
|
|
|
|
activitiesSubject.eraseToAnyPublisher() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public var didUpdateActivityPublisher: AnyPublisher<Activity, Never> { |
|
|
|
|
public nonisolated var didUpdateActivityPublisher: AnyPublisher<Activity, Never> { |
|
|
|
|
didUpdateActivitySubject.eraseToAnyPublisher() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -79,7 +79,7 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
eventsActivityDataStore: eventsActivityDataStore) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public func start() { |
|
|
|
|
public func start() async { |
|
|
|
|
let transactionsChangeset = sessionsProvider.sessions |
|
|
|
|
.receive(on: DispatchQueue.main) |
|
|
|
|
.handleEvents(receiveOutput: { [activitiesSubject] _ in activitiesSubject.send([]) }) |
|
|
|
@ -88,39 +88,36 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
transactionDataStore.transactionsChangeset(filter: transactionsFilterStrategy, servers: $0) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
let activities = activitiesGenerator.generateActivities() |
|
|
|
|
let activities = await activitiesGenerator.generateActivities() |
|
|
|
|
|
|
|
|
|
Publishers.CombineLatest(transactionsChangeset, activities) |
|
|
|
|
.sink { [weak self] data in self?.createActivities(activitiesAndTokens: data.1) } |
|
|
|
|
.store(in: &cancellable) |
|
|
|
|
.sink { data in |
|
|
|
|
Task { [weak self] in |
|
|
|
|
await self?.createActivities(activitiesAndTokens: data.1) |
|
|
|
|
} |
|
|
|
|
}.store(in: &cancellable) |
|
|
|
|
|
|
|
|
|
didUpdateActivitySubject |
|
|
|
|
.debounce(for: .seconds(5), scheduler: RunLoop.main) |
|
|
|
|
.receive(on: DispatchQueue.global()) |
|
|
|
|
.sink { [weak self] _ in |
|
|
|
|
self?.combineActivitiesWithTransactions() |
|
|
|
|
.sink { _ in |
|
|
|
|
Task { [weak self] in |
|
|
|
|
await self?.combineActivitiesWithTransactions() |
|
|
|
|
} |
|
|
|
|
}.store(in: &cancellable) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public func stop() { |
|
|
|
|
public func stop() async { |
|
|
|
|
cancellable.cancellAll() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public func copy(activitiesFilterStrategy: ActivitiesFilterStrategy, |
|
|
|
|
transactionsFilterStrategy: TransactionsFilterStrategy) -> ActivitiesServiceType { |
|
|
|
|
|
|
|
|
|
return ActivitiesService( |
|
|
|
|
sessionsProvider: sessionsProvider, |
|
|
|
|
eventsActivityDataStore: eventsActivityDataStore, |
|
|
|
|
transactionDataStore: transactionDataStore, |
|
|
|
|
activitiesFilterStrategy: activitiesFilterStrategy, |
|
|
|
|
transactionsFilterStrategy: transactionsFilterStrategy, |
|
|
|
|
tokensService: tokensService) |
|
|
|
|
public func copy(activitiesFilterStrategy: ActivitiesFilterStrategy, transactionsFilterStrategy: TransactionsFilterStrategy) async -> ActivitiesServiceType { |
|
|
|
|
return ActivitiesService(sessionsProvider: sessionsProvider, eventsActivityDataStore: eventsActivityDataStore, transactionDataStore: transactionDataStore, activitiesFilterStrategy: activitiesFilterStrategy, transactionsFilterStrategy: transactionsFilterStrategy, tokensService: tokensService) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private func createActivities(activitiesAndTokens: [ActivityTokenObjectTokenHolder]) { |
|
|
|
|
activities.set(array: activitiesAndTokens.compactMap { $0.activity }.sorted { $0.blockNumber > $1.blockNumber }) |
|
|
|
|
updateActivitiesIndexLookup(with: activities.all) |
|
|
|
|
activities = activitiesAndTokens.compactMap { $0.activity }.sorted { $0.blockNumber > $1.blockNumber } |
|
|
|
|
updateActivitiesIndexLookup(with: activities) |
|
|
|
|
|
|
|
|
|
combineActivitiesWithTransactions() |
|
|
|
|
|
|
|
|
@ -129,16 +126,15 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public func reinject(activity: Activity) { |
|
|
|
|
guard let tokenHolders = activitiesGenerator.tokensAndTokenHolders[activity.token.addressAndRPCServer] else { return } |
|
|
|
|
|
|
|
|
|
public func reinject(activity: Activity) async { |
|
|
|
|
guard let tokenHolders = await activitiesGenerator.tokensAndTokenHolders[activity.token.addressAndRPCServer] else { return } |
|
|
|
|
refreshActivity(token: activity.token, tokenHolder: tokenHolders[0], activity: activity) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private func combineActivitiesWithTransactions() { |
|
|
|
|
Task { @MainActor in |
|
|
|
|
let transactions = await transactionDataStore.transactions(forFilter: transactionsFilterStrategy, servers: Array(sessionsProvider.activeSessions.keys), oldestBlockNumber: activities.last?.blockNumber) |
|
|
|
|
let items = await combine(activities: activities.all, with: transactions) |
|
|
|
|
let items = await combine(activities: activities, with: transactions) |
|
|
|
|
let activities = ActivityCollection.sorted(activities: items) |
|
|
|
|
activitiesSubject.send(activities) |
|
|
|
|
} |
|
|
|
@ -204,7 +200,7 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
} else if transaction.localizedOperations.count == 1 { |
|
|
|
|
return [.standaloneTransaction(transaction: transaction, activity: activity)] |
|
|
|
|
} else { |
|
|
|
|
let isSwap = self.isSwap(activities: activities.all, operations: transaction.localizedOperations, wallet: wallet) |
|
|
|
|
let isSwap = self.isSwap(activities: activities, operations: transaction.localizedOperations, wallet: wallet) |
|
|
|
|
var results: [ActivityRowModel] = .init() |
|
|
|
|
results.append(.parentTransaction(transaction: transaction, isSwap: isSwap, activities: .init())) |
|
|
|
|
results.append(contentsOf: await transaction.localizedOperations.asyncMap { |
|
|
|
@ -227,26 +223,30 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
return hasSend && hasReceive |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
//Important to pass in the `TokenHolder` instance and not re-create so that we don't override the subscribable values for the token with ones that are not resolved yet |
|
|
|
|
private func refreshActivity(token: Token, tokenHolder: TokenHolder, activity: Activity) { |
|
|
|
|
let attributeValues = AssetAttributeValues(attributeValues: tokenHolder.values) |
|
|
|
|
cancellableSet[token.addressAndRPCServer] = attributeValues.resolveAllAttributes() |
|
|
|
|
.sink(receiveValue: { [weak self] resolvedAttributeNameValues in |
|
|
|
|
guard let stronSelf = self else { return } |
|
|
|
|
|
|
|
|
|
//NOTE: Fix crush when element with index out of range |
|
|
|
|
if let (index, oldActivity) = stronSelf.activitiesIndexLookup[activity.id] { |
|
|
|
|
private func updatedActivityWithResolvedValues(token: Token, tokenHolder: TokenHolder, activity: Activity, resolvedAttributeNameValues: [AttributeId: AssetInternalValue]) { |
|
|
|
|
//NOTE: Fix crash when element with index out of range |
|
|
|
|
if let (index, oldActivity) = activitiesIndexLookup[activity.id] { |
|
|
|
|
let updatedValues = (token: oldActivity.values.token.merging(resolvedAttributeNameValues) { _, new in new }, card: oldActivity.values.card) |
|
|
|
|
let updatedActivity: Activity = .init(id: oldActivity.id, rowType: oldActivity.rowType, token: token, server: oldActivity.server, name: oldActivity.name, eventName: oldActivity.eventName, blockNumber: oldActivity.blockNumber, transactionId: oldActivity.transactionId, transactionIndex: oldActivity.transactionIndex, logIndex: oldActivity.logIndex, date: oldActivity.date, values: updatedValues, view: oldActivity.view, itemView: oldActivity.itemView, isBaseCard: oldActivity.isBaseCard, state: oldActivity.state) |
|
|
|
|
|
|
|
|
|
if stronSelf.activities.contains(index: index) { |
|
|
|
|
stronSelf.activities[index] = updatedActivity |
|
|
|
|
|
|
|
|
|
stronSelf.didUpdateActivitySubject.send(updatedActivity) |
|
|
|
|
if activities.indices.contains(index) { |
|
|
|
|
activities[index] = updatedActivity |
|
|
|
|
didUpdateActivitySubject.send(updatedActivity) |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
//no-op. We should be able to find it unless the list of activities has changed |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
//Important to pass in the `TokenHolder` instance and not re-create so that we don't override the subscribable values for the token with ones that are not resolved yet |
|
|
|
|
private func refreshActivity(token: Token, tokenHolder: TokenHolder, activity: Activity) { |
|
|
|
|
let attributeValues = AssetAttributeValues(attributeValues: tokenHolder.values) |
|
|
|
|
cancellableSet[token.addressAndRPCServer] = attributeValues.resolveAllAttributes() |
|
|
|
|
.sink(receiveValue: { [weak self] resolvedAttributeNameValues in |
|
|
|
|
guard let strongSelf = self else { return } |
|
|
|
|
Task { [weak self] in |
|
|
|
|
await self?.updatedActivityWithResolvedValues(token: token, tokenHolder: tokenHolder, activity: activity, resolvedAttributeNameValues: resolvedAttributeNameValues) |
|
|
|
|
} |
|
|
|
|
}) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -256,7 +256,7 @@ public class ActivitiesService: ActivitiesServiceType { |
|
|
|
|
for (index, each) in activities.enumerated() { |
|
|
|
|
newValue[each.id] = (index, each) |
|
|
|
|
} |
|
|
|
|
activitiesIndexLookup.set(value: newValue) |
|
|
|
|
activitiesIndexLookup = newValue |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|