Eager unsubscribe for Observable.using?

[davidmoten]

When using Observable.using to read from a file (a synchronous source) I want to complete the process by writing back to the same file. Because using doesn't dispose its resource till the downstream is complete I can't perform my secondary action (unless I block or introduce asynchronicity). I'm wondering about creating an Operator say OperatorUnsubscribeEagerly that could be chained with the using command to enable my use case.

For example:

Observable.using(...).toUnsubscribeEagerly().concatWith(...)...

An alternative would be an overload on using with an additional boolean parameter unsubscribeEagerly.

What do people think? Perhaps I'm overlooking an existing idiom?

[akarnokd]

If you have access to the resource outside the factory function, you could call doOnCompleted(() -> resource.close());

[davidmoten]

Not really. There's no mention of the resource outside of using, and that
is what makes using so handy after all.
On 23 Jan 2015 19:41, "David Karnok" notifications@github.com wrote:

If you have access to the resource outside the factory function, you could
call doOnCompleted(() -> resource.close());

—
Reply to this email directly or view it on GitHub
#2492 (comment).

[davidmoten]

Actually I think you are on to it. I'll just adjust the resource factory
function so it uses doOnCompleted before returning its observable.
On 23 Jan 2015 20:19, "Dave Moten" davidmoten@gmail.com wrote:

Not really. There's no mention of the resource outside of using, and that
is what makes using so handy after all.
On 23 Jan 2015 19:41, "David Karnok" notifications@github.com wrote:

If you have access to the resource outside the factory function, you
could call doOnCompleted(() -> resource.close());

—
Reply to this email directly or view it on GitHub
#2492 (comment).

[davidmoten]

After looking at the options I think the best way is to use OperatorUnsubscribeEagerly.

If one is able to adjust the resource factory function (and it won't always be the case when using a third party method) then using doOnCompleted is not great because as a general technique a volatile bit of state needs to be maintained in the resource to ensure that resource disposal does not happen twice. For something like an InputStream I don't care but other resources may not be so friendly.

OperatorUnsubscribeEagerly is about as simple as it gets:

public final class OperatorUnsubscribeEagerly<T> implements Operator<T, T> {

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> child) {
        Subscriber<T> parent = new Subscriber<T>() {

            @Override
            public void onCompleted() {
                unsubscribe();
                child.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                unsubscribe();
                child.onError(e);
            }

            @Override
            public void onNext(T t) {
                child.onNext(t);
            }

        };
        child.add(parent);
        return parent;
    }

Is this a reasonable addition to the Observable API? Might be used like this:

Observable.using(...).toUnsubscribeEagerly()

[davidmoten]

Observable.using overload proposed in #2759

[akarnokd]

Closed via PR #2759