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

i did call this method five times in onCreate() for example but the ' Log.e("DATA -> ", accounts); ' statement only ran twice — Daniyal Javaid, Jun 30 '19 at 08:45
yeah that worked, thanks.That printed 5 times. I thought there was a code issue — Daniyal Javaid, Jun 30 '19 at 09:03
this was a test example, the same classes were not emitting data for the real project which I used in, will recheck them and let you know. — Daniyal Javaid, Jun 30 '19 at 09:06
Possible duplicate of [android logcat logs chatty module line expire message](https://stackoverflow.com/questions/34587273/android-logcat-logs-chatty-module-line-expire-message) — Gustavo Pagani, Jun 30 '19 at 09:07

score 1 · Accepted Answer · answered Jun 30 '19 at 09:10

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>'

RxJava - ReplaySubject only emitting data twice

1 Answers1