Mon application dispose effectivement d'une file d'opérations BLE à effectuer avec mon périphérique. Chaque opération commence par établir une connexion avec le périphérique, ce qui renvoie Observable<RxBleConnection>. Le premier élément de la file d'attente initie la connexion et les opérations suivantes reçoivent simplement ceci (partagé) RxBleConnection.RxBleConnection avec PublishSubject ne libère pas la connexion
En forme simpliee, la file d'attente est exécutée via:
Observable.concatDelayError(queuedOperations)
Si la connexion ne peut être établie, ou il est tombé au cours d'une opération, les opérations en attente restantes chaque nouvelle tentative, qui tentent de rétablir la connexion.
J'ai décidé de modifier le comportement de telle sorte qu'une fois que la connexion est devenue invalide, les opérations en file d'attente devraient toujours recevoir le RxBleConnection maintenant invalide plutôt que de rétablir une nouvelle connexion. La logique de nouvelle tentative est toujours exécutée mais échoue immédiatement dans ces cas. Il existe d'autres raisons pour lesquelles l'opération peut échouer, ce qui n'a rien à voir avec la connexion.
Pour produire ce comportement, je compose un PublishSubject directement après le Observable<RxBleConnection>. Ce sujet délègue simplement l'original RxBleConnection --- voir le code ci-dessous. Si la connexion atteint un état d'erreur, les abonnements suivants au sujet recevront l'erreur; sinon, la connexion partagée est renvoyée. C'est exactement le comportement que je voulais, et il semble fonctionner comme prévu lorsqu'une erreur se produit; Cependant, j'ai des problèmes maintenant quand tout réussit.
Avant la modification, une fois toutes les opérations de la file d'attente consommées, la connexion était automatiquement relâchée. Avec l'ajout du PublishSubject, cependant, les opérations sont réussies, mais la connexion reste ouverte. En utilisant les instructions de débogage, j'ai vérifié que onUnsubscribe et onTerminate du sujet ne sont jamais appelés. L'original RxBleConnection expire finalement --- et ses onUnsubscribe et onTerminate sont appelés.
Je me demandais ce que je fais de façon incorrecte qui provoque la connexion de l'application au périphérique.
private Observable.Transformer<RxBleConnection, RxBleConnection> createConnectionSubject() {
return rxBleConnectionObservable -> {
final PublishSubject<RxBleConnection> subject = PublishSubject.create();
rxBleConnectionObservable.subscribe(
subject::onNext,
subject::onError,
subject::onCompleted);
return subject;
};
}