Add an option to choose a custom queue for completion handlers (#334)

* [refact] Rename queues

This should make the code a tad more readable.

* Add an option to choose a custom queue for completion handlers

This required quite a bit of refactoring:

- Eliminate all explicit references to the main dispatch queue and the main operation queue. This required creating a dedicated queue for aggregate operations (like “delete by query”), called `inMemoryQueue`.

- Serialize the fallback logic in mixed online/offline requests. Since the completion queue can be parallel, this is required now.

NOTE: I chose to expose an `OperationQueue` instead of a `DispatchQueue` because (as of iOS 8.0) you can wrap a dispatch queue inside an operation queue (using `underlyingQueue`) whereas the opposite is not true.

Fixes #69.

Author: Clément Le Provost

Source/AbstractClient.swift

@@ -185,14 +185,29 @@ internal struct HostStatus {
     // NOTE: Not constant only for the sake of mocking during unit tests.
     var session: URLSession
 
-    /// Operation queue used to keep track of requests.
+    /// Operation queue used to keep track of network requests.
     /// `Request` instances are inherently asynchronous, since they are merely wrappers around `NSURLSessionTask`.
-    /// The sole purpose of the queue is to retain them for the duration of their execution!
+    /// The sole purpose of this queue is to retain them for the duration of their execution!
     ///
-    let requestQueue: OperationQueue
+    /// + Warning: This queue must allow enough parallel executions to avoid stalling when the response time is high.
+    ///   Because it's merely a memory management tool, we don't want it to be the limiting factor; `URLSession`
+    ///   already has its own logic of connection pooling.
+    ///
+    let onlineRequestQueue: OperationQueue
+    
+    /// Maximum number of concurrent requests we allow per connection.
+    /// This setting is used to dimension `onlineRequestQueue`.
+    private let maxConcurrentRequestCountPerConnection = 4
 
-    /// Dispatch queue used to run completion handlers.
-    internal var completionQueue = DispatchQueue.main
+    /// Operation queue used to run completion handlers.
+    /// Default = main queue.
+    ///
+    /// + Note: This will affect completion handlers of all methods on all indices attached to this client.
+    ///
+    /// + Warning: The queue is not retained by the client. You must ensure that it remains valid for the lifetime of
+    ///   the client.
+    ///
+    @objc public weak var completionQueue = OperationQueue.main
 
     // MARK: Constant
 
@@ -234,8 +249,9 @@ internal struct HostStatus {
         configuration.httpAdditionalHeaders = fixedHTTPHeaders
         session = Foundation.URLSession(configuration: configuration)
 
-        requestQueue = OperationQueue()
-        requestQueue.maxConcurrentOperationCount = 8
+        onlineRequestQueue = OperationQueue()
+        onlineRequestQueue.name = "AlgoliaSearch online requests"
+        onlineRequestQueue.maxConcurrentOperationCount = configuration.httpMaximumConnectionsPerHost * maxConcurrentRequestCountPerConnection
 
         super.init()
 
@@ -333,7 +349,7 @@ internal struct HostStatus {
     func performHTTPQuery(path: String, method: HTTPMethod, body: JSONObject?, hostnames: [String], isSearchQuery: Bool = false, completionHandler: CompletionHandler? = nil) -> Operation {
         let request = newRequest(method: method, path: path, body: body, hostnames: hostnames, isSearchQuery: isSearchQuery, completion: completionHandler)
         request.completionQueue = self.completionQueue
-        requestQueue.addOperation(request)
+        onlineRequestQueue.addOperation(request)
         return request
     }
 

Source/AsyncOperation.swift

@@ -114,21 +114,27 @@ internal class AsyncOperationWithCompletion: AsyncOperation {
     /// User completion block to be called.
     let completion: CompletionHandler?
 
-    /// Dispatch queue used to execute the completion handler.
-    var completionQueue: DispatchQueue?
+    /// Operation queue used to execute the completion handler.
+    weak var completionQueue: OperationQueue?
 
     init(completionHandler: CompletionHandler?) {
         self.completion = completionHandler
     }
+
+    override func start() {
+        // The completion queue should not be the same as this operation's queue, otherwise a deadlock will result.
+        assert(OperationQueue.current != self.completionQueue)
+    }
 
+
     /// Finish this operation.
     /// This method should be called exactly once per operation.
     internal func callCompletion(content: JSONObject?, error: Error?) {
         if _cancelled {
             return
         }
         if let completionQueue = completionQueue {
-            completionQueue.async {
+            completionQueue.addOperation {
                 self._callCompletion(content: content, error: error)
             }
         } else {

Source/Client.swift

@@ -48,6 +48,11 @@ import Foundation
     ///
     var indices: NSMapTable<NSString, AnyObject> = NSMapTable(keyOptions: [.strongMemory], valueOptions: [.weakMemory])
 
+    /// Queue for purely in-memory operations (no I/Os).
+    /// Typically used for aggregate, concurrent operations.
+    ///
+    internal var inMemoryQueue: OperationQueue = OperationQueue()
+    
     // MARK: Initialization
 
     /// Create a new Algolia Search client.
@@ -72,6 +77,7 @@ import Foundation
         let readHosts = [ "\(appID)-dsn.algolia.net" ] + fallbackHosts
         let writeHosts = [ "\(appID).algolia.net" ] + fallbackHosts
         super.init(appID: appID, apiKey: apiKey, readHosts: readHosts, writeHosts: writeHosts)
+        inMemoryQueue.maxConcurrentOperationCount = onlineRequestQueue.maxConcurrentOperationCount
     }
 
     /// Obtain a proxy to an Algolia index (no server call required by this method).

Source/Index.swift

@@ -331,12 +331,11 @@ import Foundation
         let cacheKey = "\(path)_body_\(request)"
         if let content = searchCache?.objectForKey(cacheKey) {
             // We *have* to return something, so we create a simple operation.
-            // Note that its execution will be deferred until the next iteration of the main run loop.
             let operation = AsyncBlockOperation(completionHandler: completionHandler) {
                 return (content, nil)
             }
             operation.completionQueue = client.completionQueue
-            OperationQueue.main.addOperation(operation)
+            client.inMemoryQueue.addOperation(operation)
             return operation
         }
         // Otherwise, run an online query.
@@ -487,7 +486,7 @@ import Foundation
     @objc
     @discardableResult public func waitTask(withID taskID: Int, completionHandler: @escaping CompletionHandler) -> Operation {
         let operation = WaitOperation(index: self, taskID: taskID, completionHandler: completionHandler)
-        operation.start()
+        client.inMemoryQueue.addOperation(operation)
         return operation
     }
 
@@ -553,7 +552,7 @@ import Foundation
     @objc
     @discardableResult public func deleteByQuery(_ query: Query, completionHandler: CompletionHandler? = nil) -> Operation {
         let operation = DeleteByQueryOperation(index: self, query: query, completionHandler: completionHandler)
-        operation.start()
+        client.inMemoryQueue.addOperation(operation)
         return operation
     }