title: RxJS: Auto (dis)connect on (un)subscribe with Websockets and Stomp
body: <p>I'm building a litte RxJS Wrapper for Stomp over Websockets, which already works.</p>
  
  <p>But now I had the idea of a really cool feature, that may (hopefully - correct me if I'm wrong) be easily done using RxJS.</p>
  
  <h2>Current behavior:</h2>
  
  <pre><code>myStompWrapper.configure("/stomp_endpoint");
  myStompWrapper.connect();      // onSuccess: set state to CONNECTED
  
  // state (Observable) can be DISCONNECTED or CONNECTED
  var subscription = myStompWrapper.getState()
      .filter(state =&gt; state == "CONNECTED")
      .flatMap(myStompWrapper.subscribeDestination("/foo"))
      .subscribe(msg =&gt; console.log(msg));
  
  // ... and some time later:
  subscription.unsubscribe();    // calls 'unsubscribe' for this stomp destination
  myStompWrapper.disconnect();   // disconnects the stomp websocket connection
  </code></pre>
  
  <p>As you can see, I <strong>must</strong> wait for <code>state == "CONNECTED"</code> in order to subscribe to <code>subscribeDestination(..)</code>. Else I'd get an Error from the Stomp Library.</p>
  
  <h2>The new behavior:</h2>
  
  <p>The next implementation should make things easier for the user. Here's what I imagine:</p>
  
  <pre><code>myStompWrapper.configure("/stomp_endpoint");
  
  var subscription = myStompWrapper.subscribeDestination("/foo")
      .subscribe(msg =&gt; console.log(msg));
  
  // ... and some time later:
  subscription.unsubscribe();
  </code></pre>
  
  <p><strong>How it should work internally:</strong></p>
  
  <ol>
  <li><code>configure</code> can only be called while <code>DISCONNECTED</code></li>
  <li>when <code>subscribeDestination</code> is called, there are 2 possibilities:
  
  <ol>
  <li>if <code>CONNECTED</code>: just subscribe to the destination</li>
  <li>if <code>DISCONNECTED</code>: first call <code>connect()</code>, then subscribe to the destination</li>
  </ol></li>
  <li>when <code>unsubscribe</code> is called, there are 2 possibilities:
  
  <ol>
  <li>if this was the last subscription: call <code>disconnect()</code></li>
  <li>if this wasn't the last subscription: do nothing</li>
  </ol></li>
  </ol>
  
  <p>I'm not yet sure how to get there, but that's why I ask this question here <code>;-)</code></p>
  
  <p>Thanks in advance!</p>
  
  <h1>EDIT: more code, examples and explanations</h1>
  
  <p>When <strong>configure()</strong> is called while <strong>not</strong> disconnected it should throw an <code>Error</code>. But that's not a big deal.</p>
  
  <p><strong>stompClient.connect(..)</strong> is non-blocking. It has an <code>onSuccess</code> callback:</p>
  
  <pre><code>public connect() {
    stompClient.connect({}, this.onSuccess, this.errorHandler);
  }
  
  public onSuccess = () =&gt; {
    this.state.next(State.CONNECTED);
  }
  </code></pre>
  
  <p><strong>observeDestination(..)</strong> subscribes to a Stomp Message Channel (= destination) and returns an <code>Rx.Observable</code> which then can be used to unsubscribe from this Stomp Message Channel:</p>
  
  <pre><code>public observeDestination(destination: string) {
    return this.state
        .filter(state =&gt; state == State.CONNECTED)
        .flatMap(_ =&gt; Rx.Observable.create(observer =&gt; {
          let stompSubscription = this.client.subscribe(
              destination,
              message =&gt; observer.next(message),
              {}
          );
  
          return () =&gt; {
            stompSubscription.unsubscribe();
          }
        }));
  }
  </code></pre>
  
  <p><strong>It can be used like this:</strong></p>
  
  <pre><code>myStompWrapper.configure("/stomp_endpoint");
  myStompWrapper.connect();
  
  myStompWrapper.observeDestination("/foo")
      .subscribe(..);
  
  myStompWrapper.observeDestination("/bar")
      .subscribe(..);
  </code></pre>
  
  <p><strong>Now I'd like to get rid of <code>myStompWrapper.connect()</code></strong>. The code should automatically call <code>this.connect()</code> when the first one subscribes by calling <code>observeDestination(..).subscribe(..)</code> and it should call <code>this.disconnect()</code> when the last one called <code>unsubscribe()</code>.</p>
  
  <p><strong>Example:</strong></p>
  
  <pre><code>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
  </code></pre>