RxJS：自动（断开连接（使用 Websockets 和 Stomp 取消订阅）

Question

我正在为 Stomp over Websockets 构建一个小型 RxJS 包装器，它已经可以工作了。

但现在我有了一个非常酷的功能的想法，它可能（希望 - 如果我错了，请纠正我）使用 RxJS 很容易完成。

当前行为：

myStompWrapper.configure("/stomp_endpoint");
myStompWrapper.connect();      // onSuccess: set state to CONNECTED

// state (Observable) can be DISCONNECTED or CONNECTED
var subscription = myStompWrapper.getState()
    .filter(state => state == "CONNECTED")
    .flatMap(myStompWrapper.subscribeDestination("/foo"))
    .subscribe(msg => console.log(msg));

// ... and some time later:
subscription.unsubscribe();    // calls 'unsubscribe' for this stomp destination
myStompWrapper.disconnect();   // disconnects the stomp websocket connection

如您所见，我必须等待state == "CONNECTED" 才能订阅subscribeDestination(..)。否则我会从 Stomp 库中得到一个错误。

新行为：

下一个实现应该让用户更轻松。这是我的想象：

myStompWrapper.configure("/stomp_endpoint");

var subscription = myStompWrapper.subscribeDestination("/foo")
    .subscribe(msg => console.log(msg));

// ... and some time later:
subscription.unsubscribe();

它应该如何在内部工作：

1. configure 只能在 DISCONNECTED 时调用
2. 当subscribeDestination被调用时，有两种可能：
   1. 如果CONNECTED：只需订阅目的地
   2. 如果DISCONNECTED：先调用connect()，然后订阅目的地
3. 当unsubscribe被调用时，有两种可能：
   1. 如果这是最后一次订阅：致电disconnect()
   2. 如果这不是最后一次订阅：什么都不做

我还不知道怎么去那里，但这就是为什么我在这里问这个问题;-)

提前致谢！

编辑：更多代码、示例和解释

当 configure() 被调用而 not 断开连接时，它应该抛出一个Error。但这没什么大不了的。

stompClient.connect(..) 是非阻塞的。它有一个onSuccess 回调：

public connect() {
  stompClient.connect({}, this.onSuccess, this.errorHandler);
}

public onSuccess = () => {
  this.state.next(State.CONNECTED);
}

observeDestination(..) 订阅一个 Stomp 消息频道（= 目的地）并返回一个 Rx.Observable，然后可以使用该 Rx.Observable 取消订阅此 Stomp 消息频道：

public observeDestination(destination: string) {
  return this.state
      .filter(state => state == State.CONNECTED)
      .flatMap(_ => Rx.Observable.create(observer => {
        let stompSubscription = this.client.subscribe(
            destination,
            message => observer.next(message),
            {}
        );

        return () => {
          stompSubscription.unsubscribe();
        }
      }));
}

可以这样使用：

myStompWrapper.configure("/stomp_endpoint");
myStompWrapper.connect();

myStompWrapper.observeDestination("/foo")
    .subscribe(..);

myStompWrapper.observeDestination("/bar")
    .subscribe(..);

现在我想摆脱myStompWrapper.connect()。当第一个通过调用observeDestination(..).subscribe(..) 订阅时，代码应该自动调用this.connect()，当最后一个调用unsubscribe() 时，它应该调用this.disconnect()。

示例：

myStompWrapper.configure("/stomp_endpoint");

let subscription1 = myStompWrapper.observeDestination("/foo")
    .subscribe(..); // execute connect(), because this
                    // is the first subscription

let subscription2 = myStompWrapper.observeDestination("/bar")
    .subscribe(..);

subscription2.unsubscribe();
subscription1.unsubscribe(); // execute disconnect(), because this 
                             // was the last subscription

Answer 1

RxJS: Auto (dis)connect on (un)subscribe with Websockets and Stomp

我同意你建议隐藏在 myStompWrapper 中的代码在它的新家中会更快乐。

我仍然建议使用observeDestination 这样的名称而不是subscribeDestination("/foo")，因为您实际上并没有订阅该方法，而只是完成了您的可观察链。

1. configure() 只能在 DISCONNECTED 时调用

   您没有在这里指定如果在不是DISCONNECTED 时调用它会发生什么。由于您似乎没有在此处返回任何您将使用的值，因此我假设您打算在异常状态不方便时抛出异常。为了跟踪这些状态，我将使用以DISCONNECTED 的初始值开头的BehaviourSubject。您可能希望将状态保持在 observeDestination 内，以决定是否抛出异常

2. 如果已连接：只需订阅目的地

   如果 DISCONNECTED：首先调用 connect()，然后订阅目的地

   正如我之前提到的，如果订阅不在subscribeDestination("/foo") 内发生，我认为你会更开心，而是你只是建立你的可观察链。由于您在某些情况下只是想调用 connect()，因此我只需在包含状态条件的可观察链中使用 .do() 调用。

1234563这样，每个新订阅者都不会重新创建新订阅，而是 .refCount() 将对 observable 链进行一次订阅，一旦下游没有更多订阅者，unsubscribe() 将进行一次订阅。

假设消息以 this.observedData$ 的形式进入 myStompWrappermyStompWrapper，我建议的代码如下所示：

observeDestination() {
  return Rx.Observable.create(function (observer) {
     var subscription = this.getState()
             .filter(state => state == "CONNECTED")
             .do(state => state ? this.connect() : Observable.of(true))
             .switchMap(this.observedData$)
             .refCount();
             .subscribe(value => {
               try {
                 subscriber.next(someCallback(value));
               } catch(err) {
                 subscriber.error(err);
               }
             },
             err => subscriber.error(err),
             () => subscriber.complete());

 return { unsubscribe() { this.disconnect(); subscription.unsubscribe(); } };
}

因为我遗漏了您的一些代码，所以我允许自己不测试我的代码。但希望它能说明并呈现我在回答中提到的概念。