Merge pull request #35 from sliemeobn/add-sendable

Complete sendable-checking and refactoring for simpler state-handling

Author: funcmike

Package.resolved

@@ -1,6 +1,4 @@
-// swift-tools-version: 5.7
-// The swift-tools-version declares the minimum version of Swift required to build this package.
-
+// swift-tools-version: 5.8
 import PackageDescription
 
 let package = Package(
@@ -12,14 +10,18 @@ let package = Package(
     dependencies: [
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.48.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.23.0"),
-        .package(url: "https://github.com/apple/swift-collections.git",  .upToNextMajor(from: "1.0.0"))
+        .package(url: "https://github.com/apple/swift-collections.git", .upToNextMajor(from: "1.0.0")),
     ],
     targets: [
         .target(
             name: "AMQPProtocol",
             dependencies: [
                 .product(name: "NIOCore", package: "swift-nio"),
-            ]),
+            ],
+            swiftSettings: [
+                .enableUpcomingFeature("StrictConcurrency"),
+            ]
+        ),
         .target(
             name: "AMQPClient",
             dependencies: [
@@ -29,9 +31,17 @@ let package = Package(
                 .product(name: "NIOConcurrencyHelpers", package: "swift-nio"),
                 .product(name: "NIOSSL", package: "swift-nio-ssl"),
                 .product(name: "Collections", package: "swift-collections"),
-            ]),
+            ],
+            swiftSettings: [
+                .enableUpcomingFeature("StrictConcurrency"),
+            ]
+        ),
         .testTarget(
             name: "AMQPClientTests",
-            dependencies: ["AMQPClient"]),
+            dependencies: ["AMQPClient"],
+            swiftSettings: [
+                .enableUpcomingFeature("StrictConcurrency"),
+            ]
+        ),
     ]
 )

Package.swift

@@ -11,34 +11,33 @@
 //
 //===----------------------------------------------------------------------===//
 
