Adds observer api for closing connection event (#68)

Co-authored-by: Niclas Kristek <[email protected]>

Author: ayush-yadav001

Sources/RSocketReactiveSwift/Client/ReactiveSwiftClient.swift

@@ -46,6 +46,20 @@ public struct ReactiveSwiftClient: Client {
     internal var isDisposed: Bool {
         return !coreClient.channel.isActive
     }
+    /// This methods helps to get call back whenever connection is closed
+    /// - Returns: SignalProducer<Void, Swift.Error> to represent task result
+    public var shutdownProducer: SignalProducer<Void, Swift.Error> {
+        SignalProducer { observer, _ in
+            coreClient.channel.closeFuture.whenComplete { result in
+                switch result {
+                case .success:
+                    observer.sendCompleted()
+                case .failure(let error):
+                    observer.send(error: error)
+                }
+            }
+        }
+    }
 }
 
 extension ClientBootstrap where Client == CoreClient, Responder == RSocketCore.RSocket  {

Tests/RSocketReactiveSwiftTests/RSocketReactiveSwiftTests.swift

@@ -404,4 +404,46 @@ final class RSocketReactiveSwiftTests: XCTestCase {
         // checking if connection is inactive
         XCTAssertTrue(client.isDisposed)
     }
+    func testConnectionDisposeListener() {
+        // Creating expectation
+        let didReceiveConnectionclosedEvent = expectation(description: "did receive Connection closed  event ")
+        let serverResponder: RSocketReactiveSwift.ResponderRSocket? = TestRSocket()
+        let clientResponder: RSocketReactiveSwift.ResponderRSocket? = TestRSocket()
+        let (serverMultiplexer, clientMultiplexer) = TestDemultiplexer.pipe(
+            serverResponder: serverResponder.map { ResponderAdapter(responder: $0, encoding: .default) },
+            clientResponder: clientResponder.map { ResponderAdapter(responder: $0, encoding: .default) }
+        )
+        // Channel creation
+        let serverChannel = EmbeddedChannel()
+        // Making channel Active
+        XCTAssertNoThrow(try serverChannel.connect(to: SocketAddress.init(ipAddress: "127.0.0.1", port: 0)).wait())
+        let clientChannel = EmbeddedChannel()
+        XCTAssertNoThrow(try clientChannel.connect(to: SocketAddress.init(ipAddress: "127.0.0.1", port: 0)).wait())
+        // Creating Reactive swift client
+        let server =  ReactiveSwiftClient(CoreClient(requester: serverMultiplexer.requester, channel: serverChannel))
+        let client =  ReactiveSwiftClient(CoreClient(requester: clientMultiplexer.requester, channel: clientChannel))
+        defer {
+            // closing channel connection
+            XCTAssertNoThrow(try serverChannel.finish())
+        }
+        XCTAssertNotNil(server)
+        XCTAssertNotNil(client)
+        // client connection closed event signal producer
+        client.shutdownProducer.startWithSignal({ signal, interruptHandle in
+            signal.flatMapError({ error -> Signal<Void, Error> in
+                XCTFail("\(error)")
+                return .empty
+            }).materialize().collect().observeValues { values in
+                didReceiveConnectionclosedEvent.fulfill()
+            }
+        })
+        // checking if connection is active
+        XCTAssertFalse(client.isDisposed)
+        // closing connection using dispose method
+        XCTAssertNoThrow(try clientChannel.finish())
+        // checking if connection is inactive
+        XCTAssertTrue(client.isDisposed)
+        self.wait(for: [didReceiveConnectionclosedEvent], timeout: 0.1)
+    }
 }
+