LongPolling With rxSwift

Question

我正在研究 rxSwift，我想为 c longpolling 服务器与模仿永久连接的该服务进行交互服务。我写了它，但在我看来，这个决定不是可以做得更好吗？是否可以以某种方式重复可观察，无论错误如何，并且取决于长轮服务器响应。

任何人都可以分享解决方案吗？或者帮助提供建议？如何更好地组织？我希望看到更好的解决方案，因为才开始研究 rxswift

class LongPollingService {
public var messageReciver: PublishSubject<EventProtocol> = PublishSubject<EventProtocol>()
private let transport = DefaultTransport()
private let disposeBag = DisposeBag()
private var currentRequestInfo = Variable<LongpollingServerInfo?>(nil)
private var currentRequestDisposable: Disposable?
private var currentLongpollingConnection: Disposable? // Subsribee for request server info
private var eventListener : Disposable?
private var currentReqursiveConnection: Disposable? // Subscriber for event listener from longpoll server
func startObservableEvents() {
getServerConnection()
subscribeServerInfo()
//testing listen events
eventListener = messageReciver.showMessagesInDebugMode().subscribe()
eventListener?.addDisposableTo(disposeBag)
}
func disconnect() {
currentRequestDisposable?.dispose()
currentLongpollingConnection?.dispose()
currentReqursiveConnection?.dispose()
}
private func subscribeServerInfo() {
currentLongpollingConnection = currentRequestInfo
.asObservable()
.filter({$0 != nil})
.subscribe(onNext: { [weak self] (info) in
guard let sSelf = self else { return }
sSelf.subscribeToEvents(timeStamp: info!.ts)
})
currentLongpollingConnection?.addDisposableTo(disposeBag)
}
private func subscribeToEvents(timeStamp: TimeInterval) {
if let serverInfo = currentRequestInfo.value {
currentReqursiveConnection?.dispose()
currentReqursiveConnection = getEventsFromLongpollServer(serverInfo: serverInfo, with: timeStamp)
.flatMap(parseUpdates)
.flatMap(reciveEvents)
.showErrorsSwiftMessagesInDebugMode()
.subscribe(onNext: { [weak self] updates in
guard let sSelf = self else { return }
sSelf.subscribeToEvents(timeStamp: updates)
},
onError: { [weak self] error in
guard let sSelf = self else { return }
if let error = error as? LongPollError {
switch error {
case .olderHistory(let ts): sSelf.subscribeToEvents(timeStamp: ts)
default: sSelf.getServerConnection()
}
}
})
currentReqursiveConnection?.addDisposableTo(disposeBag)
}
}
private func getServerConnection() {
//get longpolling server info for connection.
currentRequestDisposable = getLongpollServerInfo()
.subscribe(onNext: {[weak self] info in
guard let sSelf = self else { return }
sSelf.currentRequestInfo.value = info
})
currentRequestDisposable?.addDisposableTo(disposeBag)
}
private func parseUpdates(json: Any) throws -> Observable<LongPollingUpdates> {
let response = try Mapper<LongPollingUpdates>().map(JSONObject: json)
return .just(response)
}
private func reciveEvents(updates:LongPollingUpdates) throws -> Observable<TimeInterval> {
if let errors = updates.failed {
throw parseErrors(errors: errors)
}
if let events = updates.updates {
parseUpdates(updates: events)
}
return Observable.just(updates.timeStamp!)
}
private func parseUpdates(updates: [[Any]]) {
updates.forEach { (array) in
let firstElementInUpdate = array.first
if let update = firstElementInUpdate as? Int {
switch update {
case 1: break
case 2: break
case 3: break
case 4: messageReciver.onNext(NewMessage(array: array))
default: break
}
}
}
}
private func parseErrors(errors: [String: Any]) -> LongPollError {
if let error = errors["failed"] as? Int {
switch error {
case 1:
guard let ts = errors["ts"] as? TimeInterval else { return .unkownError }
return .olderHistory(ts: ts)
case 2: return .needNewkey
case 3: return .needCaseAndTs
case 4: return .unkownVersion
default:
return .unkownError
}
}
return .unkownError
}
private func getEventsFromLongpollServer(serverInfo: LongpollingServerInfo, with ts: TimeInterval) -> Observable<Any> {
let url = buildLongPollingServerRoute(from: serverInfo, with: ts)
let request = buldLongPollRequst(route: url)
let requestConvert = try? URLEncoding.default.encode(request!, with: nil)
return transport.makeRequest(request: requestConvert!)
}
private func getEventsFromLongpollServer(serverInfo: LongpollingServerInfo) -> Observable<Any> {
let url = buildLongPollingServerRoute(from: serverInfo)
let request = buldLongPollRequst(route: url)
let requestConvert = try? URLEncoding.default.encode(request!, with: nil)
return transport.makeRequest(request: requestConvert!)
}
private func getLongpollServerInfo() -> Observable<LongpollingServerInfo> {
let request = MessageRouter.getLongpollServer(useSsl: false, needPts: false)
return transport.makeModel(request: request)
}
}

Answer 1

所以假设你有一个这样的函数：

func getData() -> Observable<Data>

并且您想在特定period进行长时间轮询，您可以执行以下操作：

Observable<Int>.interval(period, scheduler: MainScheduler.instance)
.map { _ in return }
.flatMap(getData)
.subscribe( /* ... handle data ... */)
.disposed(by: disposeBag)

如果更合适，您可以使用除MainScheduler以外的其他计划程序。

现在，如果您还想处理getData可能发出的Error并且您不希望它必然取消订阅长轮询，那么您可以这样做：

func handleError(error: Error) -> Observable<Data> {
return Observable.empty()
}
Observable<Int>.interval(period, scheduler: MainScheduler.instance)
.map { _ in return }
.flatMap { return getData.catchError(handleError) }
.subscribe( /* ... handle data ... */)
.disposed(by: disposeBag)

您还可以分析handleError中的错误，并决定是通过发出空Observable来继续，还是通过发出另一个错误来取消长轮询。