RxJava - ReplaySubject only emitting data twice

Question

I am new to ReactiveX and I have a case where I want my observable to emit data to a late subscriber(whenever the observer subscribes, observable should emit the same data that it emitted previously). I made this Observable class that provide ReplaySubject's same instance to all observers (it is singleton class).

public class AccountsObservable {
    private static ConnectableObservable<String> hotObservable;
    private static AccountsObservable accountsObservable;


    public static AccountsObservable getObject() {
        if (accountsObservable == null) {
            accountsObservable = new AccountsObservable();
        }
        return accountsObservable;
    }

    public ConnectableObservable<String> getObservable() {
        if (hotObservable == null) {
            Observable<String> observable = ReplaySubject.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    emitter.onNext("XYZ");
                    emitter.onComplete();

                }
            });
            hotObservable = observable.replay();//publish
        }
        return hotObservable;
    }
}

Similarly, this is the observer class that creates new observer instance.

public class AccountsObserver {
    AccountsFetchListener listener;

    public AccountsObserver(AccountsFetchListener listener) {
        this.listener = listener;
    }

    public Observer<String> getObserver() {
        return new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String accounts) {
                listener.onSuccess(accounts);
            }

            @Override
            public void onError(Throwable e) {
                listener.onFailure();
            }

            @Override
            public void onComplete() {

            }
        };

    }

    public interface AccountsFetchListener {
        void onSuccess(String accounts);

        void onFailure();
    }
}

Here is the function where I test these observables

private void testObs() {
    ConnectableObservable<String> observable = AccountsObservable.getObject().getObservable();
    Observer<String> observer = new AccountsObserver(new AccountsObserver.AccountsFetchListener() {
        @Override
        public void onSuccess(String accounts) {
            Log.e("DATA -> ", accounts);
        }

        @Override
        public void onFailure() {
        }
    }).getObserver();
    observable.subscribe(observer);
    observable.connect();

}

I called this function "testObs()" 5 times but it emitted data only 2 times. The problem seems to be in AccountsObservable class where I provide ReplaySUbject's instance. Thanks

Answer 1

Your code runs fine as it is, your logs are being suppressed in logcat as per this:

We declared an application as too chatty once it logs more than 5 lines a second. Please file a bug against the application's owner that is producing this developer-verbose-debug-level class logging spam. The logs are 256KB, that means the application is creating a DOS attack and shortening the logs timepan to 6 seconds(!) making it useless for all others.

You can avoid this behaviour by whitelisting your app for logcat:

adb logcat -P '<pid or uid of your app>'