code: public static <T> Observable<T> wrapColdWithAutoRelease(final Observable<T> source) {
          return Observable.create(new Observable.OnSubscribe<T>() {
              @Override
              public void call(final Subscriber<? super T> subscriber) {
                  source.subscribe(new Subscriber<T>() {
                      @Override
                      public void onCompleted() {
                          if (!subscriber.isUnsubscribed()) {
                              subscriber.onCompleted();
                          }
                      }
  
                      @Override
                      public void onError(Throwable e) {
                          if(!subscriber.isUnsubscribed()) {
                              subscriber.onError(e);
                          }
                      }
  
                      @Override
                      public void onNext(T t) {
                          if (!subscriber.isUnsubscribed()) {
                              subscriber.onNext(t);
                          } else {
                              ReferenceCountUtil.release(t);
                          }
                      }
                  });
              }
          });
      }
docs: Wrap an observable and free a reference counted item if unsubscribed in the meantime.
  
  This can and should be used if a hot observable is used as the source but it is not guaranteed that
  there will always be a subscriber that consumes the reference counted item. If an item is emitted
  by the source observable and no subscriber is attached (because it unsubscribed) the item will
  be freed.
  
  Note that also non reference counted items can be passed in, but there is no impact other than
  making it cold (in which case defer could be used).
  
  It is very important that if subscribed, the caller needs to release the reference counted item.
  It wil only be released on behalf of the caller when unsubscribed.
  
  @param source the source observable to wrap.
  @return the wrapped cold observable with refcnt release logic.