Merge commit '772abc68aa9d5d1da519dc33455de89d3c1de78c' into long-polling

pull/26/head
Mikunj 6 years ago
commit fe034be749

@ -5,7 +5,7 @@ internal extension LokiAPI {
private static let receivedMessageHashValuesKey = "receivedMessageHashValuesKey"
private static let receivedMessageHashValuesCollection = "receivedMessageHashValuesCollection"
internal static func getLastMessageHashValue(for target: Target) -> String? {
internal static func getLastMessageHashValue(for target: LokiAPITarget) -> String? {
var result: String? = nil
// Uses a read/write connection because getting the last message hash value also removes expired messages as needed
storage.dbReadWriteConnection.readWrite { transaction in
@ -14,7 +14,7 @@ internal extension LokiAPI {
return result
}
internal static func setLastMessageHashValue(for target: Target, hashValue: String, expiresAt: UInt64) {
internal static func setLastMessageHashValue(for target: LokiAPITarget, hashValue: String, expiresAt: UInt64) {
storage.dbReadWriteConnection.readWrite { transaction in
storage.setLastMessageHash(forServiceNode: target.address, hash: hashValue, expiresAt: expiresAt, transaction: transaction)
}

@ -5,7 +5,7 @@ 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 usedSnodes = [LokiAPITarget]()
private static var cancels = [Callback]()
private static let hexEncodedPublicKey = OWSIdentityManager.shared().identityKeyPair()!.hexEncodedPublicKey
@ -60,7 +60,7 @@ public extension LokiAPI {
cancels.removeAll()
}
private static func getUnusedSnodes() -> [Target] {
private static func getUnusedSnodes() -> [LokiAPITarget] {
let snodes = getCachedSnodes(for: hexEncodedPublicKey)
return snodes.filter { !usedSnodes.contains($0) }
}
@ -82,7 +82,7 @@ public extension LokiAPI {
// Add the snode to the used array
usedSnodes.append(nextSnode)
func getMessagesInfinitely(from target: Target) -> Promise<Void> {
func getMessagesInfinitely(from target: LokiAPITarget) -> Promise<Void> {
// The only way to exit the infinite loop is to throw an error 3 times or cancel
return getRawMessages(from: target).then { rawMessages -> Promise<Void> in
// Check if we need to abort

@ -11,42 +11,40 @@ public extension LokiAPI {
private static let swarmCacheKey = "swarmCacheKey"
private static let swarmCacheCollection = "swarmCacheCollection"
fileprivate static var swarmCache: [String:[Target]] {
fileprivate static var swarmCache: [String:[LokiAPITarget]] {
get {
var result: [String:[Target]]? = nil
var result: [String:[LokiAPITarget]]? = nil
storage.dbReadConnection.read { transaction in
let intermediate = transaction.object(forKey: swarmCacheKey, inCollection: swarmCacheCollection) as! [String:[TargetWrapper]]?
result = intermediate?.mapValues { $0.map { Target(from: $0) } }
result = transaction.object(forKey: swarmCacheKey, inCollection: swarmCacheCollection) as! [String:[LokiAPITarget]]?
}
return result ?? [:]
}
set {
let intermediate = newValue.mapValues { $0.map { TargetWrapper(from: $0) } }
storage.dbReadWriteConnection.readWrite { transaction in
transaction.setObject(intermediate, forKey: swarmCacheKey, inCollection: swarmCacheCollection)
transaction.setObject(newValue, forKey: swarmCacheKey, inCollection: swarmCacheCollection)
}
}
}
// MARK: Internal API
private static func getRandomSnode() -> Promise<Target> {
return Promise<Target> { seal in
seal.fulfill(Target(address: "http://13.236.173.190", port: defaultSnodePort)) // TODO: For debugging purposes
private static func getRandomSnode() -> Promise<LokiAPITarget> {
return Promise<LokiAPITarget> { seal in
seal.fulfill(LokiAPITarget(address: "http://13.236.173.190", port: defaultSnodePort)) // TODO: For debugging purposes
}
}
internal static func getCachedSnodes(for hexEncodedPublicKey: String) -> [Target] {
internal static func getCachedSnodes(for hexEncodedPublicKey: String) -> [LokiAPITarget] {
return swarmCache[hexEncodedPublicKey] ?? []
}
internal static func removeCachedSnode(_ target: Target, for hexEncodedPublicKey: String) {
internal static func removeCachedSnode(_ target: LokiAPITarget, for hexEncodedPublicKey: String) {
guard let cache = swarmCache[hexEncodedPublicKey] else { return }
swarmCache[hexEncodedPublicKey] = cache.filter { $0 != target }
}
internal static func fetchSwarmIfNeeded(for hexEncodedPublicKey: String) -> Promise<[Target]> {
internal static func fetchSwarmIfNeeded(for hexEncodedPublicKey: String) -> Promise<[LokiAPITarget]> {
if let cachedSwarm = swarmCache[hexEncodedPublicKey], cachedSwarm.count >= minimumSnodeCount {
return Promise<[Target]> { $0.fulfill(cachedSwarm) }
return Promise<[LokiAPITarget]> { $0.fulfill(cachedSwarm) }
} else {
let parameters: [String:Any] = [ "pubKey" : hexEncodedPublicKey ]
return getRandomSnode().then { invoke(.getSwarm, on: $0, associatedWith: hexEncodedPublicKey, parameters: parameters) }.map { parseTargets(from: $0) }.get { swarmCache[hexEncodedPublicKey] = $0 }
@ -54,16 +52,16 @@ public extension LokiAPI {
}
// MARK: Public API
internal static func getTargetSnodes(for hexEncodedPublicKey: String) -> Promise<[Target]> {
internal static func getTargetSnodes(for hexEncodedPublicKey: String) -> Promise<[LokiAPITarget]> {
// shuffled() uses the system's default random generator, which is cryptographically secure
return fetchSwarmIfNeeded(for: hexEncodedPublicKey).map { Array($0.shuffled().prefix(targetSnodeCount)) }
}
// MARK: Parsing
private static func parseTargets(from rawResponse: Any) -> [Target] {
private static func parseTargets(from rawResponse: Any) -> [LokiAPITarget] {
// TODO: For debugging purposes
// ========
let target = Target(address: "http://13.236.173.190", port: defaultSnodePort)
let target = LokiAPITarget(address: "http://13.236.173.190", port: defaultSnodePort)
return Array(repeating: target, count: 3)
// ========
// guard let json = rawResponse as? JSON, let addresses = json["snodes"] as? [String] else {
@ -77,7 +75,7 @@ public extension LokiAPI {
// MARK: Error Handling
internal extension Promise {
internal func handlingSwarmSpecificErrorsIfNeeded(for target: LokiAPI.Target, associatedWith hexEncodedPublicKey: String) -> Promise<T> {
internal func handlingSwarmSpecificErrorsIfNeeded(for target: LokiAPITarget, associatedWith hexEncodedPublicKey: String) -> Promise<T> {
return recover { error -> Promise<T> in
if let error = error as? NetworkManagerError {
switch error.statusCode {

@ -1,26 +0,0 @@
internal extension LokiAPI {
internal struct Target : Hashable {
internal let address: String
internal let port: UInt16
internal init(address: String, port: UInt16) {
self.address = address
self.port = port
}
internal init(from targetWrapper: TargetWrapper) {
self.address = targetWrapper.address
self.port = targetWrapper.port
}
internal enum Method : String {
/// Only supported by snode targets.
case getSwarm = "get_snodes_for_pubkey"
/// Only supported by snode targets.
case getMessages = "retrieve"
case sendMessage = "store"
}
}
}

@ -31,7 +31,7 @@ import PromiseKit
override private init() { }
// MARK: Internal API
internal static func invoke(_ method: Target.Method, on target: Target, associatedWith hexEncodedPublicKey: String, parameters: [String:Any] = [:], headers: [String:String] = [:], timeout: TimeInterval? = nil) -> RawResponsePromise {
internal static func invoke(_ method: LokiAPITarget.Method, on target: LokiAPITarget, associatedWith hexEncodedPublicKey: String, parameters: [String:Any] = [:], headers: [String:String] = [:], timeout: TimeInterval? = nil) -> RawResponsePromise {
let url = URL(string: "\(target.address):\(target.port)/\(version)/storage_rpc")!
let request = TSRequest(url: url, method: "POST", parameters: [ "method" : method.rawValue, "params" : parameters ])
request.allHTTPHeaderFields = headers
@ -44,11 +44,11 @@ import PromiseKit
}
internal static func getMessages(from target: Target, longPolling: Bool = true) -> MessageListPromise {
internal static func getMessages(from target: LokiAPITarget, longPolling: Bool = true) -> MessageListPromise {
return getRawMessages(from: target, longPolling: longPolling).map { process(rawMessages: $0, from: target) }
}
internal static func getRawMessages(from target: Target, longPolling: Bool = true) -> Promise<[JSON]> {
internal static func getRawMessages(from target: LokiAPITarget, longPolling: Bool = true) -> Promise<[JSON]> {
let hexEncodedPublicKey = OWSIdentityManager.shared().identityKeyPair()!.hexEncodedPublicKey
let lastHashValue = getLastMessageHashValue(for: target) ?? ""
let parameters: [String:Any] = [ "pubKey" : hexEncodedPublicKey, "lastHash" : lastHashValue ]
@ -60,7 +60,7 @@ import PromiseKit
}
}
internal static func process(rawMessages: [JSON], from target: Target) -> [SSKProtoEnvelope] {
internal static func process(rawMessages: [JSON], from target: LokiAPITarget) -> [SSKProtoEnvelope] {
updateLastMessageHashValueIfPossible(for: target, from: rawMessages)
let newRawMessages = removeDuplicates(from: rawMessages)
return parseProtoEnvelopes(from: newRawMessages)
@ -78,7 +78,7 @@ import PromiseKit
public static func sendSignalMessage(_ signalMessage: SignalMessage, with timestamp: UInt64, onP2PSuccess: @escaping () -> Void) -> Promise<Set<RawResponsePromise>> {
guard let lokiMessage = Message.from(signalMessage: signalMessage, with: timestamp) else { return Promise(error: Error.messageConversionFailed) }
let destination = lokiMessage.destination
func sendLokiMessage(_ lokiMessage: Message, to target: Target) -> RawResponsePromise {
func sendLokiMessage(_ lokiMessage: Message, to target: LokiAPITarget) -> RawResponsePromise {
let parameters = lokiMessage.toJSON()
return invoke(.sendMessage, on: target, associatedWith: destination, parameters: parameters)
}
@ -90,7 +90,7 @@ import PromiseKit
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
}
if let peer = LokiP2PManager.getInfo(for: destination), (lokiMessage.isPing || peer.isOnline) {
let target = Target(address: peer.address, port: peer.port)
let target = LokiAPITarget(address: peer.address, port: peer.port)
return Promise.value([ target ]).mapValues { sendLokiMessage(lokiMessage, to: $0) }.map { Set($0) }.retryingIfNeeded(maxRetryCount: maxRetryCount).get { _ in
LokiP2PManager.markOnline(destination)
onP2PSuccess()
@ -123,7 +123,7 @@ import PromiseKit
// The parsing utilities below use a best attempt approach to parsing; they warn for parsing failures but don't throw exceptions.
private static func updateLastMessageHashValueIfPossible(for target: Target, from rawMessages: [JSON]) {
private static func updateLastMessageHashValueIfPossible(for target: LokiAPITarget, from rawMessages: [JSON]) {
guard let lastMessage = rawMessages.last, let hashValue = lastMessage["hash"] as? String, let expiresAt = lastMessage["expiration"] as? Int else {
if rawMessages.count > 0 { Logger.warn("[Loki] Failed to update last message hash value from: \(rawMessages).") }
return

@ -0,0 +1,32 @@
internal final class LokiAPITarget : NSObject, NSCoding {
internal let address: String
internal let port: UInt16
// MARK: Types
internal enum Method : String {
/// Only supported by snode targets.
case getSwarm = "get_snodes_for_pubkey"
/// Only supported by snode targets.
case getMessages = "retrieve"
case sendMessage = "store"
}
// MARK: Initialization
internal init(address: String, port: UInt16) {
self.address = address
self.port = port
}
// MARK: Coding
internal init?(coder: NSCoder) {
address = coder.decodeObject(forKey: "address") as! String
port = coder.decodeObject(forKey: "port") as! UInt16
super.init()
}
internal func encode(with coder: NSCoder) {
coder.encode(address, forKey: "address")
coder.encode(port, forKey: "port")
}
}

@ -17,7 +17,7 @@
}
/// Our p2p address
private static var ourP2PAddress: LokiAPI.Target? = nil
private static var ourP2PAddress: LokiAPITarget? = nil
/// This is where we store the p2p details of our contacts
private static var peerInfo = [String:PeerInfo]()
@ -29,7 +29,7 @@
/// - Parameter url: The url to our local server
@objc public static func setOurP2PAddress(url: URL) {
guard let scheme = url.scheme, let host = url.host, let port = url.port else { return }
let target = LokiAPI.Target(address: "\(scheme)://\(host)", port: UInt16(port))
let target = LokiAPITarget(address: "\(scheme)://\(host)", port: UInt16(port))
ourP2PAddress = target
}

@ -1,22 +0,0 @@
@objc internal final class TargetWrapper : NSObject, NSCoding {
internal let address: String
internal let port: UInt16
internal init(from target: LokiAPI.Target) {
address = target.address
port = target.port
super.init()
}
internal init?(coder: NSCoder) {
address = coder.decodeObject(forKey: "address") as! String
port = coder.decodeObject(forKey: "port") as! UInt16
super.init()
}
internal func encode(with coder: NSCoder) {
coder.encode(address, forKey: "address")
coder.encode(port, forKey: "port")
}
}
Loading…
Cancel
Save