Merge pull request #408 from oxen-io/variable-rate-polling
Variable Rate Closed Group Polling
This commit is contained in:
commit
950d075c9b
|
@ -142,7 +142,7 @@ final class HomeVC : BaseVC, UITableViewDataSource, UITableViewDelegate, NewConv
|
||||||
if OWSIdentityManager.shared().identityKeyPair() != nil {
|
if OWSIdentityManager.shared().identityKeyPair() != nil {
|
||||||
let appDelegate = UIApplication.shared.delegate as! AppDelegate
|
let appDelegate = UIApplication.shared.delegate as! AppDelegate
|
||||||
appDelegate.startPollerIfNeeded()
|
appDelegate.startPollerIfNeeded()
|
||||||
appDelegate.startClosedGroupPollerIfNeeded()
|
appDelegate.startClosedGroupPoller()
|
||||||
appDelegate.startOpenGroupPollersIfNeeded()
|
appDelegate.startOpenGroupPollersIfNeeded()
|
||||||
// Do this only if we created a new Session ID, or if we already received the initial configuration message
|
// Do this only if we created a new Session ID, or if we already received the initial configuration message
|
||||||
if UserDefaults.standard[.hasSyncedInitialConfiguration] {
|
if UserDefaults.standard[.hasSyncedInitialConfiguration] {
|
||||||
|
|
|
@ -10,8 +10,6 @@ extern NSString *const AppDelegateStoryboardMain;
|
||||||
|
|
||||||
- (void)startPollerIfNeeded;
|
- (void)startPollerIfNeeded;
|
||||||
- (void)stopPoller;
|
- (void)stopPoller;
|
||||||
- (void)startClosedGroupPollerIfNeeded;
|
|
||||||
- (void)stopClosedGroupPoller;
|
|
||||||
- (void)startOpenGroupPollersIfNeeded;
|
- (void)startOpenGroupPollersIfNeeded;
|
||||||
- (void)stopOpenGroupPollers;
|
- (void)stopOpenGroupPollers;
|
||||||
|
|
||||||
|
|
|
@ -48,7 +48,6 @@ static NSTimeInterval launchStartedAt;
|
||||||
@property (nonatomic) BOOL areVersionMigrationsComplete;
|
@property (nonatomic) BOOL areVersionMigrationsComplete;
|
||||||
@property (nonatomic) BOOL didAppLaunchFail;
|
@property (nonatomic) BOOL didAppLaunchFail;
|
||||||
@property (nonatomic) LKPoller *poller;
|
@property (nonatomic) LKPoller *poller;
|
||||||
@property (nonatomic) LKClosedGroupPoller *closedGroupPoller;
|
|
||||||
|
|
||||||
@end
|
@end
|
||||||
|
|
||||||
|
@ -414,7 +413,7 @@ static NSTimeInterval launchStartedAt;
|
||||||
[[SNSnodeAPI getSnodePool] retainUntilComplete];
|
[[SNSnodeAPI getSnodePool] retainUntilComplete];
|
||||||
|
|
||||||
[self startPollerIfNeeded];
|
[self startPollerIfNeeded];
|
||||||
[self startClosedGroupPollerIfNeeded];
|
[self startClosedGroupPoller];
|
||||||
[self startOpenGroupPollersIfNeeded];
|
[self startOpenGroupPollersIfNeeded];
|
||||||
|
|
||||||
if (![UIApplication sharedApplication].isRegisteredForRemoteNotifications) {
|
if (![UIApplication sharedApplication].isRegisteredForRemoteNotifications) {
|
||||||
|
@ -564,7 +563,7 @@ static NSTimeInterval launchStartedAt;
|
||||||
[self.readReceiptManager setAreReadReceiptsEnabled:YES];
|
[self.readReceiptManager setAreReadReceiptsEnabled:YES];
|
||||||
|
|
||||||
[self startPollerIfNeeded];
|
[self startPollerIfNeeded];
|
||||||
[self startClosedGroupPollerIfNeeded];
|
[self startClosedGroupPoller];
|
||||||
[self startOpenGroupPollersIfNeeded];
|
[self startOpenGroupPollersIfNeeded];
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -727,19 +726,6 @@ static NSTimeInterval launchStartedAt;
|
||||||
|
|
||||||
- (void)stopPoller { [self.poller stop]; }
|
- (void)stopPoller { [self.poller stop]; }
|
||||||
|
|
||||||
- (void)startClosedGroupPollerIfNeeded
|
|
||||||
{
|
|
||||||
if (self.closedGroupPoller == nil) {
|
|
||||||
NSString *userPublicKey = [SNGeneralUtilities getUserPublicKey];
|
|
||||||
if (userPublicKey != nil) {
|
|
||||||
self.closedGroupPoller = [[LKClosedGroupPoller alloc] init];
|
|
||||||
}
|
|
||||||
}
|
|
||||||
[self.closedGroupPoller startIfNeeded];
|
|
||||||
}
|
|
||||||
|
|
||||||
- (void)stopClosedGroupPoller { [self.closedGroupPoller stop]; }
|
|
||||||
|
|
||||||
- (void)startOpenGroupPollersIfNeeded
|
- (void)startOpenGroupPollersIfNeeded
|
||||||
{
|
{
|
||||||
[SNOpenGroupManager.shared startPolling];
|
[SNOpenGroupManager.shared startPolling];
|
||||||
|
|
|
@ -31,4 +31,13 @@ extension AppDelegate {
|
||||||
}
|
}
|
||||||
return promise
|
return promise
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@objc func startClosedGroupPoller() {
|
||||||
|
guard OWSIdentityManager.shared().identityKeyPair() != nil else { return }
|
||||||
|
ClosedGroupPoller.shared.start()
|
||||||
|
}
|
||||||
|
|
||||||
|
@objc func stopClosedGroupPoller() {
|
||||||
|
ClosedGroupPoller.shared.stop()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -408,6 +408,8 @@ extension MessageReceiver {
|
||||||
Storage.shared.addClosedGroupEncryptionKeyPair(encryptionKeyPair, for: groupPublicKey, using: transaction)
|
Storage.shared.addClosedGroupEncryptionKeyPair(encryptionKeyPair, for: groupPublicKey, using: transaction)
|
||||||
// Store the formation timestamp
|
// Store the formation timestamp
|
||||||
Storage.shared.setClosedGroupFormationTimestamp(to: messageSentTimestamp, for: groupPublicKey, using: transaction)
|
Storage.shared.setClosedGroupFormationTimestamp(to: messageSentTimestamp, for: groupPublicKey, using: transaction)
|
||||||
|
// Start polling
|
||||||
|
ClosedGroupPoller.shared.startPolling(for: groupPublicKey)
|
||||||
// Notify the PN server
|
// Notify the PN server
|
||||||
let _ = PushNotificationAPI.performOperation(.subscribe, for: groupPublicKey, publicKey: getUserHexEncodedPublicKey())
|
let _ = PushNotificationAPI.performOperation(.subscribe, for: groupPublicKey, publicKey: getUserHexEncodedPublicKey())
|
||||||
}
|
}
|
||||||
|
@ -539,6 +541,7 @@ extension MessageReceiver {
|
||||||
if wasCurrentUserRemoved {
|
if wasCurrentUserRemoved {
|
||||||
Storage.shared.removeClosedGroupPublicKey(groupPublicKey, using: transaction)
|
Storage.shared.removeClosedGroupPublicKey(groupPublicKey, using: transaction)
|
||||||
Storage.shared.removeAllClosedGroupEncryptionKeyPairs(for: groupPublicKey, using: transaction)
|
Storage.shared.removeAllClosedGroupEncryptionKeyPairs(for: groupPublicKey, using: transaction)
|
||||||
|
ClosedGroupPoller.shared.stopPolling(for: groupPublicKey)
|
||||||
let _ = PushNotificationAPI.performOperation(.unsubscribe, for: groupPublicKey, publicKey: userPublicKey)
|
let _ = PushNotificationAPI.performOperation(.unsubscribe, for: groupPublicKey, publicKey: userPublicKey)
|
||||||
}
|
}
|
||||||
let storage = SNMessagingKitConfiguration.shared.storage
|
let storage = SNMessagingKitConfiguration.shared.storage
|
||||||
|
|
|
@ -43,6 +43,8 @@ extension MessageSender {
|
||||||
// Notify the user
|
// Notify the user
|
||||||
let infoMessage = TSInfoMessage(timestamp: NSDate.ows_millisecondTimeStamp(), in: thread, messageType: .groupCreated)
|
let infoMessage = TSInfoMessage(timestamp: NSDate.ows_millisecondTimeStamp(), in: thread, messageType: .groupCreated)
|
||||||
infoMessage.save(with: transaction)
|
infoMessage.save(with: transaction)
|
||||||
|
// Start polling
|
||||||
|
ClosedGroupPoller.shared.startPolling(for: groupPublicKey)
|
||||||
// Return
|
// Return
|
||||||
return when(fulfilled: promises).map2 { thread }
|
return when(fulfilled: promises).map2 { thread }
|
||||||
}
|
}
|
||||||
|
@ -272,6 +274,7 @@ extension MessageSender {
|
||||||
// Remove the group from the database and unsubscribe from PNs
|
// Remove the group from the database and unsubscribe from PNs
|
||||||
Storage.shared.removeAllClosedGroupEncryptionKeyPairs(for: groupPublicKey, using: transaction)
|
Storage.shared.removeAllClosedGroupEncryptionKeyPairs(for: groupPublicKey, using: transaction)
|
||||||
Storage.shared.removeClosedGroupPublicKey(groupPublicKey, using: transaction)
|
Storage.shared.removeClosedGroupPublicKey(groupPublicKey, using: transaction)
|
||||||
|
ClosedGroupPoller.shared.stopPolling(for: groupPublicKey)
|
||||||
let _ = PushNotificationAPI.performOperation(.unsubscribe, for: groupPublicKey, publicKey: userPublicKey)
|
let _ = PushNotificationAPI.performOperation(.unsubscribe, for: groupPublicKey, publicKey: userPublicKey)
|
||||||
}
|
}
|
||||||
}.map { _ in }
|
}.map { _ in }
|
||||||
|
|
|
@ -3,11 +3,12 @@ import PromiseKit
|
||||||
|
|
||||||
@objc(LKClosedGroupPoller)
|
@objc(LKClosedGroupPoller)
|
||||||
public final class ClosedGroupPoller : NSObject {
|
public final class ClosedGroupPoller : NSObject {
|
||||||
private var isPolling = false
|
private var isPolling: [String:Bool] = [:]
|
||||||
private var timer: Timer?
|
private var timers: [String:Timer] = [:]
|
||||||
|
|
||||||
// MARK: Settings
|
// MARK: Settings
|
||||||
private static let pollInterval: TimeInterval = 3
|
private static let minPollInterval: Double = 4
|
||||||
|
private static let maxPollInterval: Double = 2 * 60
|
||||||
|
|
||||||
// MARK: Error
|
// MARK: Error
|
||||||
private enum Error : LocalizedError {
|
private enum Error : LocalizedError {
|
||||||
|
@ -22,48 +23,97 @@ public final class ClosedGroupPoller : NSObject {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// MARK: Initialization
|
||||||
|
public static let shared = ClosedGroupPoller()
|
||||||
|
|
||||||
|
private override init() { }
|
||||||
|
|
||||||
// MARK: Public API
|
// MARK: Public API
|
||||||
@objc public func startIfNeeded() {
|
@objc public func start() {
|
||||||
#if DEBUG
|
#if DEBUG
|
||||||
assert(Thread.current.isMainThread) // Timers don't do well on background queues
|
assert(Thread.current.isMainThread) // Timers don't do well on background queues
|
||||||
#endif
|
#endif
|
||||||
guard !isPolling else { return }
|
let storage = SNMessagingKitConfiguration.shared.storage
|
||||||
isPolling = true
|
let allGroupPublicKeys = storage.getUserClosedGroupPublicKeys()
|
||||||
timer = Timer.scheduledTimer(withTimeInterval: ClosedGroupPoller.pollInterval, repeats: true) { [weak self] _ in
|
allGroupPublicKeys.forEach { startPolling(for: $0) }
|
||||||
let _ = self?.poll()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public func pollOnce() -> [Promise<Void>] {
|
public func startPolling(for groupPublicKey: String) {
|
||||||
guard !isPolling else { return [] }
|
guard !isPolling(for: groupPublicKey) else { return }
|
||||||
isPolling = true
|
setUpPolling(for: groupPublicKey)
|
||||||
return poll()
|
isPolling[groupPublicKey] = true
|
||||||
}
|
}
|
||||||
|
|
||||||
@objc public func stop() {
|
@objc public func stop() {
|
||||||
isPolling = false
|
let storage = SNMessagingKitConfiguration.shared.storage
|
||||||
timer?.invalidate()
|
let allGroupPublicKeys = storage.getUserClosedGroupPublicKeys()
|
||||||
|
allGroupPublicKeys.forEach { stopPolling(for: $0) }
|
||||||
|
}
|
||||||
|
|
||||||
|
public func stopPolling(for groupPublicKey: String) {
|
||||||
|
timers[groupPublicKey]?.invalidate()
|
||||||
|
isPolling[groupPublicKey] = false
|
||||||
}
|
}
|
||||||
|
|
||||||
// MARK: Private API
|
// MARK: Private API
|
||||||
private func poll() -> [Promise<Void>] {
|
private func setUpPolling(for groupPublicKey: String) {
|
||||||
guard isPolling else { return [] }
|
poll(groupPublicKey).done2 { [weak self] _ in
|
||||||
let publicKeys = Storage.shared.getUserClosedGroupPublicKeys()
|
DispatchQueue.main.async { // Timers don't do well on background queues
|
||||||
return publicKeys.map { publicKey in
|
self?.pollRecursively(groupPublicKey)
|
||||||
let promise = SnodeAPI.getSwarm(for: publicKey).then2 { [weak self] swarm -> Promise<[JSON]> in
|
}
|
||||||
|
}.catch2 { [weak self] error in
|
||||||
|
// The error is logged in poll(_:)
|
||||||
|
DispatchQueue.main.async { // Timers don't do well on background queues
|
||||||
|
self?.pollRecursively(groupPublicKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func pollRecursively(_ groupPublicKey: String) {
|
||||||
|
let groupID = LKGroupUtilities.getEncodedClosedGroupIDAsData(groupPublicKey)
|
||||||
|
guard isPolling(for: groupPublicKey),
|
||||||
|
let thread = TSGroupThread.fetch(uniqueId: TSGroupThread.threadId(fromGroupId: groupID)) else { return }
|
||||||
|
// Get the received date of the last message in the thread. If we don't have any messages yet, pick some
|
||||||
|
// reasonable fake time interval to use instead.
|
||||||
|
let lastMessageDate =
|
||||||
|
(thread.numberOfInteractions() > 0) ? thread.lastInteraction.receivedAtDate() : Date().addingTimeInterval(-5 * 60)
|
||||||
|
let timeSinceLastMessage = Date().timeIntervalSince(lastMessageDate)
|
||||||
|
let minPollInterval = ClosedGroupPoller.minPollInterval
|
||||||
|
let limit: Double = 12 * 60 * 60
|
||||||
|
let a = (ClosedGroupPoller.maxPollInterval - minPollInterval) / limit
|
||||||
|
let nextPollInterval = a * min(timeSinceLastMessage, limit) + minPollInterval
|
||||||
|
SNLog("Next poll interval for closed group with public key: \(groupPublicKey) is \(nextPollInterval) s.")
|
||||||
|
timers[groupPublicKey] = Timer.scheduledTimer(withTimeInterval: nextPollInterval, repeats: false) { [weak self] timer in
|
||||||
|
timer.invalidate()
|
||||||
|
self?.poll(groupPublicKey).done2 { _ in
|
||||||
|
DispatchQueue.main.async { // Timers don't do well on background queues
|
||||||
|
self?.pollRecursively(groupPublicKey)
|
||||||
|
}
|
||||||
|
}.catch2 { error in
|
||||||
|
// The error is logged in poll(_:)
|
||||||
|
DispatchQueue.main.async { // Timers don't do well on background queues
|
||||||
|
self?.pollRecursively(groupPublicKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func poll(_ groupPublicKey: String) -> Promise<Void> {
|
||||||
|
guard isPolling(for: groupPublicKey) else { return Promise.value(()) }
|
||||||
|
let promise = SnodeAPI.getSwarm(for: groupPublicKey).then2 { [weak self] swarm -> Promise<[JSON]> in
|
||||||
// randomElement() uses the system's default random generator, which is cryptographically secure
|
// randomElement() uses the system's default random generator, which is cryptographically secure
|
||||||
guard let snode = swarm.randomElement() else { return Promise(error: Error.insufficientSnodes) }
|
guard let snode = swarm.randomElement() else { return Promise(error: Error.insufficientSnodes) }
|
||||||
guard let self = self, self.isPolling else { return Promise(error: Error.pollingCanceled) }
|
guard let self = self, self.isPolling(for: groupPublicKey) else { return Promise(error: Error.pollingCanceled) }
|
||||||
return SnodeAPI.getRawMessages(from: snode, associatedWith: publicKey).map2 {
|
return SnodeAPI.getRawMessages(from: snode, associatedWith: groupPublicKey).map2 {
|
||||||
SnodeAPI.parseRawMessagesResponse($0, from: snode, associatedWith: publicKey)
|
SnodeAPI.parseRawMessagesResponse($0, from: snode, associatedWith: groupPublicKey)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
promise.done2 { [weak self] messages in
|
promise.done2 { [weak self] rawMessages in
|
||||||
guard let self = self, self.isPolling else { return }
|
guard let self = self, self.isPolling(for: groupPublicKey) else { return }
|
||||||
if !messages.isEmpty {
|
if !rawMessages.isEmpty {
|
||||||
SNLog("Received \(messages.count) new message(s) in closed group with public key: \(publicKey).")
|
SNLog("Received \(rawMessages.count) new message(s) in closed group with public key: \(groupPublicKey).")
|
||||||
}
|
}
|
||||||
messages.forEach { json in
|
rawMessages.forEach { json in
|
||||||
guard let envelope = SNProtoEnvelope.from(json) else { return }
|
guard let envelope = SNProtoEnvelope.from(json) else { return }
|
||||||
do {
|
do {
|
||||||
let data = try envelope.serializedData()
|
let data = try envelope.serializedData()
|
||||||
|
@ -77,10 +127,13 @@ public final class ClosedGroupPoller : NSObject {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
promise.catch2 { error in
|
promise.catch2 { error in
|
||||||
SNLog("Polling failed for closed group with public key: \(publicKey) due to error: \(error).")
|
SNLog("Polling failed for closed group with public key: \(groupPublicKey) due to error: \(error).")
|
||||||
}
|
}
|
||||||
promise.retainUntilComplete()
|
|
||||||
return promise.map { _ in }
|
return promise.map { _ in }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// MARK: Convenience
|
||||||
|
private func isPolling(for groupPublicKey: String) -> Bool {
|
||||||
|
return isPolling[groupPublicKey] ?? false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -18,7 +18,7 @@ public final class OpenGroupPoller : NSObject {
|
||||||
}
|
}
|
||||||
|
|
||||||
// MARK: Settings
|
// MARK: Settings
|
||||||
private let pollForNewMessagesInterval: TimeInterval = 8
|
private let pollForNewMessagesInterval: TimeInterval = 20
|
||||||
private let pollForDeletedMessagesInterval: TimeInterval = 30
|
private let pollForDeletedMessagesInterval: TimeInterval = 30
|
||||||
private let pollForModeratorsInterval: TimeInterval = 10 * 60
|
private let pollForModeratorsInterval: TimeInterval = 10 * 60
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue