Как отписаться от вызова RXjava в широковещательном приемнике

Я использую RXJava2 для отправки электронной почты в широковещательном приемнике, и я хотел бы знать, когда я должен отказаться от подписки на событие. Код в основном:

        getSmsMmsObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> Timber.e(throwable, "Error sending mail."))
            .map(smsMmsAddress1 -> {
                smsMmsAddress = smsMmsAddress1;
                return doInBackgroundSendEmail();
            })
            .map(stringSingle -> {
                mMsgResponse = stringSingle;
                this.done = true;
                return deleteFile();
            })
            .subscribe(success -> {
                if (success) {
                    Toast.makeText(context, "Message Status: " + mMsgResponse, Toast.LENGTH_LONG).show();
                }

            });

Когда мне отписаться (в приемнике нет onPause или onDestroy) и как я узнаю, что onReceive завершен? Получатель регистрируется через манифест. Я подумал о том, чтобы сделать составную наблюдаемую, а затем избавиться от нее в разделе subscribe().

    compositeDisposable.add(
    getSmsMmsObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> Timber.e(throwable, "Error sending mail."))
            .map(smsMmsAddress1 -> {
                smsMmsAddress = smsMmsAddress1;
                return doInBackgroundSendEmail();
            })
            .map(stringSingle -> {
                mMsgResponse = stringSingle;
                this.done = true;
                return deleteFile();
            })
            .subscribe(success -> {
                if (success) {
                    Toast.makeText(context, "Message Status: " + mMsgResponse, Toast.LENGTH_LONG).show();
                }
                compositeDisposable.dispose();

            })
    );