mirror of https://github.com/oxen-io/session-ios
Added long polling.
Mikunj
6 years ago
parent
d0bed4b129
commit
04bdaff3c6
@ -1 +1 @@
|
|||||||
Subproject commit bff3e9db302b7808bf99451c738138c326ad16e3
|
Subproject commit 0c0008c2306e7b5be87c2c943fade9bd5612ee17
|
||||||
@ -1,26 +0,0 @@
|
|||||||
import PromiseKit
|
|
||||||
|
|
||||||
@objc final class Poller : NSObject {
|
|
||||||
private var isStarted = false
|
|
||||||
private var currentJob: Promise<Void>?
|
|
||||||
|
|
||||||
// MARK: Configuration
|
|
||||||
private static let interval: TimeInterval = 5 * 60
|
|
||||||
|
|
||||||
// MARK: Initialization
|
|
||||||
@objc static let shared = Poller()
|
|
||||||
|
|
||||||
private override init() { }
|
|
||||||
|
|
||||||
// MARK: General
|
|
||||||
@objc func startIfNeeded() {
|
|
||||||
guard !isStarted else { return }
|
|
||||||
Timer.scheduledTimer(timeInterval: Poller.interval, target: self, selector: #selector(poll), userInfo: nil, repeats: true)
|
|
||||||
isStarted = true
|
|
||||||
}
|
|
||||||
|
|
||||||
@objc private func poll() {
|
|
||||||
guard currentJob == nil else { return }
|
|
||||||
currentJob = AppEnvironment.shared.messageFetcherJob.run().ensure { [weak self] in self?.currentJob = nil }
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -0,0 +1,117 @@
|
|||||||
|
import PromiseKit
|
||||||
|
|
||||||
|
private typealias Callback = () -> Void
|
||||||
|
|
||||||
|
public extension LokiAPI {
|
||||||
|
private static var isLongPolling = false
|
||||||
|
private static var shouldStopPolling = false
|
||||||
|
private static var usedSnodes = [Target]()
|
||||||
|
private static var cancels = [Callback]()
|
||||||
|
|
||||||
|
private static let hexEncodedPublicKey = OWSIdentityManager.shared().identityKeyPair()!.hexEncodedPublicKey
|
||||||
|
|
||||||
|
/// Start long polling.
|
||||||
|
/// This will send a notification if new messages were received
|
||||||
|
@objc public static func startLongPollingIfNecessary() {
|
||||||
|
guard !isLongPolling else { return }
|
||||||
|
isLongPolling = true
|
||||||
|
shouldStopPolling = false
|
||||||
|
|
||||||
|
Logger.info("[Loki] Started long polling")
|
||||||
|
|
||||||
|
longPoll()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Stop long polling
|
||||||
|
@objc public static func stopLongPolling() {
|
||||||
|
shouldStopPolling = true
|
||||||
|
isLongPolling = false
|
||||||
|
usedSnodes.removeAll()
|
||||||
|
cancelAllPromises()
|
||||||
|
|
||||||
|
Logger.info("[Loki] Stopped long polling")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The long polling loop
|
||||||
|
private static func longPoll() {
|
||||||
|
// This is here so we can stop the infinite loop
|
||||||
|
guard !shouldStopPolling else { return }
|
||||||
|
|
||||||
|
fetchSwarmIfNeeded(for: hexEncodedPublicKey).then { _ -> Guarantee<[Result<Void>]> in
|
||||||
|
var promises = [Promise<Void>]()
|
||||||
|
let connections = 3
|
||||||
|
for i in 0..<connections {
|
||||||
|
let (promise, cancel) = openConnection()
|
||||||
|
promises.append(promise)
|
||||||
|
cancels.append(cancel)
|
||||||
|
}
|
||||||
|
return when(resolved: promises)
|
||||||
|
}.done { _ in
|
||||||
|
// Since all promises are complete, we can clear the cancels
|
||||||
|
cancelAllPromises()
|
||||||
|
|
||||||
|
// Keep long polling until it is stopped
|
||||||
|
longPoll()
|
||||||
|
}.retainUntilComplete()
|
||||||
|
}
|
||||||
|
|
||||||
|
private static func cancelAllPromises() {
|
||||||
|
cancels.forEach { cancel in cancel() }
|
||||||
|
cancels.removeAll()
|
||||||
|
}
|
||||||
|
|
||||||
|
private static func getUnusedSnodes() -> [Target] {
|
||||||
|
let snodes = getCachedSnodes(for: hexEncodedPublicKey)
|
||||||
|
return snodes.filter { !usedSnodes.contains($0) }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Open a connection to an unused snode and get messages from it
|
||||||
|
private static func openConnection() -> (Promise<Void>, cancel: Callback) {
|
||||||
|
var isCancelled = false
|
||||||
|
|
||||||
|
let cancel = {
|
||||||
|
isCancelled = true
|
||||||
|
}
|
||||||
|
|
||||||
|
func connectToNextSnode() -> Promise<Void> {
|
||||||
|
guard let nextSnode = getUnusedSnodes().first else {
|
||||||
|
// We don't have anymore unused snodes
|
||||||
|
return Promise.value(())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Add the snode to the used array
|
||||||
|
usedSnodes.append(nextSnode)
|
||||||
|
|
||||||
|
func getMessagesInfinitely(from target: Target) -> Promise<Void> {
|
||||||
|
// The only way to exit the infinite loop is to throw an error 3 times or cancel
|
||||||
|
return getMessages(from: target).then { messages -> Promise<Void> in
|
||||||
|
// Send our messages as a notification
|
||||||
|
NotificationCenter.default.post(name: .receivedNewMessages, object: nil, userInfo: ["messages": messages])
|
||||||
|
|
||||||
|
// Check if we need to abort
|
||||||
|
guard !isCancelled else { throw PMKError.cancelled }
|
||||||
|
|
||||||
|
// Continue fetching if we haven't cancelled
|
||||||
|
return getMessagesInfinitely(from: target)
|
||||||
|
}.retryingIfNeeded(maxRetryCount: 3)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Keep getting messages for this snode
|
||||||
|
// If we errored out then connect to the next snode
|
||||||
|
return getMessagesInfinitely(from: nextSnode).recover { _ -> Promise<Void> in
|
||||||
|
// Connect to the next snode if we haven't cancelled
|
||||||
|
if (!isCancelled) {
|
||||||
|
// We also need to remove the cached snode so we don't contact it again
|
||||||
|
removeCachedSnode(nextSnode, for: hexEncodedPublicKey)
|
||||||
|
return connectToNextSnode()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cancelled, so just return successfully
|
||||||
|
return Promise.value(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Keep connecting to snodes
|
||||||
|
return (connectToNextSnode(), cancel)
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -1,4 +0,0 @@
|
|||||||
|
|
||||||
public extension Notification.Name {
|
|
||||||
public static let contactOnlineStatusChanged = Notification.Name("contactOnlineStatusChanged")
|
|
||||||
}
|
|
||||||
Loading…
Reference in New Issue