Pourquoi l'événement d'achèvement empêche-t-il le passage de ce test?

Question

Je me rends compte qu'il y a beaucoup de code ici pour une question SO, mais c'est le meilleur que je peux faire pour l'instant ... Vous pouvez simplement copier/coller le code dans un terrain de jeu rx pour voir le problème.

Sur la ligne 89, il y a un morceau de code let creds = Observable.just(credentials)//.concat(Observable.never()) commenté. Si je supprime le // et autorise la concat, le code passera son test. Quelqu'un peut-il donner une idée de la raison pour laquelle ce code échoue au test si creds est autorisé à envoyer un événement d'achèvement?

import Foundation 
import RxSwift 
import RxCocoa 
import UIKit 

typealias Credentials = (email: String, password: String) 

struct User { 
    let id: String 
    let properties: [Property] 
} 

struct Property { 
    let id: String 
    let name: String 
} 

struct LoginParams { 
    let touchIDPossible: Bool 
} 

class LoginScreen { 
    var attemptLogin: Observable<Credentials> { 
     assert(_attemptLogin == nil) 
     _attemptLogin = PublishSubject() 
     return _attemptLogin! 
    } 

    var _attemptLogin: PublishSubject<(email: String, password: String)>? 
} 

class DashboardScreen { 
    func display(property: Observable<Property?>) { 
     property.subscribe(onNext: { [unowned self] in 
      self._property = $0 
     }).disposed(by: bag) 
    } 

    var _property: Property? 
    let bag = DisposeBag() 
} 

class Interface { 
    func login(params: LoginParams) -> Observable<LoginScreen> { 
     assert(_login == nil) 
     _login = PublishSubject() 
     return _login! 
    } 

    func dashboard() -> Observable<DashboardScreen> { 
     assert(_dashboard == nil) 
     _dashboard = PublishSubject() 
     return _dashboard! 
    } 

    var _login: PublishSubject<LoginScreen>? 
    var _dashboard: PublishSubject<DashboardScreen>? 
    let bag = DisposeBag() 
} 

class Server { 
    func user(credentials: Credentials) -> Observable<User> { 
     assert(_user == nil) 
     _user = PublishSubject() 
     return _user! 
    } 

    func property(id: String) -> Observable<Property> { 
     assert(_property == nil) 
     _property = PublishSubject() 
     return _property! 
    } 

    var _user: PublishSubject<User>? 
    var _property: PublishSubject<Property>? 
} 

class Coordinator { 

    init(interface: Interface, server: Server) { 
     self.interface = interface 
     self.server = server 
    } 

    func start() { 
     let credentials = (email: "foo", password: "bar") 

     // remove the `//` and the test will pass. Why does it fail when `creds` completes? 
     let creds = Observable.just(credentials)//.concat(Observable.never()) 

     let autoUser = creds.flatMap { 
      self.server.user(credentials: $0) 
       .materialize() 
       .filter { !$0.isCompleted } 
      }.shareReplayLatestWhileConnected() 

     let login = autoUser.filter { $0.error != nil } 
      .flatMap { _ in self.interface.login(params: LoginParams(touchIDPossible: false)) } 

     let attempt = login.flatMap { $0.attemptLogin } 
      .shareReplayLatestWhileConnected() 

     let user = attempt.flatMap { 
      self.server.user(credentials: $0) 
       .materialize() 
       .filter { !$0.isCompleted } 
      }.shareReplayLatestWhileConnected() 

     let propertyID = Observable.merge(autoUser, user).map { $0.element } 
      .filter { $0 != nil }.map { $0! } 
      .map { $0.properties.sorted(by: { $0.name < $1.name }).map({ $0.id }).first } 

     let property = propertyID.filter { $0 != nil }.map { $0! } 
      .flatMap { self.server.property(id: $0) 
       .map { Optional.some($0) } 
       .catchErrorJustReturn(nil) 
      }.debug("property").shareReplayLatestWhileConnected() 

     let dashboard = property.flatMap { _ in self.interface.dashboard() } 

     dashboard.map { $0.display(property: property) } 
      .subscribe() 
      .disposed(by: bag) 
    } 

    let interface: Interface 
    let server: Server 
    let bag = DisposeBag() 
} 

