signalapp/Signal-iOS
signalapp/Signal-iOS
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 4 Wiki Activity

Add message processing class.

Browse Source 
* Modify message processing to allow observation of websocket queue being drained.
* Extend MessageProcessing to allow observation of REST message fetching and "all message fetching and processing".
This commit is contained in:
Matthew Chen 2019-12-20 20:12:25 -03:00
parent 5bc638227f
commit bcfba85356
32 changed files with 826 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Signal.xcodeproj/project.pbxproj
View File

@ -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 = "<group>"; };
452D1AF02081059C00A67F7F /* StringAdditionsTest.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = StringAdditionsTest.swift; sourceTree = "<group>"; };
452EC6DE205E9E30000E787C /* MediaGallery.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MediaGallery.swift; sourceTree = "<group>"; };
452ECA4C1E087E7200E2F016 /* MessageFetcherJob.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = MessageFetcherJob.swift; sourceTree = "<group>"; };
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 = "<group>"; };
4535186D1FC635DD00210559 /* Base */ = {isa = PBXFileReference; lastKnownFileType = file.storyboard; name = Base; path = Base.lproj/MainInterface.storyboard; sourceTree = "<group>"; };
@ -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 */,

15
Signal/src/AppDelegate.m
View File

@ -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 =

4
Signal/src/Jobs/BroadcastMediaMessageJob.swift
View File

@ -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"

4
Signal/src/Jobs/SessionResetJob.swift
View File

@ -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()

4
Signal/src/ViewControllers/AppSettings/AdvancedSettingsTableViewController.m
View File

@ -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;

2
Signal/src/ViewControllers/AppSettings/AppSettingsViewController.m
View File

@ -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;

25
Signal/src/ViewControllers/HomeView/ConversationListViewController.m
View File

@ -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

6
Signal/src/environment/AppEnvironment.swift
View File

@ -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()

6
Signal/src/environment/PushRegistrationManager.swift
View File

@ -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<Void>).retainUntilComplete()
self.messageFetcherJob.run().promise.retainUntilComplete()
}
}
}

5
Signal/src/util/MainAppContext.m
View File

@ -404,6 +404,11 @@ NSString *const ReportedApplicationStateDidChangeNotification = @"ReportedApplic
return YES;
}
- (BOOL)shouldProcessIncomingMessages
{
return YES;
}
@end
NS_ASSUME_NONNULL_END

4
SignalMessaging/contacts/IncomingContactSyncJobQueue.swift
View File

@ -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"

4
SignalMessaging/contacts/OWSGroupSyncProcessingJobQueue.swift
View File

@ -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"

6
SignalMessaging/environment/AppSetup.m
View File

@ -109,6 +109,8 @@ NS_ASSUME_NONNULL_BEGIN
OWSSounds *sounds = [OWSSounds new];
id<OWSProximityMonitoringManager> 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();

208
Signal/src/Jobs/MessageFetcherJob.swift → SignalServiceKit/src/Messages/MessageFetcherJob.swift
View File

@ -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<Void>
// 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<UUID>()
// 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<Void> {
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<Void>) {
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<Void> in
fetchMessagesViaRest()
.done {
resolver.fulfill(())
}.catch { error in
resolver.reject(error)
}.retainUntilComplete()
}
// MARK: -
private class func fetchMessagesViaRest() -> Promise<Void> {
Logger.debug("")
return fetchBatchViaRest().then { (envelopes: [SSKProtoEnvelope], more: Bool) -> Promise<Void> 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<Void>? = 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<Void>
let resolver: Resolver<Void>
override required init() {
let (promise, resolver) = Promise<Void>.pending()
self.promise = promise
self.resolver = resolver
}
public override func run() {
Logger.debug("")
MessageFetcherJob.fetchMessages(resolver: resolver)
promise.ensure {
self.reportSuccess()
}.retainUntilComplete()
}
}

415
SignalServiceKit/src/Messages/MessageProcessing.swift Normal file
View File

@ -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<Void>]()
private func decryptStepPromise() -> Promise<Void> {
let (promise, resolver) = Promise<Void>.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<Void>]()
private func processingStepPromise() -> Promise<Void> {
let (promise, resolver) = Promise<Void>.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<Void>]()
// 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<Void> {
let (promise, resolver) = Promise<Void>.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<Void>]()
// 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<Void> {
let (promise, resolver) = Promise<Void>.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<Void>]()
// This promise can be used by the Groups v2 logic
// to block until all messages are fetched and processed.
public func allMessageFetchingAndProcessingPromise() -> Promise<Void> {
let (promise, resolver) = Promise<Void>.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()
}
}
}

8
SignalServiceKit/src/Messages/OWSBatchMessageProcessor.h
View File

@ -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

35
SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m
View File

@ -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

6
SignalServiceKit/src/Messages/OWSMessageReceiver.h
View File

@ -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

40
SignalServiceKit/src/Messages/OWSMessageReceiver.m
View File

@ -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

4
SignalServiceKit/src/Network/MessageSenderJobQueue.swift
View File

@ -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 {

12
SignalServiceKit/src/Network/Receiving/MessageDecryptJobQueue.swift
View File

@ -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 {

1
SignalServiceKit/src/Network/WebSockets/OWSWebSocket.h
View File

@ -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;

34
SignalServiceKit/src/Network/WebSockets/OWSWebSocket.m
View File

@ -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");

5
SignalServiceKit/src/Network/WebSockets/TSSocketManager.h
View File

@ -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.

9
SignalServiceKit/src/Network/WebSockets/TSSocketManager.m
View File

@ -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

8
SignalServiceKit/src/SSKEnvironment.h
View File

@ -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<StorageServiceManagerProtocol>)storageServiceManager
storageCoordinator:(StorageCoordinator *)storageCoordinator
sskPreferences:(SSKPreferences *)sskPreferences
groupsV2:(id<GroupsV2>)groupsV2 NS_DESIGNATED_INITIALIZER;
groupsV2:(id<GroupsV2>)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;

8
SignalServiceKit/src/SSKEnvironment.m
View File

@ -46,6 +46,8 @@ static SSKEnvironment *sharedSSKEnvironment;
@property (nonatomic) StorageCoordinator *storageCoordinator;
@property (nonatomic) SSKPreferences *sskPreferences;
@property (nonatomic) id<GroupsV2> 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>)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;
}

6
SignalServiceKit/src/TestUtils/MockSSKEnvironment.m
View File

@ -101,6 +101,8 @@ NS_ASSUME_NONNULL_BEGIN
OWSFakeStorageServiceManager *storageServiceManager = [OWSFakeStorageServiceManager new];
SSKPreferences *sskPreferences = [SSKPreferences new];
id<GroupsV2> 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;

7
SignalServiceKit/src/TestUtils/TestAppContext.m
View File

@ -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

2
SignalServiceKit/src/Util/AppContext.h
View File

@ -125,6 +125,8 @@ NSString *NSStringForUIApplicationState(UIApplicationState value);
- (BOOL)canPresentNotifications;
@property (nonatomic, readonly) BOOL shouldProcessIncomingMessages;
@end
id<AppContext> CurrentAppContext(void);

17
SignalServiceKit/src/Util/JobQueue.swift
View File

@ -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)
}
}

5
SignalShareExtension/utils/ShareAppExtensionContext.m
View File

@ -308,6 +308,11 @@ NS_ASSUME_NONNULL_BEGIN
return NO;
}
- (BOOL)shouldProcessIncomingMessages
{
return NO;
}
@end
NS_ASSUME_NONNULL_END