-
 struct AMQPChannels {
     private enum ChannelSlot {
         case reserved
         case channel(AMQPChannel)
     }
-    
+
     private var channels: [UInt16: ChannelSlot] = [:]
-    
+
     let channelMax: UInt16
-    
+
     init(channelMax: UInt16) {
         self.channelMax = channelMax
     }
 
     func get(id: UInt16) -> AMQPChannel? {
-        guard case let .channel(channel) = self.channels[id] else {
+        guard case let .channel(channel) = channels[id] else {
             return nil
         }
         return channel
     }
-    
+
     mutating func reserveNext() -> UInt16? {
-        guard self.channels.count < channelMax else {
+        guard channels.count < channelMax else {
             return nil
         }
-            
-        for i in 1...channelMax where channels[i] == nil {
+
+        for i in 1 ... channelMax where channels[i] == nil {
             channels[i] = .reserved
             return i
         }
@@ -47,10 +46,10 @@ struct AMQPChannels {
     }
 
     mutating func add(channel: AMQPChannel) {
-        self.channels[channel.ID] = .channel(channel)
+        channels[channel.ID] = .channel(channel)
     }
 
     mutating func remove(id: UInt16) {
-      self.channels[id] = nil
+        channels[id] = nil
     }
 }

Sources/AMQPClient/AMQPChannel.swift

@@ -11,40 +11,36 @@
 //
 //===----------------------------------------------------------------------===//
 
+import NIOConcurrencyHelpers
 import NIOCore
 import NIOPosix
 import NIOSSL
-import NIOConcurrencyHelpers
 
-public final class AMQPConnection {
+public final class AMQPConnection: Sendable {
     internal enum ConnectionState {
         case open
         case shuttingDown
         case closed
     }
-    
+
     public var isConnected: Bool {
         // `Channel.isActive` is set to false before the `closeFuture` resolves in cases where the channel might be
         // closed, or closing, before our state has been updated
-        return self.channel.isActive && self.state.withLockedValue { $0 == .open }
+        return channel.isActive && state.withLockedValue { $0 == .open }
     }
 
-    public var closeFuture: NIOCore.EventLoopFuture<Void> {
-        return self.channel.closeFuture
-    }
-
-    public var eventLoop: EventLoop { return self.channel.eventLoop }
+    public var closeFuture: NIOCore.EventLoopFuture<Void> { connectionHandler.channel.closeFuture }
+    public var eventLoop: EventLoop { return connectionHandler.channel.eventLoop }
 
-    private let channel: NIOCore.Channel
-    private let multiplexer: AMQPConnectionMultiplexHandler
+    private let connectionHandler: AMQPConnectionHandler
+    private var channel: NIOCore.Channel { connectionHandler.channel }
 
     private let state = NIOLockedValueBox(ConnectionState.open)
     private let channels: NIOLockedValueBox<AMQPChannels>
 
-    init(channel: NIOCore.Channel, multiplexer: AMQPConnectionMultiplexHandler, channelMax: UInt16) {
-        self.channel = channel
-        self.multiplexer = multiplexer
-        self.channels = .init(AMQPChannels(channelMax: channelMax))
+    init(connectionHandler: AMQPConnectionHandler, channelMax: UInt16) {
+        self.connectionHandler = connectionHandler
+        channels = .init(AMQPChannels(channelMax: channelMax))
     }
 
     /// Connect to broker.
@@ -53,22 +49,15 @@ public final class AMQPConnection {
     ///     - config: Configuration data.
     /// - Returns:  EventLoopFuture with AMQP Connection.
     public static func connect(use eventLoop: EventLoop, from config: AMQPConnectionConfiguration) -> EventLoopFuture<AMQPConnection> {
-        let promise = eventLoop.makePromise(of: AMQPResponse.self)
-        let multiplexer = AMQPConnectionMultiplexHandler(eventLoop: eventLoop, config: config.server, onReady: promise)
-
-        return eventLoop.flatSubmit { () -> EventLoopFuture<AMQPConnection> in
-            let result = self.boostrapChannel(use: eventLoop, from: config, with: multiplexer).flatMap { channel in
-                promise.futureResult.flatMapThrowing { response in
-                    guard case .connection(let connection) = response, case .connected(let connected) = connection else {
-                        throw AMQPConnectionError.invalidResponse(response)
-                    }
-
-                    return AMQPConnection(channel: channel, multiplexer: multiplexer, channelMax: connected.channelMax)
+        eventLoop.flatSubmit {
+            self.boostrapChannel(use: eventLoop, from: config).flatMap { connectionHandler in
+                connectionHandler.startConnection().map {
+                    AMQPConnection(
+                        connectionHandler: connectionHandler,
+                        channelMax: $0.channelMax
+                    )
                 }
             }
-
-            result.whenFailure { err in multiplexer.failAllResponses(because: err) }
-            return result
         }
     }
 
@@ -77,24 +66,22 @@ public final class AMQPConnection {
     /// Channel ID is automatically assigned (next free one).
     /// - Returns: EventLoopFuture with AMQP Channel.
     public func openChannel() -> EventLoopFuture<AMQPChannel> {
-        guard self.isConnected else { return self.eventLoop.makeFailedFuture(AMQPConnectionError.connectionClosed()) }
+        guard isConnected else { return eventLoop.makeFailedFuture(AMQPConnectionError.connectionClosed()) }
 
         let channelID = channels.withLockedValue { $0.reserveNext() }
-        
+
         guard let channelID = channelID else {
-            return self.eventLoop.makeFailedFuture(AMQPConnectionError.tooManyOpenedChannels)
+            return eventLoop.makeFailedFuture(AMQPConnectionError.tooManyOpenedChannels)
         }
 
-        return self.eventLoop.flatSubmit {
-            let future = self.multiplexer.openChannel(id: channelID)
+        let future = connectionHandler.openChannel(id: channelID)
 
-            future.whenFailure { _ in self.channels.withLockedValue { $0.remove(id: channelID) } }
- 
-            return future.map { channel in
-                let amqpChannel = AMQPChannel(channelID: channelID, eventLoop: self.eventLoop, channel: channel)
-                self.channels.withLockedValue { $0.add(channel: amqpChannel) }
-                return amqpChannel
-            }
+        future.whenFailure { _ in self.channels.withLockedValue { $0.remove(id: channelID) } }
+
+        return future.map { channel in
+            let amqpChannel = AMQPChannel(channelID: channelID, eventLoop: self.eventLoop, channel: channel)
+            self.channels.withLockedValue { $0.add(channel: amqpChannel) }
+            return amqpChannel
         }
     }
 
@@ -109,69 +96,64 @@ public final class AMQPConnection {
                 state = .shuttingDown
                 return true
             }
-            
+
             return false
         }
-        
-        guard shouldClose else { return self.channel.closeFuture }
-        
-        return self.eventLoop.flatSubmit {
-            let result = self.multiplexer.close(reason: reason, code: code)
-                .map { () in
-                    return nil as Error?
+
+        guard shouldClose else { return closeFuture }
+
+        let result = connectionHandler.close(reason: reason, code: code)
+            .map { () in
+                nil as Error?
+            }
+            .recover { $0 }
+            .flatMap { result in
+                self.channel.close().map {
+                    self.state.withLockedValue { $0 = .closed }
+                    return (result, nil) as (Error?, Error?)
                 }
-                .recover { $0 }
-                .flatMap { result in
-                    self.channel.close().map {
+                .recover { error in
+                    if case ChannelError.alreadyClosed = error {
                         self.state.withLockedValue { $0 = .closed }
-                        return (result, nil) as (Error?, Error?)
-                    }
-                    .recover { error in
-                        if case ChannelError.alreadyClosed = error  {
-                            self.state.withLockedValue { $0 = .closed }
-                            return (result, nil)
-                        }
-                        
-                        return (result, error)
+                        return (result, nil)
                     }
+
+                    return (result, error)
                 }
-            return result.flatMapThrowing {
-                let (broker, conn) = $0
-                if (broker ?? conn) != nil { throw AMQPConnectionError.connectionClose(broker: broker, connection: conn) }
-                return ()
             }
+        return result.flatMapThrowing {
+            let (broker, conn) = $0
+            if (broker ?? conn) != nil { throw AMQPConnectionError.connectionClose(broker: broker, connection: conn) }
+            return ()
         }
     }
 
     private static func boostrapChannel(
         use eventLoop: EventLoop,
-        from config: AMQPConnectionConfiguration,
-        with handler: AMQPConnectionMultiplexHandler
-    ) -> EventLoopFuture<NIOCore.Channel> {
-        let channelPromise = eventLoop.makePromise(of: NIOCore.Channel.self)
-
+        from config: AMQPConnectionConfiguration
+    ) -> EventLoopFuture<AMQPConnectionHandler> {
         do {
             let bootstrap = try boostrapClient(use: eventLoop, from: config)
+            let multiplexer = NIOLoopBound(
+                AMQPConnectionMultiplexHandler(eventLoop: eventLoop, config: config.server),
+                eventLoop: eventLoop)
 
-            bootstrap
+            return bootstrap
                 .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
                 .channelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
                 .connectTimeout(config.server.timeout)
                 .channelInitializer { channel in
                     channel.pipeline.addHandlers([
                         MessageToByteHandler(AMQPFrameEncoder()),
                         ByteToMessageHandler(AMQPFrameDecoder()),
-                        handler
+                        multiplexer.value,
                     ])
                 }
                 .connect(host: config.server.host, port: config.server.port)
-                .map { channelPromise.succeed($0) }
-                .cascadeFailure(to: channelPromise)
+                .map { AMQPConnectionHandler(channel: $0, multiplexer: multiplexer.value) }
         } catch {
-            channelPromise.fail(error)
+            return eventLoop.makeFailedFuture(error)
         }
-
-        return channelPromise.futureResult        
     }
 
     private static func boostrapClient(
@@ -182,17 +164,17 @@ public final class AMQPConnection {
             preconditionFailure("Cannot create bootstrap for the supplied EventLoop")
         }
 
-        switch config.connection {            
-        case .plain: 
+        switch config.connection {
+        case .plain:
             return NIOClientTCPBootstrap(clientBootstrap, tls: NIOInsecureNoTLS())
-        case .tls(let tls, let sniServerName):
+        case let .tls(tls, sniServerName):
             let sslContext = try NIOSSLContext(configuration: tls ?? TLSConfiguration.clientDefault)
             let tlsProvider = try NIOSSLClientTLSProvider<ClientBootstrap>(context: sslContext, serverHostname: sniServerName ?? config.server.host)
             let bootstrap = NIOClientTCPBootstrap(clientBootstrap, tls: tlsProvider)
             return bootstrap.enableTLS()
-        }        
+        }
     }
-    
+
     deinit {
         if isConnected {
             assertionFailure("close() was not called before deinit!")