How to use Rx to manage multiple observers to keep only one open connection to the service?

Question

In my application, some actions receive information from the server, and I need to make sure that the connection is already established before trying to retrieve the data. I use BehaviorSubject to notify subscribers when a connection is established, so Activity can load data.

The problem is that more than one action in the same thread has the same behavior. In our API, we should call connect()in onStart()and disconnect()in onStop, but if the user moves to another action that also uses the connection, there is no need to recreate the connection, we could use it.

I am currently implementing the following:

• When you call connect(), it returns BehaviorSubjectwhich will be signed from the calling class

• In the method, disconnect()it is actually disabled only if BehaviorSubjectthere are no observers indicating that no actions are waiting for an answer.

• The calling class must remove the Observable before the call disconnect(), otherwise the method hasObservers()will never returnfalse

  @CheckResult
  @Override
  public BehaviorSubject<Boolean> connect() {
     if (!connectionManager.isConnected()) {
          connectionManager.connect(TIMEOUT);
     }
  
     return mSubject;
  }
  
  @Override
  public void disconnect() {
      if (connectionManager.isConnected() && !mSubject.hasObservers()){
          connectionManager.disconnect();
      }
  }
  

In my previous implementation, I used listeners to achieve this. Each time it is called connect(), it must accept the listener as a parameter, which will be added to the List of listeners, and will be notified later one by one when the connection is established.

, disconnect(), , . connectionManager.disconnect() , , , .

?

Answer 1

, , , - / :

subject.doOnUnsubscribe(()-> {
  if(!subject.hasObservers()) {
    closeConnection();
  }
});
subject.doOnSubscribe(() -> {
  openConnectionIfNotOpen();
});

, disconnect(),