如何处置时调用的方法

Question

我在RxSwift这段代码，它包装UDPBroadcastConnection（https://github.com/gunterhager/UDPBroadcastConnection）我的问题是，我有什么做观察的处置时调用closeConnection（）？

struct UDPBroadcastResponse {
    let ip: String
    let port: Int
    let message: [UInt8]
}

final class RxBroadcaster: UDPBroadcastConnection {
    let response = PublishSubject<UDPBroadcastResponse>()
    private let disposeBag = DisposeBag()
}

extension RxBroadcaster: ReactiveCompatible {}

extension Reactive where Base: RxBroadcaster {
    func send(message: String) -> Observable<UDPBroadcastResponse> {
        DDLogVerbose("Sending UDP broadcast message: \(message)")
        self.base.sendBroadcast(message)
        return self.base.response
    }
}

extension RxBroadcaster {
    convenience init(port: UInt16) {
        let rely = ReplaySubject<UDPBroadcastResponse>.createUnbounded()
        self.init(port: port) { ip, port, message in
            let response = UDPBroadcastResponse(ip: ip, port: port, message: message)
            // DDLogVerbose("Receiving UDP message: IP: \(response.ip). Port: \(response.port). Message: \(response.message.map{ String(format: "%02X", $0) }.joined())")
            DDLogVerbose("Receiving UDP message: IP: \(response.ip). Port: \(response.port)")
            rely.onNext(response)
        }
        rely.bind(to: self.response).disposed(by: disposeBag)
    }
}

在这里，我看到一个例子：http://adamborek.com/practical-introduction-rxswift/但我不知道涉及到的对象的生命周期。你能帮助我吗？感谢您的建议。

Answer 1

您可以使用do操作员从订阅钩到生命周期事件，并执行副作用。您订阅或绑定之前，你可以在任何地方调用do可观察链。例：

rely.do(onDispose: {
    // called as your subscription is being disposed
}).bind(to: self.response).disposed(by: disposeBag)

Answer 2

我会更倾向于实现这样的RxBroadcaster：

final class RxBroadcaster {
    let response: Observable<UDPBroadcastResponse>

    init(port: UInt16) {
        let _response = PublishSubject<UDPBroadcastResponse>()
        let _connection = UDPBroadcastConnection(port: port, handler: { ipAddress, port, response in
            _response.onNext(UDPBroadcastResponse(ip: ipAddress, port: port, message: response))
        })

        response = _response.asObservable()
        completion = _response.asObserver()
        connection = _connection
    }

    deinit {
        connection.closeConnection()
        completion.onCompleted()
    }

    private let connection: UDPBroadcastConnection
    private let completion: AnyObserver<UDPBroadcastResponse>
}

extension RxBroadcaster: ObserverType {
    public func on(_ event: RxSwift.Event<Data>) {
        guard case let .next(data) = event else { return }
        connection.sendBroadcast(data)
    }
}

现在，您可以只绑定到RxBroadcaster myData.bind(to: myBroadcaster)发送数据，并通过订阅广播的响应接收数据。该连接将在转播超出范围自然亲密。

这感觉有点奇怪我做一个呼叫/响应运营商像你这样。 UDP套接字的一点是，你将获得独立的你送的那些的消息。