code: private void maybeOpenAndSend(final CouchbaseRequest request) {
          LOGGER.debug(logIdent(hostname, PooledService.this)
                  + "Need to open a new Endpoint (current size {})", endpoints.size());
  
          final Endpoint endpoint = endpointFactory.create(
              hostname, bucket, username, password, port, ctx
          );
  
          synchronized (epMutex) {
              endpoints.add(endpoint);
              endpointStates.register(endpoint, endpoint);
          }
  
          final Subscription subscription = whenState(endpoint, LifecycleState.CONNECTED,
              new Action1<LifecycleState>() {
                  @Override
                  public void call(LifecycleState lifecycleState) {
                      if (disconnect) {
                          RetryHelper.retryOrCancel(env, request, responseBuffer);
                      } else {
                          endpoint.send(request);
                          endpoint.send(SignalFlush.INSTANCE);
                      }
                  }
              }
          );
  
          endpoint.connect().subscribe(new Subscriber<LifecycleState>() {
              @Override
              public void onCompleted() {
                  // ignored on purpose
              }
  
              @Override
              public void onError(Throwable e) {
                  unsubscribeAndRetry(subscription, request);
              }
  
              @Override
              public void onNext(LifecycleState state) {
                  if (state == LifecycleState.DISCONNECTING || state == LifecycleState.DISCONNECTED) {
                      unsubscribeAndRetry(subscription, request);
                  }
              }
          });
      }
docs: Helper method to try and open new endpoints as needed and correctly integrate
  them into the state of the service.