db.select(sql).dependsOnOperator().getXXX() doesn't seem to work

[Crystark]

Hi,

I have the following piece of code that doesn't seem to work as I want:

updateObservable = Observable.interval(0, 10, TimeUnit.SECONDS)
    .doOnNext(s -> System.out.println("try " + s))
    .lift(db
        .select("SELECT id FROM mytable ORDER BY updated_at ASC")
        .dependsOnOperator()
        .getTupleN()
    )
    .doOnNext(s -> System.out.println("result " + s));

updateObservable.subscribe();

This prints out:

try 0
try 1
...

Although this seems to work fine:

updateObservable = Observable.interval(0, 10, TimeUnit.SECONDS)
    .doOnNext(s -> System.out.println("try " + s))
    .lift(db
        .select("SELECT id FROM mytable WHERE id <> ? ORDER BY updated_at ASC")
        .parameterOperator()
        .getTupleN()
    )
    .doOnNext(s -> System.out.println("result " + s));
updateObservable.subscribe();

Because it prints the following:

try 0
result TupleN [values=[18]]
result TupleN [values=[16]]
result TupleN [values=[17]]
try 1
result TupleN [values=[18]]
result TupleN [values=[16]]
result TupleN [values=[17]]
...

Am I doing something wrong in the first example ?

FYI I've simplified the above example but what I really want to do is to continuously poll that request and process its elements. It would probably look like that:

updateObservable = Observable.interval(0, 1, TimeUnit.NANOSECONDS)
    .onBackpressureLatest()
    .lift(db
        .select("SELECT id FROM mytable ORDER BY updated_at ASC")
        .dependsOnOperator()
        .getTupleN()
    )
    .<List<String>> compose(new MyTableTransformer())
    .lift(db
        .update("UPDATE mytable SET updated_at = ? WHERE id = ?")
        .parameterOperator()
    );

updateObservable.subscribe();

Thanks

[davidmoten]

It is possible that 0.6.5 release fixes this as well. Can you test please?

[davidmoten]

Sorry I should say please retry with 0.6.7

[Crystark]

The issue seems to still be there with 0.6.7

[davidmoten]

Righto I've read this one properly and yep there's a problem in your example. If you call

observable.lift(db.select(sql).dependsOnOperator().getTupleN());

then the select query doesn't run till observable completes. So nothing should happen till your Observable.interval completes.

[Crystark]

Right, it thought it would trigger each time the interval would emit a new value which is what I want to do.

Maybe what I need is parameterListOperator() with empty observables. Something like that:

updateObservable = Observable.interval(0, 10, TimeUnit.SECONDS)
    .doOnNext(s -> System.out.println("try " + s))
    .map(t -> Observable.empty())
    .lift(db
        .select("SELECT id FROM mytable ORDER BY updated_at ASC")
        .parameterListOperator()
        .getTupleN()
    )
    .doOnNext(s -> System.out.println("result " + s));

updateObservable.subscribe();

Would this work as I expect it to ? Isn't there a way I can do this without creating the empty observable ?

Thanks again for your help !