Added ConnectionPool for redis. Fixed bug with hostName.

Author: John-Connolly

Sources/Configuration.swift

@@ -23,7 +23,7 @@ public struct Configuration {
     /// Used in order to specify a custom queue to consume from
     let queue: String
     /// Used for consumer specific processing queues.
-    /// If not provided Host.current().name will be used
+    /// If not provided the servers hostname will be used.
     let consumerName: String?
 
     let middleware: [Middleware]
@@ -59,16 +59,19 @@ public struct RedisConfig {
     let port: UInt16
 
     let password: String?
+    /// Max number of connections that will be created by the connection pool.
+    let connections: Int
 
     public static var development: RedisConfig {
-        return .init(redisDB: nil, hostname: "127.0.0.1", port: 6379, password: nil)
+        return .init(redisDB: nil, hostname: "127.0.0.1", port: 6379, password: nil, connections: 4)
     }
 
-    public init(redisDB: Int?, hostname: String, port: UInt16, password: String?) {
+    public init(redisDB: Int?, hostname: String, port: UInt16, password: String?, connections: Int) {
         self.redisDB = redisDB
         self.hostname = hostname
         self.port = port
         self.password = password
+        self.connections = connections
     }
 
 }

Sources/ConnectionPool.swift

@@ -0,0 +1,59 @@
+//
+//  ConnectionPool.swift
+//  SwiftQ
+//
+//  Created by John Connolly on 2017-08-09.
+//
+//
+
+import Foundation
+import Dispatch
+
+final class ConnectionPool<T> {
+    
+    var connections = [T]()
+    
+    private let connectionFactory: () throws -> T
+    
+    private let max: Int
+    
+    private let semaphore: DispatchSemaphore
+    /// Semaphore seems to perform much better that a serial queue
+    /// Probably related to the extra heap allocation of the sync { } closure.
+    private let lock = DispatchSemaphore(value: 1)
+    
+    
+    init(max: Int, factory: @escaping () throws -> T) throws {
+        self.max = max
+        self.connectionFactory = factory
+        self.semaphore = DispatchSemaphore(value: max)
+        
+        try (0...max).forEach { _ in
+            connections.append(try factory())
+        }
+    }
+    
+    /// Borrows a connection from the pool.  Any borrowed connections
+    /// have to be returned.  If all connections are borrowed the semaphore
+    /// will block until one is returned.
+    func borrow() -> T {
+        semaphore.wait()
+        defer {
+            lock.signal()
+        }
+        lock.wait()
+        return connections.removeFirst()
+        
+    }
+    
+    /// Returns the connection to the pool.
+    func takeBack(connection: T) {
+        semaphore.signal()
+        defer {
+            lock.signal()
+        }
+        lock.wait()
+        connections.append(connection)
+    }
+    
+}

Sources/Decoder.swift

@@ -12,11 +12,13 @@ struct Decoder {
 
     let types:  [Task.Type]
 
-    private var zippedTasks: [(String , Task.Type)] {
+    private let zippedTasks: [(String , Task.Type)]
+    
+    init(types: [Task.Type]) {
+        self.types = types
         let taskNames = types.map(String.init(describing:))
-        return zip(taskNames, types).array
+        self.zippedTasks = zip(taskNames, types).array
     }
-    
 
     /// Returns the correct task type based on the zipped tasks
     func decode(data: Foundation.Data) throws -> DecoderResult {

Sources/Host.swift

@@ -0,0 +1,37 @@
+//
+//  Host.swift
+//  SwiftQ
+//
+//  Created by John Connolly on 2017-08-11.
+//
+//
+
+import Foundation
+#if os(Linux)
+    import Glibc
+#endif
+
+struct Host {
+    
+    let name: String
+    
+    
+    init() {
+        self.name = Host.currentHostName()
+    }
+    
+    /// https://github.com/apple/swift-corelibs-foundation/blob/master/Foundation/Host.swift
+    private static func currentHostName() -> String {
+        let hname = UnsafeMutablePointer<Int8>.allocate(capacity: Int(NI_MAXHOST))
+        defer {
+            hname.deinitialize()
+            hname.deallocate(capacity: Int(NI_MAXHOST))
+        }
+        let r = gethostname(hname, Int(NI_MAXHOST))
+        if r < 0 || hname[0] == 0 {
+            return "localhost"
+        }
+        return String(cString: hname)
+    }
+    
+}

Sources/IPAddress.swift

@@ -8,41 +8,43 @@
 
 import Foundation
 import Redis
-import Dispatch
 
 final class RedisAdaptor: Adaptor {
-   
-    private let client: Redis.TCPClient
 
-    private let dispatchQueue = DispatchQueue(label: "com.swiftq.redis")
+    private let client: Redis.TCPClient
 
+    let pool: ConnectionPool<Redis.TCPClient>
 
     init(config: RedisConfig) throws {
+        self.pool = try ConnectionPool(max: config.connections) {
+            return try Redis.TCPClient(hostname: config.hostname, port: config.port, password: config.password)
+        }
         self.client = try Redis.TCPClient(hostname: config.hostname, port: config.port, password: config.password)
         if let database = config.redisDB {
-            try execute(Command(command: "select", args: [.string(database.description)]))
+            try execute(Command(command: "SELECT", args: [.string(database.description)]))
         }
     }
 
-    
     @discardableResult
     func execute(_ command: Command) throws -> RedisResponseRepresentable {
-        return try dispatchQueue.sync {
-            VaporRedisResponse(response: try client.command(Redis.Command(command.command), command.bytes))
-        }
+        let client = pool.borrow()
+        defer { pool.takeBack(connection: client) }
+        return VaporRedisResponse(response: try client.command(Redis.Command(command.command), command.bytes))
     }
 
 
     @discardableResult
     func pipeline(_ commands: () -> ([Command])) throws -> [RedisResponseRepresentable] {
-        return try dispatchQueue.sync {
-            let arguments = commands()
-            let pipeline = client.makePipeline()
-            try arguments.forEach { argument in
-                try pipeline.enqueue(Redis.Command(argument.command), argument.bytes)
-            }
-            return try pipeline.execute().map(VaporRedisResponse.init)
+        let client = pool.borrow()
+        defer {
+            pool.takeBack(connection: client)
+        }
+        let arguments = commands()
+        let pipeline = client.makePipeline()
+        try arguments.forEach { argument in
+            try pipeline.enqueue(Redis.Command(argument.command), argument.bytes)
         }
+        return try pipeline.execute().map(VaporRedisResponse.init)
     }
 
 }

Sources/RedisAdaptor.swift

@@ -9,13 +9,12 @@
 import Foundation
 import Redis
 
-
 struct VaporRedisResponse: RedisResponseRepresentable {
 
     let response: Redis.Data?
 
     var data: Foundation.Data? {
-        return response?.string?.data(using: .utf8)
+        return response?.bytes.map(Foundation.Data.init(bytes:))
     }
 
     var string: String? {
@@ -27,9 +26,9 @@ struct VaporRedisResponse: RedisResponseRepresentable {
     }
 
     var array: [Foundation.Data]? {
-        return response?
-            .array?
-            .flatMap { $0?.string?.data(using: .utf8) }
+        return response?.array?.flatMap { data -> Foundation.Data? in
+            return data?.bytes.map(Foundation.Data.init(bytes:))
+        }
     }
 
 }

Sources/RedisResponse.swift

@@ -13,9 +13,7 @@ final class ReliableQueue {
 
     private let redisAdaptor: Adaptor
 
-    private let blockingRedisAdaptor: Adaptor // Maybe have a pool here instead
-    
-    private let ipAddress = IPAddress().address
+    private let hostName = Host().name
 
     private let consumer: String?
 
@@ -26,7 +24,6 @@ final class ReliableQueue {
         self.queue = queue
         self.consumer = consumer
         self.redisAdaptor = try RedisAdaptor(config: config)
-        self.blockingRedisAdaptor = try RedisAdaptor(config: config)
     }
 
 
@@ -35,7 +32,7 @@ final class ReliableQueue {
     }
 
     var consumerName: String {
-        return consumer ?? ipAddress
+        return consumer ?? hostName
     }
 
 
@@ -112,11 +109,11 @@ final class ReliableQueue {
             .string(RedisKey.workQ(queue).name),
             .string(processingQKey),
             .string("0")])
-        return try blockingRedisAdaptor.execute(dequeueCommand).string
+        return try redisAdaptor.execute(dequeueCommand).string
             .map { id in
                 return Command(command: "GET", args:[.string(id)])
             }.flatMap { command in
-                return try blockingRedisAdaptor.execute(command).data
+                return try redisAdaptor.execute(command).data
         }
     }
 
@@ -126,9 +123,11 @@ final class ReliableQueue {
         let incrKey = success ? RedisKey.success(consumerName).name : RedisKey.failure(consumerName).name
         try redisAdaptor.pipeline {
             let commands = [
+                Command(command: "MULTI"),
                 Command(command: "LREM", args: [.string(processingQKey), .string("0"), .string(item.uuid)]),
                 Command(command: "INCR", args: [.string(incrKey) ]),
-                Command(command: "DEL", args: [.string(item.uuid)])
+                Command(command: "DEL", args: [.string(item.uuid)]),
+                Command(command: "EXEC")
             ]
             return commands
         }

Sources/ReliableQueue.swift

@@ -46,7 +46,6 @@ final class Worker {
                 do {
                     guard let data = try self.reliableQueue.dequeue() else { return }
                     try self.decode(data)
-                    
                 } catch {
                     Logger.log(("Decoding Failure", error))
                 }