do { 
    let interface = Interface() 
    let server = Server() 
    let coordinator = Coordinator(interface: interface, server: server) 

    coordinator.start() 

    assert(server._user != nil) 

    let simpleProperty = Property(id: "bar", name: "tampa") 
    let user = User(id: "foo", properties: [simpleProperty]) 
    server._user?.onNext(user) 
    server._user?.onCompleted() 
    server._user = nil 

    assert(interface._login == nil) 

    assert(server._property != nil) 

    let property = Property(id: "bar", name: "tampa") 
    server._property!.onNext(property) 
    server._property!.onCompleted() 
    server._property = nil 

    assert(interface._dashboard != nil) 

    let dashboard = DashboardScreen() 
    interface._dashboard?.onNext(dashboard) 
    interface._dashboard?.onCompleted() 

    assert(dashboard._property != nil) 
    print("test passed") 
} 

Voici la sortie du code tel qu'il est au-dessus:

2017-06-01 22:22:42.534: property -> subscribed 
2017-06-01 22:22:42.552: property -> Event next(Optional(__lldb_expr_134.Property(id: "bar", name: "tampa"))) 
2017-06-01 22:22:42.557: property -> Event completed 
2017-06-01 22:22:42.557: property -> isDisposed 
2017-06-01 22:22:42.559: property -> subscribed 
assertion failed: file MyPlayground.playground, line 159 

Pourquoi le property étant abonné à après il a été disposé?

Voici la sortie si vous supprimez le \\:

2017-06-01 22:23:51.540: property -> subscribed 
2017-06-01 22:23:51.553: property -> Event next(Optional(__lldb_expr_136.Property(id: "bar", name: "tampa"))) 
test passed 

Answer 1

Je suggère d'abord garder le dashboard autour d'un DisposeBag de sorte que lorsque start() finalise, la référence ne disparaît pas trop tôt. Le PO a depuis mis à jour le code, donc voici une tentative mise à jour pour une réponse.

---

Lorsque vous ajoutez plus d'informations de débogage:

let dashboard = property.debug("prop in") 
    .flatMap { _ in self.interface.dashboard().debug("dash in") } 
    .debug("dash out") 

Le journal révèle que la propriété se termine au début, à savoir juste après la séquence intérieure a été souscrit (« tiret -> souscrit »):

2017-06-03 08:33:27.442: property -> Event next(Optional(Property(id: "bar", name: "tampa"))) 
2017-06-03 08:33:27.442: prop in -> Event next(Optional(Property(id: "bar", name: "tampa"))) 
2017-06-03 08:33:27.449: dash in -> subscribed 
2017-06-03 08:33:27.452: property -> Event completed 
2017-06-03 08:33:27.452: property -> isDisposed 
2017-06-03 08:33:27.452: prop in -> Event completed 
2017-06-03 08:33:27.452: prop in -> isDisposed 
2017-06-03 08:33:27.456: dash in -> Event next(DashboardScreen) 
2017-06-03 08:33:27.456: dash out -> Event next(DashboardScreen) 

Si vous .concat(.never()), l'événement d'achèvement ne se déclenche pas et ne gêne pas le processus.

Le problème est que votre code de test est écrit impérativement. Vous start() le processus, puis publier les modifications. Mais le tout tombe en morceaux plus tôt si vous mettez les différents événements onNext dans la file d'attente principale de manière asynchrone. La conception de votre coordinateur se lit comme du code déclaratif, mais est vraiment utilisée comme un chemin de code séquentiel impératif. Un remède consiste à tenir compte de la rapidité d'exécution. PublishSubjects n'a pas d'antécédents; Si vous utilisez BehaviorSubjects qui rejoue leur dernière valeur à la place, vous pouvez configurer toutes les modifications avant en appelant start() et cela fonctionnera. Je suppose que vous utilisez PublishSubject s parce que vous appelez d'abord start() pour ouvrir le canal et que vous voulez y faire passer les modifications l'une après l'autre. Le problème est que votre pipe est faite d'une manière qui n'attend pas que vous poussiez tout. La vanne d'entrée se ferme de manière indépendante, pour ainsi dire.

Ouais, cette métaphore n'a pas été le meilleur dans toute l'histoire humaine de

Ainsi, les options sont vraiment:

1. Faire tout le travail du coordonnateur un grand Observable.combineLatest afin que toute la séquence de transformation ISN « t commencé jusqu'à ce que chaque séquence avait leur mot à dire,
2. utilisation tampon/rejouant sujets et les mettre en place à l'avance,
3. remplacer le .just (qui complète) avec une séquence de base qui n'a jamais complet es pour garder le pipeline ouvert; vous pouvez en faire un Observable<Observable<Credentials>> où la séquence externe reste vivante et les séquences internes utilisent Observable.just - bien que je doute que votre code de production dépende de ce petit détail du tout.