added logic to cancel in-flight promises when connection drops

Sources/Redis/Client/RedisClient.swift

@@ -20,6 +20,9 @@ public final class RedisClient: DatabaseConnection, BasicWorker {
 
  /// The channel
  private let channel: Channel
+
+ /// Stores the inflight promises so they can be fulfilled when the channel drops
+ var inflightPromises: [String:EventLoopPromise<RedisData>] = [:]
 
  /// Creates a new Redis client on the provided data source and sink.
  init(queue: RedisCommandHandler, channel: Channel) {
@@ -28,6 +31,12 @@ public final class RedisClient: DatabaseConnection, BasicWorker {
  self.extend = [:]
  self.isClosed = false
  channel.closeFuture.always {
+ // send closed error for the promises that have not been fulfilled
+ for promise in self.inflightPromises.values {
+ promise.fail(error: ChannelError.ioOnClosedChannel)
+ }
+ self.inflightPromises.removeAll()
+
  self.isClosed = true
  }
  }
@@ -55,6 +64,13 @@ public final class RedisClient: DatabaseConnection, BasicWorker {
  // create a new promise to fulfill later
  let promise = eventLoop.newPromise(RedisData.self)
 
+ // logic to store in-flight requests that can be cancelled when connection drops
+ let key = UUID().uuidString
+ self.inflightPromises[key] = promise
+ promise.futureResult.always {
+ self.inflightPromises.removeValue(forKey: key)
+ }
+
  // write the message and the promise to the channel, which the `RequestResponseHandler` will capture
  return self.channel.writeAndFlush((message, promise))
  .flatMap { return promise.futureResult }

Tests/RedisTests/RedisDatabaseTests.swift

@@ -26,6 +26,34 @@ class RedisDatabaseTests: XCTestCase {
  try redis.delete("hello").wait()
  XCTAssertNil(try redis.get("hello", as: String.self).wait())
  }
+
+ func testDroppedConnection() throws {
+ let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+ let config = RedisClientConfig.makeTest()
+ let database = try RedisDatabase(config: config)
+ let redis = try database.newConnection(on: group).wait()
+ defer { redis.close() }
+
+ let timeout: UInt32 = 1
+ var dataReceived = false
+ var errorReceived = false
+
+ let command = RedisData.array(["brpop", "hello", "\(timeout)"].map { RedisData(bulk: $0) })
+ _ = redis.send(command).do { data in
+ dataReceived = true
+ }.catch { error in
+ errorReceived = true
+ }
+
+ // Close the connection
+ redis.close()
+
+ // Sleep for an extra second seconds to give the transaction time to complete
+ sleep(timeout+1)
+
+ XCTAssertEqual(dataReceived, false)
+ XCTAssertEqual(errorReceived, true)
+ }
 
  func testSelect() throws {
  let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)