Support setting enablePeerToPeer. (#22) (#24)

Motivation:

Some users may wish to use peer-to-peer networking with bonjour.

Modifications:

Expose peer-to-peer networking via a new channel option.

Result:

Users will be able to use peer-to-peer networking

(cherry picked from commit e9c1e41fd1)

Sources/NIOTransportServices/NIOTSChannelOptions.swift

@@ -32,8 +32,23 @@ public enum NIOTSWaitForActivityOption: ChannelOption {
 }
 
 
+/// `NIOTSEnablePeerToPeerOption` controls whether the `Channel` will advertise services using peer-to-peer
+/// connectivity. Setting this to true is the equivalent of setting `NWParameters.enablePeerToPeer` to
+/// `true`. By default this option is set to `false`.
+///
+/// This option must be set on the bootstrap: setting it after the channel is initialized will have no effect.
+public enum NIOTSEnablePeerToPeerOption: ChannelOption {
+    public typealias AssociatedValueType = ()
+    public typealias OptionType = Bool
+
+    case const(())
+}
+
+
 /// Options that can be set explicitly and only on bootstraps provided by `NIOTransportServices`.
 public struct NIOTSChannelOptions {
     /// - seealso: `NIOTSWaitForActivityOption`.
     public static let waitForActivity = NIOTSWaitForActivityOption.const(())
+
+    public static let enablePeerToPeer = NIOTSEnablePeerToPeerOption.const(())
 }

Sources/NIOTransportServices/NIOTSConnectionChannel.swift

@@ -201,6 +201,9 @@ internal final class NIOTSConnectionChannel {
     /// The value of SO_REUSEPORT.
     private var reusePort = false
 
+    /// Whether to use peer-to-peer connectivity when connecting to Bonjour services.
+    private var enablePeerToPeer = false
+
     /// Create a `NIOTSConnectionChannel` on a given `NIOTSEventLoop`.
     ///
     /// Note that `NIOTSConnectionChannel` objects cannot be created on arbitrary loops types.
@@ -318,6 +321,8 @@ extension NIOTSConnectionChannel: Channel {
                 // We're in waiting now, so we should drop the connection.
                 self.close0(error: err, mode: .all, promise: nil)
             }
+        case is NIOTSEnablePeerToPeerOption:
+            self.enablePeerToPeer = value as! NIOTSEnablePeerToPeerOption.OptionType
         default:
             fatalError("option \(type(of: option)).\(option) not supported")
         }
@@ -361,6 +366,8 @@ extension NIOTSConnectionChannel: Channel {
             return self.backpressureManager.waterMarks as! T.OptionType
         case _ as NIOTSWaitForActivityOption:
             return self.options.waitForActivity as! T.OptionType
+        case is NIOTSEnablePeerToPeerOption:
+            return self.enablePeerToPeer as! T.OptionType
         default:
             fatalError("option \(type(of: option)).\(option) not supported")
         }
@@ -441,6 +448,8 @@ extension NIOTSConnectionChannel: StateManagedChannel {
         // either.
         parameters.allowLocalEndpointReuse = self.reuseAddress || self.reusePort
 
+        parameters.includePeerToPeer = self.enablePeerToPeer
+
         let connection = NWConnection(to: target, using: parameters)
         connection.stateUpdateHandler = self.stateUpdateHandler(newState:)
         connection.betterPathUpdateHandler = self.betterPathHandler

Sources/NIOTransportServices/NIOTSListenerChannel.swift

@@ -77,6 +77,9 @@ internal final class NIOTSListenerChannel {
     /// The value of SO_REUSEPORT.
     private var reusePort = false
 
+    /// Whether to enable peer-to-peer connectivity when using Bonjour services.
+    private var enablePeerToPeer = false
+
     /// Create a `NIOTSListenerChannel` on a given `NIOTSEventLoop`.
     ///
     /// Note that `NIOTSListenerChannel` objects cannot be created on arbitrary loops types.
@@ -166,6 +169,8 @@ extension NIOTSListenerChannel: Channel {
             default:
                 try self.tcpOptions.applyChannelOption(option: optionValue, value: value as! SocketOptionValue)
             }
+        case is NIOTSEnablePeerToPeerOption:
+            self.enablePeerToPeer = value as! NIOTSEnablePeerToPeerOption.OptionType
         default:
             fatalError("option \(option) not supported")
         }
@@ -201,6 +206,8 @@ extension NIOTSListenerChannel: Channel {
             default:
                 return try self.tcpOptions.valueFor(socketOption: optionValue) as! T.OptionType
             }
+        case is NIOTSEnablePeerToPeerOption:
+            return self.enablePeerToPeer as! T.OptionType
         default:
             fatalError("option \(option) not supported")
         }
@@ -277,6 +284,8 @@ extension NIOTSListenerChannel: StateManagedChannel {
         // either.
         parameters.allowLocalEndpointReuse = self.reuseAddress || self.reusePort
 
+        parameters.includePeerToPeer = self.enablePeerToPeer
+
         let listener: NWListener
         do {
             listener = try NWListener(using: parameters)

Tests/NIOTransportServicesTests/NIOTSConnectionChannelTests.swift

@@ -579,6 +579,30 @@ class NIOTSConnectionChannelTests: XCTestCase {
         XCTAssertNoThrow(try conn.close().wait())
     }
 
+    func testCanObserveValueOfEnablePeerToPeer() throws {
+        let listener = try NIOTSListenerBootstrap(group: self.group)
+            .bind(host: "localhost", port: 0).wait()
+        defer {
+            XCTAssertNoThrow(try listener.close().wait())
+        }
+
+        let connectFuture = NIOTSConnectionBootstrap(group: self.group)
+            .channelInitializer { channel in
+                return channel.getOption(option: NIOTSChannelOptions.enablePeerToPeer).map { value in
+                    XCTAssertFalse(value)
+                }.then {
+                    channel.setOption(option: NIOTSChannelOptions.enablePeerToPeer, value: true)
+                }.then {
+                    channel.getOption(option: NIOTSChannelOptions.enablePeerToPeer)
+                }.map { value in
+                    XCTAssertTrue(value)
+                }
+            }.connect(to: listener.localAddress!)
+
+        let conn = try connectFuture.wait()
+        XCTAssertNoThrow(try conn.close().wait())
+    }
+
     func testCanSafelyInvokeActiveFromMultipleThreads() throws {
         // This test exists to trigger TSAN violations if we screw things up.
         let listener = try NIOTSListenerBootstrap(group: self.group)

Tests/NIOTransportServicesTests/NIOTSListenerChannelTests.swift

@@ -199,4 +199,23 @@ class NIOTSListenerChannelTests: XCTestCase {
             XCTFail("Unexpected error: \(error)")
         }
     }
+
+    func testCanObserveValueOfEnablePeerToPeer() throws {
+        let listener = try NIOTSListenerBootstrap(group: self.group)
+            .serverChannelInitializer { channel in
+                return channel.getOption(option: NIOTSChannelOptions.enablePeerToPeer).map { value in
+                    XCTAssertFalse(value)
+                }.then {
+                    channel.setOption(option: NIOTSChannelOptions.enablePeerToPeer, value: true)
+                }.then {
+                    channel.getOption(option: NIOTSChannelOptions.enablePeerToPeer)
+                }.map { value in
+                    XCTAssertTrue(value)
+                }
+            }
+            .bind(host: "localhost", port: 0).wait()
+        defer {
+            XCTAssertNoThrow(try listener.close().wait())
+        }
+    }
 }