diff --git a/Signal.xcodeproj/project.pbxproj b/Signal.xcodeproj/project.pbxproj index 0eafa395b9..93ddf0c6bb 100644 --- a/Signal.xcodeproj/project.pbxproj +++ b/Signal.xcodeproj/project.pbxproj @@ -388,7 +388,6 @@ 452C468F1E427E200087B011 /* OutboundCallInitiator.swift in Sources */ = {isa = PBXBuildFile; fileRef = 452C468E1E427E200087B011 /* OutboundCallInitiator.swift */; }; 452D1AF12081059C00A67F7F /* StringAdditionsTest.swift in Sources */ = {isa = PBXBuildFile; fileRef = 452D1AF02081059C00A67F7F /* StringAdditionsTest.swift */; }; 452EC6DF205E9E30000E787C /* MediaGallery.swift in Sources */ = {isa = PBXBuildFile; fileRef = 452EC6DE205E9E30000E787C /* MediaGallery.swift */; }; - 452ECA4D1E087E7200E2F016 /* MessageFetcherJob.swift in Sources */ = {isa = PBXBuildFile; fileRef = 452ECA4C1E087E7200E2F016 /* MessageFetcherJob.swift */; }; 4535186B1FC635DD00210559 /* ShareViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 4535186A1FC635DD00210559 /* ShareViewController.swift */; }; 4535186E1FC635DD00210559 /* MainInterface.storyboard in Resources */ = {isa = PBXBuildFile; fileRef = 4535186C1FC635DD00210559 /* MainInterface.storyboard */; }; 453518721FC635DD00210559 /* SignalShareExtension.appex in Embed App Extensions */ = {isa = PBXBuildFile; fileRef = 453518681FC635DD00210559 /* SignalShareExtension.appex */; settings = {ATTRIBUTES = (RemoveHeadersOnCopy, ); }; }; @@ -1237,7 +1236,6 @@ 452C468E1E427E200087B011 /* OutboundCallInitiator.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = OutboundCallInitiator.swift; sourceTree = ""; }; 452D1AF02081059C00A67F7F /* StringAdditionsTest.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = StringAdditionsTest.swift; sourceTree = ""; }; 452EC6DE205E9E30000E787C /* MediaGallery.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MediaGallery.swift; sourceTree = ""; }; - 452ECA4C1E087E7200E2F016 /* MessageFetcherJob.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = MessageFetcherJob.swift; sourceTree = ""; }; 453518681FC635DD00210559 /* SignalShareExtension.appex */ = {isa = PBXFileReference; explicitFileType = "wrapper.app-extension"; includeInIndex = 0; path = SignalShareExtension.appex; sourceTree = BUILT_PRODUCTS_DIR; }; 4535186A1FC635DD00210559 /* ShareViewController.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ShareViewController.swift; sourceTree = ""; }; 4535186D1FC635DD00210559 /* Base */ = {isa = PBXFileReference; lastKnownFileType = file.storyboard; name = Base; path = Base.lproj/MainInterface.storyboard; sourceTree = ""; }; @@ -2640,7 +2638,6 @@ children = ( 4C9C50FF22F495F60054A33F /* BroadcastMediaMessageJob.swift */, 4CC0B59B20EC5F2E00CF6EE0 /* ConversationConfigurationSyncOperation.swift */, - 452ECA4C1E087E7200E2F016 /* MessageFetcherJob.swift */, 45D231761DC7E8F10034FA89 /* SessionResetJob.swift */, 45CD81EE1DC030E7004C9430 /* SyncPushTokensJob.swift */, ); @@ -4292,7 +4289,6 @@ 343A65951FC47D5E000477A1 /* DebugUISyncMessages.m in Sources */, 4C63550222F15A6700A8ECE6 /* ThemeHeaderView.swift in Sources */, 45C0DC1E1E69011F00E04C47 /* UIStoryboard+OWS.swift in Sources */, - 452ECA4D1E087E7200E2F016 /* MessageFetcherJob.swift in Sources */, 4556FA681F54AA9500AF40DD /* DebugUIProfile.swift in Sources */, 4C2EBB7F2356B2B900BBC171 /* SecondaryLinkingSetDeviceNameViewController.swift in Sources */, 4C30E224234F9F34009558B7 /* SecondaryLinkingPrepViewController.swift in Sources */, diff --git a/Signal/src/AppDelegate.m b/Signal/src/AppDelegate.m index 5ce64a69d0..05e940d4e2 100644 --- a/Signal/src/AppDelegate.m +++ b/Signal/src/AppDelegate.m @@ -186,6 +186,11 @@ NSString *NSStringForLaunchFailure(LaunchFailure launchFailure) return Environment.shared.launchJobs; } +- (nullable MessageFetcherJob *)messageFetcherJob +{ + return SSKEnvironment.shared.messageFetcherJob; +} + #pragma mark - - (void)applicationDidEnterBackground:(UIApplication *)application @@ -705,7 +710,7 @@ NSString *NSStringForLaunchFailure(LaunchFailure launchFailure) dispatch_async(dispatch_get_main_queue(), ^{ [self.socketManager requestSocketOpen]; [Environment.shared.contactsManager fetchSystemContactsOnceIfAlreadyAuthorized]; - [[AppEnvironment.shared.messageFetcherJob run] retainUntilComplete]; + [[self.messageFetcherJob runObjc] retainUntilComplete]; if (![UIApplication sharedApplication].isRegisteredForRemoteNotifications) { OWSLogInfo(@"Retrying to register for remote notifications since user hasn't registered yet."); @@ -1118,7 +1123,7 @@ NSString *NSStringForLaunchFailure(LaunchFailure launchFailure) } [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - [[AppEnvironment.shared.messageFetcherJob run] retainUntilComplete]; + [[self.messageFetcherJob runObjc] retainUntilComplete]; }]; } @@ -1137,7 +1142,7 @@ NSString *NSStringForLaunchFailure(LaunchFailure launchFailure) } [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - [[AppEnvironment.shared.messageFetcherJob run] retainUntilComplete]; + [[self.messageFetcherJob runObjc] retainUntilComplete]; dispatch_after(dispatch_time(DISPATCH_TIME_NOW, 20 * NSEC_PER_SEC), dispatch_get_main_queue(), ^{ completionHandler(UIBackgroundFetchResultNewData); }); @@ -1149,7 +1154,7 @@ NSString *NSStringForLaunchFailure(LaunchFailure launchFailure) { OWSLogInfo(@"performing background fetch"); [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - __block AnyPromise *job = [AppEnvironment.shared.messageFetcherJob run].then(^{ + __block AnyPromise *job = [self.messageFetcherJob runObjc].then(^{ // HACK: Call completion handler after n seconds. // // We don't currently have a convenient API to know when message fetching is *done* when @@ -1228,7 +1233,7 @@ NSString *NSStringForLaunchFailure(LaunchFailure launchFailure) // Fetch messages as soon as possible after launching. In particular, when // launching from the background, without this, we end up waiting some extra // seconds before receiving an actionable push notification. - [[AppEnvironment.shared.messageFetcherJob run] retainUntilComplete]; + [[self.messageFetcherJob runObjc] retainUntilComplete]; // This should happen at any launch, background or foreground. __unused AnyPromise *pushTokenpromise = diff --git a/Signal/src/Jobs/BroadcastMediaMessageJob.swift b/Signal/src/Jobs/BroadcastMediaMessageJob.swift index 03a9d54841..dfa3bdc7db 100644 --- a/Signal/src/Jobs/BroadcastMediaMessageJob.swift +++ b/Signal/src/Jobs/BroadcastMediaMessageJob.swift @@ -36,6 +36,10 @@ public class BroadcastMediaMessageJobQueue: NSObject, JobQueue { // no special handling } + public func didFlushQueue(transaction: SDSAnyWriteTransaction) { + // Do nothing. + } + let defaultQueue: OperationQueue = { let operationQueue = OperationQueue() operationQueue.name = "BroadcastMediaMessageJobQueue" diff --git a/Signal/src/Jobs/SessionResetJob.swift b/Signal/src/Jobs/SessionResetJob.swift index 0190d3a817..c222f320ac 100644 --- a/Signal/src/Jobs/SessionResetJob.swift +++ b/Signal/src/Jobs/SessionResetJob.swift @@ -47,6 +47,10 @@ public class SessionResetJobQueue: NSObject, JobQueue { // no special handling } + public func didFlushQueue(transaction: SDSAnyWriteTransaction) { + // Do nothing. + } + let operationQueue: OperationQueue = { // no need to serialize the operation queuing, since sending will ultimately be serialized by MessageSender let operationQueue = OperationQueue() diff --git a/Signal/src/ViewControllers/AppSettings/AdvancedSettingsTableViewController.m b/Signal/src/ViewControllers/AppSettings/AdvancedSettingsTableViewController.m index 8fbcecdc5d..30517bda39 100644 --- a/Signal/src/ViewControllers/AppSettings/AdvancedSettingsTableViewController.m +++ b/Signal/src/ViewControllers/AppSettings/AdvancedSettingsTableViewController.m @@ -145,7 +145,7 @@ NS_ASSUME_NONNULL_BEGIN OWSTableSection *censorshipSection = [OWSTableSection new]; censorshipSection.headerTitle = NSLocalizedString(@"SETTINGS_ADVANCED_CENSORSHIP_CIRCUMVENTION_HEADER", @"Table header for the 'censorship circumvention' section."); - BOOL isAnySocketOpen = TSSocketManager.shared.highestSocketState == OWSWebSocketStateOpen; + BOOL isAnySocketOpen = TSSocketManager.shared.socketState == OWSWebSocketStateOpen; if (OWSSignalService.sharedInstance.hasCensoredPhoneNumber) { if (OWSSignalService.sharedInstance.isCensorshipCircumventionManuallyDisabled) { censorshipSection.footerTitle @@ -195,7 +195,7 @@ NS_ASSUME_NONNULL_BEGIN return YES; } else if (service.hasCensoredPhoneNumber && service.isCensorshipCircumventionManuallyDisabled) { return YES; - } else if (TSSocketManager.shared.highestSocketState == OWSWebSocketStateOpen) { + } else if (TSSocketManager.shared.socketState == OWSWebSocketStateOpen) { return NO; } else { return reachability.isReachable; diff --git a/Signal/src/ViewControllers/AppSettings/AppSettingsViewController.m b/Signal/src/ViewControllers/AppSettings/AppSettingsViewController.m index 85d23bfd82..6b59a36045 100644 --- a/Signal/src/ViewControllers/AppSettings/AppSettingsViewController.m +++ b/Signal/src/ViewControllers/AppSettings/AppSettingsViewController.m @@ -161,7 +161,7 @@ @"Error indicating that this device is no longer linked."); accessoryLabel.textColor = UIColor.ows_accentRedColor; } else { - switch (TSSocketManager.shared.highestSocketState) { + switch (TSSocketManager.shared.socketState) { case OWSWebSocketStateClosed: accessoryLabel.text = NSLocalizedString(@"NETWORK_STATUS_OFFLINE", @""); accessoryLabel.textColor = UIColor.ows_accentRedColor; diff --git a/Signal/src/ViewControllers/HomeView/ConversationListViewController.m b/Signal/src/ViewControllers/HomeView/ConversationListViewController.m index ebff6ddf7a..7cd2fb531e 100644 --- a/Signal/src/ViewControllers/HomeView/ConversationListViewController.m +++ b/Signal/src/ViewControllers/HomeView/ConversationListViewController.m @@ -161,6 +161,11 @@ NSString *const kArchiveButtonPseudoGroup = @"kArchiveButtonPseudoGroup"; return SSKEnvironment.shared.primaryStorage; } +- (nullable MessageFetcherJob *)messageFetcherJob +{ + return SSKEnvironment.shared.messageFetcherJob; +} + #pragma mark - - (void)observeNotifications @@ -1377,16 +1382,18 @@ NSString *const kArchiveButtonPseudoGroup = @"kArchiveButtonPseudoGroup"; OWSAssertIsOnMainThread(); OWSLogInfo(@"beggining refreshing."); - [[AppEnvironment.shared.messageFetcherJob run].then(^{ - if (TSAccountManager.sharedInstance.isRegisteredPrimaryDevice) { - return [AnyPromise promiseWithValue:nil]; - } + [[self.messageFetcherJob runObjc] + .then(^{ + if (TSAccountManager.sharedInstance.isRegisteredPrimaryDevice) { + return [AnyPromise promiseWithValue:nil]; + } - return [SSKEnvironment.shared.syncManager sendAllSyncRequestMessagesWithTimeout:20]; - }).ensure(^{ - OWSLogInfo(@"ending refreshing."); - [refreshControl endRefreshing]; - }) retainUntilComplete]; + return [SSKEnvironment.shared.syncManager sendAllSyncRequestMessagesWithTimeout:20]; + }) + .ensure(^{ + OWSLogInfo(@"ending refreshing."); + [refreshControl endRefreshing]; + }) retainUntilComplete]; } #pragma mark - Edit Actions diff --git a/Signal/src/environment/AppEnvironment.swift b/Signal/src/environment/AppEnvironment.swift index cdd0a6c8f1..666c552b6a 100644 --- a/Signal/src/environment/AppEnvironment.swift +++ b/Signal/src/environment/AppEnvironment.swift @@ -1,5 +1,5 @@ // -// Copyright (c) 2019 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // import Foundation @@ -34,9 +34,6 @@ import SignalMessaging @objc public var outboundCallInitiator: OutboundCallInitiator - @objc - public var messageFetcherJob: MessageFetcherJob - @objc public var accountManager: AccountManager @@ -65,7 +62,6 @@ import SignalMessaging self.callMessageHandler = WebRTCCallMessageHandler() self.callService = CallService() self.outboundCallInitiator = OutboundCallInitiator() - self.messageFetcherJob = MessageFetcherJob() self.accountManager = AccountManager() self.notificationPresenter = NotificationPresenter() self.pushRegistrationManager = PushRegistrationManager() diff --git a/Signal/src/environment/PushRegistrationManager.swift b/Signal/src/environment/PushRegistrationManager.swift index 5fe2eadd3e..3176adddb1 100644 --- a/Signal/src/environment/PushRegistrationManager.swift +++ b/Signal/src/environment/PushRegistrationManager.swift @@ -1,5 +1,5 @@ // -// Copyright (c) 2019 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // import Foundation @@ -22,7 +22,7 @@ public enum PushRegistrationError: Error { // MARK: - Dependencies private var messageFetcherJob: MessageFetcherJob { - return AppEnvironment.shared.messageFetcherJob + return SSKEnvironment.shared.messageFetcherJob } private var notificationPresenter: NotificationPresenter { @@ -110,7 +110,7 @@ public enum PushRegistrationError: Error { preauthChallengeResolver.fulfill(challenge) self.preauthChallengeResolver = nil } else { - (self.messageFetcherJob.run() as Promise).retainUntilComplete() + self.messageFetcherJob.run().promise.retainUntilComplete() } } } diff --git a/Signal/src/util/MainAppContext.m b/Signal/src/util/MainAppContext.m index 5be1cfa4a8..f317597c43 100644 --- a/Signal/src/util/MainAppContext.m +++ b/Signal/src/util/MainAppContext.m @@ -404,6 +404,11 @@ NSString *const ReportedApplicationStateDidChangeNotification = @"ReportedApplic return YES; } +- (BOOL)shouldProcessIncomingMessages +{ + return YES; +} + @end NS_ASSUME_NONNULL_END diff --git a/SignalMessaging/contacts/IncomingContactSyncJobQueue.swift b/SignalMessaging/contacts/IncomingContactSyncJobQueue.swift index 2c2c928dd4..3fc1b8d0a4 100644 --- a/SignalMessaging/contacts/IncomingContactSyncJobQueue.swift +++ b/SignalMessaging/contacts/IncomingContactSyncJobQueue.swift @@ -47,6 +47,10 @@ public class IncomingContactSyncJobQueue: NSObject, JobQueue { // no special handling } + public func didFlushQueue(transaction: SDSAnyWriteTransaction) { + // Do nothing. + } + let defaultQueue: OperationQueue = { let operationQueue = OperationQueue() operationQueue.name = "IncomingContactSyncJobQueue" diff --git a/SignalMessaging/contacts/OWSGroupSyncProcessingJobQueue.swift b/SignalMessaging/contacts/OWSGroupSyncProcessingJobQueue.swift index c2701d5c0f..8e75952249 100644 --- a/SignalMessaging/contacts/OWSGroupSyncProcessingJobQueue.swift +++ b/SignalMessaging/contacts/OWSGroupSyncProcessingJobQueue.swift @@ -41,6 +41,10 @@ public class IncomingGroupSyncJobQueue: NSObject, JobQueue { // no special handling } + public func didFlushQueue(transaction: SDSAnyWriteTransaction) { + // Do nothing. + } + let defaultQueue: OperationQueue = { let operationQueue = OperationQueue() operationQueue.name = "IncomingGroupSyncJobQueue" diff --git a/SignalMessaging/environment/AppSetup.m b/SignalMessaging/environment/AppSetup.m index 46f7b9fa41..db310d0ad4 100644 --- a/SignalMessaging/environment/AppSetup.m +++ b/SignalMessaging/environment/AppSetup.m @@ -109,6 +109,8 @@ NS_ASSUME_NONNULL_BEGIN OWSSounds *sounds = [OWSSounds new]; id proximityMonitoringManager = [OWSProximityMonitoringManagerImpl new]; OWSWindowManager *windowManager = [[OWSWindowManager alloc] initDefault]; + MessageProcessing *messageProcessing = [MessageProcessing new]; + MessageFetcherJob *messageFetcherJob = [MessageFetcherJob new]; [Environment setShared:[[Environment alloc] initWithAudioSession:audioSession incomingContactSyncJobQueue:incomingContactSyncJobQueue @@ -159,7 +161,9 @@ NS_ASSUME_NONNULL_BEGIN storageServiceManager:storageServiceManager storageCoordinator:storageCoordinator sskPreferences:sskPreferences - groupsV2:groupsV2]]; + groupsV2:groupsV2 + messageProcessing:messageProcessing + messageFetcherJob:messageFetcherJob]]; appSpecificSingletonBlock(); diff --git a/Signal/src/Jobs/MessageFetcherJob.swift b/SignalServiceKit/src/Messages/MessageFetcherJob.swift similarity index 55% rename from Signal/src/Jobs/MessageFetcherJob.swift rename to SignalServiceKit/src/Messages/MessageFetcherJob.swift index 5d62251f1c..a3d59d12a0 100644 --- a/Signal/src/Jobs/MessageFetcherJob.swift +++ b/SignalServiceKit/src/Messages/MessageFetcherJob.swift @@ -1,12 +1,30 @@ // -// Copyright (c) 2019 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // import Foundation import PromiseKit -import SignalServiceKit -@objc(OWSMessageFetcherJob) +// This token can be used to observe the completion of a given fetch cycle. +public struct MessageFetchCycle: Hashable, Equatable { + public let uuid = UUID() + public let promise: Promise + + // MARK: Hashable + + public func hash(into hasher: inout Hasher) { + hasher.combine(uuid) + } + + // MARK: Equatable + + public static func == (lhs: MessageFetchCycle, rhs: MessageFetchCycle) -> Bool { + return lhs.uuid == rhs.uuid + } +} + +// MARK: - + public class MessageFetcherJob: NSObject { private var timer: Timer? @@ -20,43 +38,154 @@ public class MessageFetcherJob: NSObject { // MARK: Singletons - private var networkManager: TSNetworkManager { + private class var networkManager: TSNetworkManager { return SSKEnvironment.shared.networkManager } - private var messageReceiver: OWSMessageReceiver { + private class var messageReceiver: OWSMessageReceiver { return SSKEnvironment.shared.messageReceiver } - private var signalService: OWSSignalService { + private class var signalService: OWSSignalService { return OWSSignalService.sharedInstance() } - private var tsAccountManager: TSAccountManager { + private class var tsAccountManager: TSAccountManager { return TSAccountManager.sharedInstance() } - // MARK: + // MARK: - + + // This operation queue ensures that only one fetch operation is + // running at a given time. + private let operationQueue: OperationQueue = { + let operationQueue = OperationQueue() + operationQueue.name = "MessageFetcherJob.operationQueue" + operationQueue.maxConcurrentOperationCount = 1 + return operationQueue + }() + + fileprivate var activeOperationCount: Int { + return operationQueue.operationCount + } + + private let serialQueue = DispatchQueue(label: "org.signal.messageFetcherJob.serialQueue") + + private let completionQueue = DispatchQueue(label: "org.signal.messageFetcherJob.completionQueue") + + // This property should only be accessed on serialQueue. + private var activeFetchCycles = Set() + + // This property should only be accessed on serialQueue. + private var completedFetchCyclesCounter: UInt = 0 + + @objc + public static let didChangeStateNotificationName = Notification.Name("MessageFetcherJob.didChangeStateNotificationName") @discardableResult - public func run() -> Promise { + public func run() -> MessageFetchCycle { + Logger.debug("") + + // Use an operation queue to ensure that only one fetch cycle is done + // at a time. + let operation = MessageFetchOperation() + let promise = operation.promise + let fetchCycle = MessageFetchCycle(promise: promise) + + _ = self.serialQueue.sync { + activeFetchCycles.insert(fetchCycle.uuid) + } + + operationQueue.addOperation(operation) + + promise.retainUntilComplete() + + completionQueue.async { + self.operationQueue.waitUntilAllOperationsAreFinished() + + _ = self.serialQueue.sync { + self.activeFetchCycles.remove(fetchCycle.uuid) + self.completedFetchCyclesCounter += 1 + } + + self.postDidChangeState() + } + + self.postDidChangeState() + + return fetchCycle + } + + @objc + @discardableResult + public func runObjc() -> AnyPromise { + return AnyPromise(run().promise) + } + + private func postDidChangeState() { + NotificationCenter.default.postNotificationNameAsync(MessageFetcherJob.didChangeStateNotificationName, object: nil) + } + + public func isFetchCycleComplete(fetchCycle: MessageFetchCycle) -> Bool { + return self.serialQueue.sync { + return self.activeFetchCycles.contains(fetchCycle.uuid) + } + } + + public var areAllFetchCyclesComplete: Bool { + return self.serialQueue.sync { + return self.activeFetchCycles.isEmpty + } + } + + public var completedRestFetches: UInt { + return self.serialQueue.sync { + return self.completedFetchCyclesCounter + } + } + + public class var shouldUseWebSocket: Bool { + return CurrentAppContext().isMainApp && !signalService.isCensorshipCircumventionActive + } + + // MARK: - + + fileprivate class func fetchMessages(resolver: Resolver) { Logger.debug("") guard tsAccountManager.isRegisteredAndReady else { assert(AppReadiness.isAppReady()) Logger.warn("not registered") - return Promise.value(()) + return resolver.fulfill(()) } - guard signalService.isCensorshipCircumventionActive else { + if shouldUseWebSocket { Logger.debug("delegating message fetching to SocketManager since we're using normal transport.") TSSocketManager.shared.requestSocketOpen() - return Promise.value(()) + return resolver.fulfill(()) + } else if CurrentAppContext().shouldProcessIncomingMessages { + // Main app should use REST if censorship circumvention is active. + // Notification extension that should always use REST. + } else { + return resolver.reject(OWSAssertionError("App extensions should not fetch messages.")) } - Logger.info("fetching messages via REST.") + Logger.info("Fetching messages via REST.") - let promise = self.fetchUndeliveredMessages().then { (envelopes: [SSKProtoEnvelope], more: Bool) -> Promise in + fetchMessagesViaRest() + .done { + resolver.fulfill(()) + }.catch { error in + resolver.reject(error) + }.retainUntilComplete() + } + + // MARK: - + + private class func fetchMessagesViaRest() -> Promise { + Logger.debug("") + + return fetchBatchViaRest().then { (envelopes: [SSKProtoEnvelope], more: Bool) -> Promise in for envelope in envelopes { Logger.info("received envelope.") do { @@ -70,30 +199,23 @@ public class MessageFetcherJob: NSObject { if more { Logger.info("fetching more messages.") - return self.run() + + return self.fetchMessagesViaRest() } else { // All finished return Promise.value(()) } } - - promise.retainUntilComplete() - - return promise } - @objc - @discardableResult - public func run() -> AnyPromise { - return AnyPromise(run() as Promise) - } + // MARK: - Run Loop // use in DEBUG or wherever you can't receive push notifications to poll for messages. // Do not use in production. public func startRunLoop(timeInterval: Double) { Logger.error("Starting message fetch polling. This should not be used in production.") timer = WeakTimer.scheduledTimer(timeInterval: timeInterval, target: self, userInfo: nil, repeats: true) {[weak self] _ in - let _: Promise? = self?.run() + _ = self?.run() return } } @@ -103,9 +225,11 @@ public class MessageFetcherJob: NSObject { timer = nil } - private func parseMessagesResponse(responseObject: Any?) -> (envelopes: [SSKProtoEnvelope], more: Bool)? { + // MARK: - + + private class func parseMessagesResponse(responseObject: Any?) -> (envelopes: [SSKProtoEnvelope], more: Bool)? { guard let responseObject = responseObject else { - Logger.error("response object was surpringly nil") + Logger.error("response object was unexpectedly nil") return nil } @@ -136,7 +260,7 @@ public class MessageFetcherJob: NSObject { ) } - private func buildEnvelope(messageDict: [String: Any]) -> SSKProtoEnvelope? { + private class func buildEnvelope(messageDict: [String: Any]) -> SSKProtoEnvelope? { do { let params = ParamParser(dictionary: messageDict) @@ -182,7 +306,7 @@ public class MessageFetcherJob: NSObject { } } - private func fetchUndeliveredMessages() -> Promise<(envelopes: [SSKProtoEnvelope], more: Bool)> { + private class func fetchBatchViaRest() -> Promise<(envelopes: [SSKProtoEnvelope], more: Bool)> { return Promise { resolver in let request = OWSRequestFactory.getMessagesRequest() self.networkManager.makeRequest( @@ -206,7 +330,7 @@ public class MessageFetcherJob: NSObject { } } - private func acknowledgeDelivery(envelope: SSKProtoEnvelope) { + private class func acknowledgeDelivery(envelope: SSKProtoEnvelope) { let request: TSRequest if let serverGuid = envelope.serverGuid, serverGuid.count > 0 { request = OWSRequestFactory.acknowledgeMessageDeliveryRequest(withServerGuid: serverGuid) @@ -226,3 +350,27 @@ public class MessageFetcherJob: NSObject { }) } } + +// MARK: - + +private class MessageFetchOperation: OWSOperation { + + let promise: Promise + let resolver: Resolver + + override required init() { + + let (promise, resolver) = Promise.pending() + self.promise = promise + self.resolver = resolver + } + + public override func run() { + Logger.debug("") + + MessageFetcherJob.fetchMessages(resolver: resolver) + promise.ensure { + self.reportSuccess() + }.retainUntilComplete() + } +} diff --git a/SignalServiceKit/src/Messages/MessageProcessing.swift b/SignalServiceKit/src/Messages/MessageProcessing.swift new file mode 100644 index 0000000000..beac24c0a2 --- /dev/null +++ b/SignalServiceKit/src/Messages/MessageProcessing.swift @@ -0,0 +1,415 @@ +// +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. +// + +import Foundation +import PromiseKit + +@objc +public class MessageProcessing: NSObject { + + // MARK: - Dependencies + + private var databaseStorage: SDSDatabaseStorage { + return SDSDatabaseStorage.shared + } + + private var messageReceiver: OWSMessageReceiver { + return SSKEnvironment.shared.messageReceiver + } + + private var batchMessageProcessor: OWSBatchMessageProcessor { + return SSKEnvironment.shared.batchMessageProcessor + } + + private var socketManager: TSSocketManager { + return TSSocketManager.shared + } + + private var messageFetcherJob: MessageFetcherJob { + return SSKEnvironment.shared.messageFetcherJob + } + + private var tsAccountManager: TSAccountManager { + return TSAccountManager.sharedInstance() + } + + private var signalService: OWSSignalService { + return OWSSignalService.sharedInstance() + } + + // MARK: - + + private let serialQueue = DispatchQueue(label: "org.signal.MessageProcessing") + + public override init() { + super.init() + + SwiftSingletons.register(self) + + NotificationCenter.default.addObserver(self, + selector: #selector(messageDecryptionDidFlushQueue), + name: NSNotification.Name.messageDecryptionDidFlushQueue, + object: nil) + NotificationCenter.default.addObserver(self, + selector: #selector(messageProcessingDidFlushQueue), + name: NSNotification.Name.messageProcessingDidFlushQueue, + object: nil) + NotificationCenter.default.addObserver(self, + selector: #selector(webSocketStateDidChange), + name: NSNotification.Name.webSocketStateDidChange, + object: nil) + NotificationCenter.default.addObserver(self, + selector: #selector(messageFetcherJobDidChangeState), + name: MessageFetcherJob.didChangeStateNotificationName, + object: nil) + NotificationCenter.default.addObserver(self, + selector: #selector(registrationStateDidChange), + name: .RegistrationStateDidChange, + object: nil) + } + + // MARK: - Flush decryption and message processing + + // This promise can be used by the notification extension to + // block on decryption and processing of any messages + // received before this promise is created. + // + // TODO: MessageManager uses dispatch_async() to finish + // handling certain kinds of messages outside of the + // "message processing" transaction. This isn't + // reflected in this promise yet. + @objc + public func flushMessageDecryptionAndProcessingPromise() -> AnyPromise { + // GroupsV2 TODO: Make sure the "groups v2" processing queue is flushed as well. + return AnyPromise(decryptStepPromise().then { _ in + return self.processingStepPromise() + }) + } + + // MARK: - Decrypt Step + + // This should only be accessed on serialQueue. + private var decryptStepResolvers = [Resolver]() + + private func decryptStepPromise() -> Promise { + let (promise, resolver) = Promise.pending() + + serialQueue.async { + self.decryptStepResolvers.append(resolver) + + self.tryToResolveDecryptStepPromises() + } + + return promise + } + + private func tryToResolveDecryptStepPromises() { + assertOnQueue(serialQueue) + + let decryptStepResolvers = self.decryptStepResolvers + guard !decryptStepResolvers.isEmpty else { + // No pending resolvers to resolve. + return + } + + let hasPendingJobs = databaseStorage.read { transaction in + return self.isDecryptingIncomingMessages(transaction: transaction) + } + guard !hasPendingJobs else { + return + } + + self.decryptStepResolvers = [] + + for resolver in decryptStepResolvers { + resolver.fulfill(()) + } + } + + private func isDecryptingIncomingMessages(transaction: SDSAnyReadTransaction) -> Bool { + return messageReceiver.hasPendingJobs(with: transaction) + } + + @objc + fileprivate func messageDecryptionDidFlushQueue() { + AssertIsOnMainThread() + + serialQueue.async { + self.tryToResolveDecryptStepPromises() + self.tryToResolveAllMessageFetchingAndProcessingPromises() + } + } + + // MARK: - Processing Step + + // This should only be accessed on serialQueue. + private var processingStepResolvers = [Resolver]() + + private func processingStepPromise() -> Promise { + let (promise, resolver) = Promise.pending() + + serialQueue.async { + self.processingStepResolvers.append(resolver) + + self.tryToResolveProcessingStepPromises() + } + + return promise + } + + private func tryToResolveProcessingStepPromises() { + assertOnQueue(serialQueue) + + let processingStepResolvers = self.processingStepResolvers + guard !processingStepResolvers.isEmpty else { + // No pending resolvers to resolve. + return + } + + let hasPendingJobs = databaseStorage.read { transaction in + return self.isProcessingIncomingMessages(transaction: transaction) + } + guard !hasPendingJobs else { + return + } + + self.processingStepResolvers = [] + + for resolver in processingStepResolvers { + resolver.fulfill(()) + } + } + + private func isProcessingIncomingMessages(transaction: SDSAnyReadTransaction) -> Bool { + return batchMessageProcessor.hasPendingJobs(with: transaction) + } + + @objc + fileprivate func messageProcessingDidFlushQueue() { + AssertIsOnMainThread() + + serialQueue.async { + self.tryToResolveProcessingStepPromises() + self.tryToResolveAllMessageFetchingAndProcessingPromises() + } + } + + // MARK: - WebSocket drained + + // This should only be accessed on serialQueue. + private var websocketDrainedResolvers = [UUID: Resolver]() + + // This promise can be used by the notification extension + // to detect when the websocket has drained its queue. + // + // TODO: The notification extension will eventually use + // REST (not the websocket) to receive messages. At that + // time, we'll want to add restMessageFetchingCompletePromiseObjc() + // to this class. We'll probably still need this + // websocketDrainedPromiseObjc() for usage by the main app. + @objc + public func websocketDrainedPromiseObjc() -> AnyPromise { + return AnyPromise(websocketDrainedPromise()) + } + + private func websocketDrainedPromise() -> Promise { + let (promise, resolver) = Promise.pending() + + serialQueue.async { + self.websocketDrainedResolvers[UUID()] = resolver + + self.tryToResolveWebsocketDrainedPromises() + } + + return promise + } + + private func tryToResolveWebsocketDrainedPromises() { + assertOnQueue(serialQueue) + + // We can only access the resolvers on serialQueue, + // but we can only check "isWebsocketDrained" on the main + // thread. Therefore, we first snapshot the current + // set of resolvers on the serialQueue: + let resolverKeys = self.websocketDrainedResolvers.keys + guard !resolverKeys.isEmpty else { + // No pending resolvers to resolve. + return + } + + DispatchQueue.main.async { + // Then we check for isWebsocketDrained of the main thread: + let isWebsocketDrained = (self.socketManager.socketState() == .open && + self.socketManager.hasEmptiedInitialQueue()) + guard isWebsocketDrained else { + return + } + + self.serialQueue.async { + // Lastly, if the websocket is drained, on the serialQueue + // we resolve any resolvers that were present _before_ we + // checked (to avoid races): + for key in resolverKeys { + guard let resolver = self.websocketDrainedResolvers[key] else { + continue + } + self.websocketDrainedResolvers.removeValue(forKey: key) + resolver.fulfill(()) + } + } + } + } + + @objc + fileprivate func webSocketStateDidChange() { + AssertIsOnMainThread() + + serialQueue.async { + self.tryToResolveWebsocketDrainedPromises() + self.tryToResolveAllMessageFetchingAndProcessingPromises() + } + } + + // MARK: - Specific MessageFetchJob completed + + // This should only be accessed on serialQueue. + private var specificMessageFetchJobResolvers = [MessageFetchCycle: Resolver]() + + // This promise can be used by the notification extension + // to detect when the websocket has drained its queue. + // + // TODO: Do we need an obj-c flavor of this method? If so, we'll need + // to convert MessageFetchCycle to a NSObject. + private func specificMessageFetchJobPromise(fetchCycle: MessageFetchCycle) -> Promise { + let (promise, resolver) = Promise.pending() + + serialQueue.async { + self.specificMessageFetchJobResolvers[fetchCycle] = resolver + + self.tryToResolveSpecificMessageFetchJobPromises() + } + + return promise + } + + private func tryToResolveSpecificMessageFetchJobPromises() { + assertOnQueue(serialQueue) + + guard !specificMessageFetchJobResolvers.isEmpty else { + // No pending resolvers to resolve. + return + } + for (fetchCycle, resolver) in specificMessageFetchJobResolvers { + let isFetchComplete = messageFetcherJob.isFetchCycleComplete(fetchCycle: fetchCycle) + guard isFetchComplete else { + continue + } + + specificMessageFetchJobResolvers.removeValue(forKey: fetchCycle) + resolver.fulfill(()) + } + } + + @objc + fileprivate func messageFetcherJobDidChangeState() { + AssertIsOnMainThread() + + serialQueue.async { + self.tryToResolveSpecificMessageFetchJobPromises() + self.tryToResolveAllMessageFetchingAndProcessingPromises() + } + } + + // MARK: - All message processing + + // This should only be accessed on serialQueue. + private var allMessageFetchingAndProcessingResolvers = [Resolver]() + + // This promise can be used by the Groups v2 logic + // to block until all messages are fetched and processed. + public func allMessageFetchingAndProcessingPromise() -> Promise { + let (promise, resolver) = Promise.pending() + + serialQueue.async { + self.allMessageFetchingAndProcessingResolvers.append(resolver) + + self.tryToResolveAllMessageFetchingAndProcessingPromises() + } + + return promise + } + + private func tryToResolveAllMessageFetchingAndProcessingPromises() { + assertOnQueue(serialQueue) + + let resolvers = self.allMessageFetchingAndProcessingResolvers + guard !resolvers.isEmpty else { + // No pending resolvers to resolve. + return + } + guard isAllMessageFetchingAndProcessingComplete else { + // Not complete. + return + } + + self.allMessageFetchingAndProcessingResolvers = [] + + for resolver in resolvers { + resolver.fulfill(()) + } + } + + private var isAllMessageFetchingAndProcessingComplete: Bool { + guard tsAccountManager.isRegisteredAndReady else { + owsFailDebug("Not registered.") + return false + } + // Groupsv2 TODO: We don't want to process incoming messages + // in the share extension, but we need to block on latest + // groups v2 state when sending messages. + guard CurrentAppContext().shouldProcessIncomingMessages else { + owsFailDebug("Should not process incoming messages.") + return false + } + + if MessageFetcherJob.shouldUseWebSocket { + let isWebsocketDrained = (self.socketManager.socketState() == .open && + self.socketManager.hasEmptiedInitialQueue()) + guard isWebsocketDrained else { + return false + } + } else { + guard messageFetcherJob.completedRestFetches > 0 else { + return false + } + } + + guard messageFetcherJob.areAllFetchCyclesComplete else { + return false + } + + let hasPendingDecryptionOrProcess = databaseStorage.read { (transaction: SDSAnyReadTransaction) -> Bool in + guard !self.isDecryptingIncomingMessages(transaction: transaction) else { + return true + } + guard !self.isProcessingIncomingMessages(transaction: transaction) else { + return true + } + return false + } + guard !hasPendingDecryptionOrProcess else { + return false + } + return true + } + + @objc + fileprivate func registrationStateDidChange() { + AssertIsOnMainThread() + + serialQueue.async { + self.tryToResolveAllMessageFetchingAndProcessingPromises() + } + } +} diff --git a/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.h b/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.h index 039bfeb8c4..b0bc74ba27 100644 --- a/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.h +++ b/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.h @@ -1,5 +1,5 @@ // -// Copyright (c) 2019 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // #import "BaseModel.h" @@ -7,9 +7,12 @@ NS_ASSUME_NONNULL_BEGIN @class OWSStorage; +@class SDSAnyReadTransaction; @class SDSAnyWriteTransaction; @class SSKProtoEnvelope; +extern NSNotificationName const kNSNotificationNameMessageProcessingDidFlushQueue; + @interface OWSMessageContentJob : BaseModel @property (nonatomic, readonly) NSDate *createdAt; @@ -22,6 +25,7 @@ NS_ASSUME_NONNULL_BEGIN wasReceivedByUD:(BOOL)wasReceivedByUD NS_DESIGNATED_INITIALIZER; - (nullable instancetype)initWithCoder:(NSCoder *)coder NS_DESIGNATED_INITIALIZER; - (instancetype)initWithUniqueId:(NSString *)uniqueId NS_UNAVAILABLE; +- (instancetype)initWithGrdbId:(int64_t)grdbId uniqueId:(NSString *)uniqueId NS_UNAVAILABLE; - (instancetype)init NS_UNAVAILABLE; // --- CODE GENERATION MARKER @@ -58,6 +62,8 @@ NS_SWIFT_NAME(init(grdbId:uniqueId:createdAt:envelopeData:plaintextData:wasRecei wasReceivedByUD:(BOOL)wasReceivedByUD transaction:(SDSAnyWriteTransaction *)transaction; +- (BOOL)hasPendingJobsWithTransaction:(SDSAnyReadTransaction *)transaction; + @end NS_ASSUME_NONNULL_END diff --git a/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m b/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m index 50b67344ce..cacdb0ea90 100644 --- a/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m +++ b/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m @@ -6,6 +6,7 @@ #import "AppContext.h" #import "AppReadiness.h" #import "NSArray+OWS.h" +#import "NSNotificationCenter+OWS.h" #import "NotificationsProtocol.h" #import "OWSBackgroundTask.h" #import "OWSMessageManager.h" @@ -18,6 +19,9 @@ NS_ASSUME_NONNULL_BEGIN +NSNotificationName const kNSNotificationNameMessageProcessingDidFlushQueue + = @"kNSNotificationNameMessageProcessingDidFlushQueue"; + @implementation OWSMessageContentJob + (NSString *)collection @@ -137,9 +141,7 @@ NS_ASSUME_NONNULL_BEGIN // Start processing. [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - if (CurrentAppContext().isMainApp) { - [self drainQueue]; - } + [self drainQueue]; }]; return self; @@ -193,9 +195,7 @@ NS_ASSUME_NONNULL_BEGIN OWSAssertIsOnMainThread(); [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - if (CurrentAppContext().isMainApp) { - [self drainQueue]; - } + [self drainQueue]; }]; } @@ -226,12 +226,16 @@ NS_ASSUME_NONNULL_BEGIN transaction:transaction]; } +- (BOOL)hasPendingJobsWithTransaction:(SDSAnyReadTransaction *)transaction +{ + return [self.finder jobCountWithTransaction:transaction] > 0; +} + - (void)drainQueue { OWSAssertDebugUnlessRunningTests(AppReadiness.isAppReady); - // Don't process incoming messages in app extensions. - if (!CurrentAppContext().isMainApp) { + if (!CurrentAppContext().shouldProcessIncomingMessages) { return; } if (!self.tsAccountManager.isRegisteredAndReady) { @@ -272,6 +276,12 @@ NS_ASSUME_NONNULL_BEGIN if (batchJobs.count < 1) { self.isDrainingQueue = NO; OWSLogVerbose(@"Queue is drained"); + + [[NSNotificationCenter defaultCenter] + postNotificationNameAsync:kNSNotificationNameMessageProcessingDidFlushQueue + object:nil + userInfo:nil]; + return; } @@ -376,9 +386,7 @@ NS_ASSUME_NONNULL_BEGIN _processingQueue = [OWSMessageContentQueue new]; [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - if (CurrentAppContext().isMainApp) { - [self.processingQueue drainQueue]; - } + [self.processingQueue drainQueue]; }]; return self; @@ -411,6 +419,11 @@ NS_ASSUME_NONNULL_BEGIN }]; } +- (BOOL)hasPendingJobsWithTransaction:(SDSAnyReadTransaction *)transaction +{ + return [self.processingQueue hasPendingJobsWithTransaction:transaction]; +} + @end NS_ASSUME_NONNULL_END diff --git a/SignalServiceKit/src/Messages/OWSMessageReceiver.h b/SignalServiceKit/src/Messages/OWSMessageReceiver.h index fa28d8825f..8a9bbb7274 100644 --- a/SignalServiceKit/src/Messages/OWSMessageReceiver.h +++ b/SignalServiceKit/src/Messages/OWSMessageReceiver.h @@ -1,5 +1,5 @@ // -// Copyright (c) 2019 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // #import "BaseModel.h" @@ -11,6 +11,8 @@ NS_ASSUME_NONNULL_BEGIN @class SDSAnyWriteTransaction; @class SSKProtoEnvelope; +extern NSNotificationName const kNSNotificationNameMessageDecryptionDidFlushQueue; + @interface OWSMessageDecryptJob : BaseModel @property (nonatomic, readonly) NSDate *createdAt; @@ -71,6 +73,8 @@ NS_SWIFT_NAME(init(grdbId:uniqueId:createdAt:envelopeData:)); - (void)handleReceivedEnvelopeData:(NSData *)envelopeData; +- (BOOL)hasPendingJobsWithTransaction:(SDSAnyReadTransaction *)transaction; + @end NS_ASSUME_NONNULL_END diff --git a/SignalServiceKit/src/Messages/OWSMessageReceiver.m b/SignalServiceKit/src/Messages/OWSMessageReceiver.m index 333915b8b0..de0b936d15 100644 --- a/SignalServiceKit/src/Messages/OWSMessageReceiver.m +++ b/SignalServiceKit/src/Messages/OWSMessageReceiver.m @@ -6,6 +6,7 @@ #import "AppContext.h" #import "AppReadiness.h" #import "NSArray+OWS.h" +#import "NSNotificationCenter+OWS.h" #import "NotificationsProtocol.h" #import "OWSBackgroundTask.h" #import "OWSBatchMessageProcessor.h" @@ -21,6 +22,9 @@ NS_ASSUME_NONNULL_BEGIN +NSNotificationName const kNSNotificationNameMessageDecryptionDidFlushQueue + = @"kNSNotificationNameMessageDecryptionDidFlushQueue"; + @implementation OWSMessageDecryptJob + (NSString *)collection @@ -282,9 +286,7 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin _isDrainingQueue = NO; [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - if (CurrentAppContext().isMainApp) { - [self drainQueue]; - } + [self drainQueue]; }]; [[NSNotificationCenter defaultCenter] addObserver:self @@ -330,9 +332,7 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin OWSAssertIsOnMainThread(); [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - if (CurrentAppContext().isMainApp) { - [self drainQueue]; - } + [self drainQueue]; }]; } @@ -357,8 +357,7 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin { OWSAssertDebug(AppReadiness.isAppReady || CurrentAppContext().isRunningTests); - // Don't decrypt messages in app extensions. - if (!CurrentAppContext().isMainApp) { + if (!CurrentAppContext().shouldProcessIncomingMessages) { return; } if (!self.tsAccountManager.isRegisteredAndReady) { @@ -375,6 +374,11 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin }); } +- (BOOL)hasPendingJobsWithTransaction:(SDSAnyReadTransaction *)transaction +{ + return nil != [self.finder nextJobWithTransaction:transaction]; +} + - (void)drainQueueWorkStep { AssertOnDispatchQueue(self.serialQueue); @@ -388,6 +392,12 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin if (!job) { self.isDrainingQueue = NO; OWSLogVerbose(@"Queue is drained."); + + [[NSNotificationCenter defaultCenter] + postNotificationNameAsync:kNSNotificationNameMessageDecryptionDidFlushQueue + object:nil + userInfo:nil]; + return; } @@ -497,9 +507,7 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin _yapProcessingQueue = yapProcessingQueue; [AppReadiness runNowOrWhenAppDidBecomeReady:^{ - if (CurrentAppContext().isMainApp) { - [self.yapProcessingQueue drainQueue]; - } + [self.yapProcessingQueue drainQueue]; }]; return self; @@ -563,6 +571,16 @@ NSString *const OWSMessageDecryptJobFinderExtensionGroup = @"OWSMessageProcessin } } +- (BOOL)hasPendingJobsWithTransaction:(SDSAnyReadTransaction *)transaction +{ + if (StorageCoordinator.dataStoreForUI == DataStoreYdb) { + OWSFailDebug(@"GRDB-for-all should ship before we use this method in production."); + return [self.yapProcessingQueue hasPendingJobsWithTransaction:transaction]; + } else { + return [self.messageDecryptJobQueue hasPendingJobsObjcWithTransaction:transaction]; + } +} + @end NS_ASSUME_NONNULL_END diff --git a/SignalServiceKit/src/Network/MessageSenderJobQueue.swift b/SignalServiceKit/src/Network/MessageSenderJobQueue.swift index b17d54d5db..c7e9dada9f 100644 --- a/SignalServiceKit/src/Network/MessageSenderJobQueue.swift +++ b/SignalServiceKit/src/Network/MessageSenderJobQueue.swift @@ -93,6 +93,10 @@ public class MessageSenderJobQueue: NSObject, JobQueue { } } + public func didFlushQueue(transaction: SDSAnyWriteTransaction) { + // Do nothing. + } + public func buildOperation(jobRecord: SSKMessageSenderJobRecord, transaction: SDSAnyReadTransaction) throws -> MessageSenderOperation { let message: TSOutgoingMessage if let invisibleMessage = jobRecord.invisibleMessage { diff --git a/SignalServiceKit/src/Network/Receiving/MessageDecryptJobQueue.swift b/SignalServiceKit/src/Network/Receiving/MessageDecryptJobQueue.swift index e5ab15a3c7..65b8d7284b 100644 --- a/SignalServiceKit/src/Network/Receiving/MessageDecryptJobQueue.swift +++ b/SignalServiceKit/src/Network/Receiving/MessageDecryptJobQueue.swift @@ -51,8 +51,7 @@ public class SSKMessageDecryptJobQueue: NSObject, JobQueue { @objc public func setup() { - // Don't decrypt messages in app extensions. - if !CurrentAppContext().isMainApp { + guard CurrentAppContext().shouldProcessIncomingMessages else { return } @@ -69,7 +68,11 @@ public class SSKMessageDecryptJobQueue: NSObject, JobQueue { public var isSetup = AtomicBool(false) public func didMarkAsReady(oldJobRecord: SSKMessageDecryptJobRecord, transaction: SDSAnyWriteTransaction) { + // Do nothing. + } + public func didFlushQueue(transaction: SDSAnyWriteTransaction) { + NotificationCenter.default.postNotificationNameAsync(.messageDecryptionDidFlushQueue, object: nil, userInfo: nil) } public func buildOperation(jobRecord: SSKMessageDecryptJobRecord, transaction: SDSAnyReadTransaction) throws -> SSKMessageDecryptOperation { @@ -87,6 +90,11 @@ public class SSKMessageDecryptJobQueue: NSObject, JobQueue { public func operationQueue(jobRecord: SSKMessageDecryptJobRecord) -> OperationQueue { return defaultQueue } + + @objc + public func hasPendingJobsObjc(transaction: SDSAnyReadTransaction) -> Bool { + return hasPendingJobs(transaction: transaction) + } } enum SSKMessageDecryptOperationError: Error { diff --git a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.h b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.h index bc7d102e4b..811ebd2703 100644 --- a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.h +++ b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.h @@ -23,6 +23,7 @@ typedef void (^TSSocketMessageFailure)(NSInteger statusCode, NSData *_Nullable r @interface OWSWebSocket : NSObject @property (nonatomic, readonly) OWSWebSocketState state; +@property (nonatomic, readonly) BOOL hasEmptiedInitialQueue; - (instancetype)init NS_DESIGNATED_INITIALIZER; diff --git a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.m b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.m index 5698aeba9d..b432f724e0 100644 --- a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.m +++ b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.m @@ -167,6 +167,8 @@ NSNotificationName const NSNotificationWebSocketStateDidChange = @"NSNotificatio // // We only ever access this state on the main thread. @property (nonatomic) OWSWebSocketState state; +@property (nonatomic) BOOL hasEmptiedInitialQueue; +@property (nonatomic) BOOL willEmptyInitialQueue; #pragma mark - @@ -215,6 +217,8 @@ NSNotificationName const NSNotificationWebSocketStateDidChange = @"NSNotificatio OWSAssertIsOnMainThread(); _state = OWSWebSocketStateClosed; + _hasEmptiedInitialQueue = NO; + _willEmptyInitialQueue = NO; _socketMessageMap = [NSMutableDictionary new]; return self; @@ -366,6 +370,11 @@ NSNotificationName const NSNotificationWebSocketStateDidChange = @"NSNotificatio { OWSAssertIsOnMainThread(); + if (state != OWSWebSocketStateOpen) { + self.hasEmptiedInitialQueue = NO; + self.willEmptyInitialQueue = NO; + } + // If this state update is redundant, verify that // class state and socket state are aligned. // @@ -815,6 +824,31 @@ NSNotificationName const NSNotificationWebSocketStateDidChange = @"NSNotificatio // Queue is drained. [self sendWebSocketMessageAcknowledgement:message]; + + if (!self.hasEmptiedInitialQueue) { + + self.willEmptyInitialQueue = YES; + + // We need to flush the serial queue to ensure that + // all received messages are enqueued by the message + // receiver before we: a) mark the queue as empty. + // b) notify. + // + // The socket might close and re-open while we're + // flushing the queue. We use willEmptyInitialQueue + // to detect this case. + dispatch_async(self.serialQueue, ^{ + dispatch_async(dispatch_get_main_queue(), ^{ + if (!self.willEmptyInitialQueue) { + return; + } + self.willEmptyInitialQueue = NO; + self.hasEmptiedInitialQueue = YES; + + [self notifyStatusChange]; + }); + }); + } } else { OWSLogWarn(@"Unsupported WebSocket Request"); diff --git a/SignalServiceKit/src/Network/WebSockets/TSSocketManager.h b/SignalServiceKit/src/Network/WebSockets/TSSocketManager.h index 28e17d17f2..466525a46c 100644 --- a/SignalServiceKit/src/Network/WebSockets/TSSocketManager.h +++ b/SignalServiceKit/src/Network/WebSockets/TSSocketManager.h @@ -1,5 +1,5 @@ // -// Copyright (c) 2018 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // #import "OWSWebSocket.h" @@ -21,7 +21,8 @@ NS_ASSUME_NONNULL_BEGIN // connectivity issues. We want to show the "best" or "highest" // socket state of the sockets. e.g. the UI should reflect // "open" if any of the sockets is open. -- (OWSWebSocketState)highestSocketState; +- (OWSWebSocketState)socketState; +- (BOOL)hasEmptiedInitialQueue; // If the app is in the foreground, we'll try to open the socket unless it's already // open or connecting. diff --git a/SignalServiceKit/src/Network/WebSockets/TSSocketManager.m b/SignalServiceKit/src/Network/WebSockets/TSSocketManager.m index cd255a3cdf..f73fd097c8 100644 --- a/SignalServiceKit/src/Network/WebSockets/TSSocketManager.m +++ b/SignalServiceKit/src/Network/WebSockets/TSSocketManager.m @@ -1,5 +1,5 @@ // -// Copyright (c) 2018 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // #import "TSSocketManager.h" @@ -68,11 +68,16 @@ NS_ASSUME_NONNULL_BEGIN [self.websocket cycleSocket]; } -- (OWSWebSocketState)highestSocketState +- (OWSWebSocketState)socketState { return self.websocket.state; } +- (BOOL)hasEmptiedInitialQueue +{ + return self.websocket.hasEmptiedInitialQueue; +} + @end NS_ASSUME_NONNULL_END diff --git a/SignalServiceKit/src/SSKEnvironment.h b/SignalServiceKit/src/SSKEnvironment.h index 896557d538..7d6e32b799 100644 --- a/SignalServiceKit/src/SSKEnvironment.h +++ b/SignalServiceKit/src/SSKEnvironment.h @@ -7,6 +7,8 @@ NS_ASSUME_NONNULL_BEGIN @class AccountServiceClient; @class ContactsUpdater; @class GroupsV2MessageProcessor; +@class MessageFetcherJob; +@class MessageProcessing; @class MessageSenderJobQueue; @class OWS2FAManager; @class OWSAttachmentDownloads; @@ -90,7 +92,9 @@ NS_ASSUME_NONNULL_BEGIN storageServiceManager:(id)storageServiceManager storageCoordinator:(StorageCoordinator *)storageCoordinator sskPreferences:(SSKPreferences *)sskPreferences - groupsV2:(id)groupsV2 NS_DESIGNATED_INITIALIZER; + groupsV2:(id)groupsV2 + messageProcessing:(MessageProcessing *)messageProcessing + messageFetcherJob:(MessageFetcherJob *)messageFetcherJob NS_DESIGNATED_INITIALIZER; @property (nonatomic, readonly, class) SSKEnvironment *shared; @@ -142,6 +146,8 @@ NS_ASSUME_NONNULL_BEGIN @property (nonatomic, readonly) SDSDatabaseStorage *databaseStorage; @property (nonatomic, readonly) StorageCoordinator *storageCoordinator; @property (nonatomic, readonly) SSKPreferences *sskPreferences; +@property (nonatomic, readonly) MessageProcessing *messageProcessing; +@property (nonatomic, readonly) MessageFetcherJob *messageFetcherJob; @property (nonatomic, readonly, nullable) OWSPrimaryStorage *primaryStorage; diff --git a/SignalServiceKit/src/SSKEnvironment.m b/SignalServiceKit/src/SSKEnvironment.m index e05c64fa42..3065b27818 100644 --- a/SignalServiceKit/src/SSKEnvironment.m +++ b/SignalServiceKit/src/SSKEnvironment.m @@ -46,6 +46,8 @@ static SSKEnvironment *sharedSSKEnvironment; @property (nonatomic) StorageCoordinator *storageCoordinator; @property (nonatomic) SSKPreferences *sskPreferences; @property (nonatomic) id groupsV2; +@property (nonatomic) MessageProcessing *messageProcessing; +@property (nonatomic) MessageFetcherJob *messageFetcherJob; @end @@ -96,6 +98,8 @@ static SSKEnvironment *sharedSSKEnvironment; storageCoordinator:(StorageCoordinator *)storageCoordinator sskPreferences:(SSKPreferences *)sskPreferences groupsV2:(id)groupsV2 + messageProcessing:(MessageProcessing *)messageProcessing + messageFetcherJob:(MessageFetcherJob *)messageFetcherJob { self = [super init]; if (!self) { @@ -140,6 +144,8 @@ static SSKEnvironment *sharedSSKEnvironment; OWSAssertDebug(storageCoordinator); OWSAssertDebug(sskPreferences); OWSAssertDebug(groupsV2); + OWSAssertDebug(messageProcessing); + OWSCAssertDebug(messageFetcherJob); _contactsManager = contactsManager; _linkPreviewManager = linkPreviewManager; @@ -180,6 +186,8 @@ static SSKEnvironment *sharedSSKEnvironment; _storageCoordinator = storageCoordinator; _sskPreferences = sskPreferences; _groupsV2 = groupsV2; + _messageProcessing = messageProcessing; + _messageFetcherJob = messageFetcherJob; return self; } diff --git a/SignalServiceKit/src/TestUtils/MockSSKEnvironment.m b/SignalServiceKit/src/TestUtils/MockSSKEnvironment.m index 6af8828f9e..70b7c7e92d 100644 --- a/SignalServiceKit/src/TestUtils/MockSSKEnvironment.m +++ b/SignalServiceKit/src/TestUtils/MockSSKEnvironment.m @@ -101,6 +101,8 @@ NS_ASSUME_NONNULL_BEGIN OWSFakeStorageServiceManager *storageServiceManager = [OWSFakeStorageServiceManager new]; SSKPreferences *sskPreferences = [SSKPreferences new]; id groupsV2 = [[MockGroupsV2 alloc] init]; + MessageProcessing *messageProcessing = [MessageProcessing new]; + MessageFetcherJob *messageFetcherJob = [MessageFetcherJob new]; self = [super initWithContactsManager:contactsManager linkPreviewManager:linkPreviewManager @@ -140,7 +142,9 @@ NS_ASSUME_NONNULL_BEGIN storageServiceManager:storageServiceManager storageCoordinator:storageCoordinator sskPreferences:sskPreferences - groupsV2:groupsV2]; + groupsV2:groupsV2 + messageProcessing:messageProcessing + messageFetcherJob:messageFetcherJob]; if (!self) { return nil; diff --git a/SignalServiceKit/src/TestUtils/TestAppContext.m b/SignalServiceKit/src/TestUtils/TestAppContext.m index e0b8dbe500..afe99eefdc 100644 --- a/SignalServiceKit/src/TestUtils/TestAppContext.m +++ b/SignalServiceKit/src/TestUtils/TestAppContext.m @@ -1,5 +1,5 @@ // -// Copyright (c) 2019 Open Whisper Systems. All rights reserved. +// Copyright (c) 2020 Open Whisper Systems. All rights reserved. // #import "TestAppContext.h" @@ -184,6 +184,11 @@ NS_ASSUME_NONNULL_BEGIN return NO; } +- (BOOL)shouldProcessIncomingMessages +{ + return NO; +} + @end #endif diff --git a/SignalServiceKit/src/Util/AppContext.h b/SignalServiceKit/src/Util/AppContext.h index f3afe3307b..301b26ae5b 100755 --- a/SignalServiceKit/src/Util/AppContext.h +++ b/SignalServiceKit/src/Util/AppContext.h @@ -125,6 +125,8 @@ NSString *NSStringForUIApplicationState(UIApplicationState value); - (BOOL)canPresentNotifications; +@property (nonatomic, readonly) BOOL shouldProcessIncomingMessages; + @end id CurrentAppContext(void); diff --git a/SignalServiceKit/src/Util/JobQueue.swift b/SignalServiceKit/src/Util/JobQueue.swift index 1b58f7069b..9772945965 100644 --- a/SignalServiceKit/src/Util/JobQueue.swift +++ b/SignalServiceKit/src/Util/JobQueue.swift @@ -84,6 +84,7 @@ public protocol JobQueue: DurableOperationDelegate { var isSetup: AtomicBool { get set } func setup() func didMarkAsReady(oldJobRecord: JobRecordType, transaction: SDSAnyWriteTransaction) + func didFlushQueue(transaction: SDSAnyWriteTransaction) func operationQueue(jobRecord: JobRecordType) -> OperationQueue func buildOperation(jobRecord: JobRecordType, transaction: SDSAnyReadTransaction) throws -> DurableOperationType @@ -125,6 +126,10 @@ public extension JobQueue { } } + func hasPendingJobs(transaction: SDSAnyReadTransaction) -> Bool { + return nil != finder.getNextReady(label: self.jobRecordLabel, transaction: transaction) + } + func startWorkWhenAppIsReady() { guard !CurrentAppContext().isRunningTests else { DispatchQueue.global().async { @@ -162,6 +167,7 @@ public extension JobQueue { self.databaseStorage.write { transaction in guard let nextJob: JobRecordType = self.finder.getNextReady(label: self.jobRecordLabel, transaction: transaction) else { Logger.verbose("nothing left to enqueue") + self.didFlushQueue(transaction: transaction) return } @@ -292,6 +298,8 @@ public extension JobQueue { func durableOperationDidSucceed(_ operation: DurableOperationType, transaction: SDSAnyWriteTransaction) { runningOperations.remove(operation) operation.jobRecord.anyRemove(transaction: transaction) + + notifyFlushQueueIfPossible(transaction: transaction) } func durableOperation(_ operation: DurableOperationType, didReportError: Error, transaction: SDSAnyWriteTransaction) { @@ -306,6 +314,15 @@ public extension JobQueue { func durableOperation(_ operation: DurableOperationType, didFailWithError error: Error, transaction: SDSAnyWriteTransaction) { runningOperations.remove(operation) operation.jobRecord.saveAsPermanentlyFailed(transaction: transaction) + + notifyFlushQueueIfPossible(transaction: transaction) + } + + func notifyFlushQueueIfPossible(transaction: SDSAnyWriteTransaction) { + guard nil == finder.getNextReady(label: jobRecordLabel, transaction: transaction) else { + return + } + self.didFlushQueue(transaction: transaction) } } diff --git a/SignalShareExtension/utils/ShareAppExtensionContext.m b/SignalShareExtension/utils/ShareAppExtensionContext.m index c0713ed8f3..614ee5780d 100644 --- a/SignalShareExtension/utils/ShareAppExtensionContext.m +++ b/SignalShareExtension/utils/ShareAppExtensionContext.m @@ -308,6 +308,11 @@ NS_ASSUME_NONNULL_BEGIN return NO; } +- (BOOL)shouldProcessIncomingMessages +{ + return NO; +} + @end NS_ASSUME_NONNULL_END