Create an interval and countdown using rxpy

Question

I am making a Discord bot which allows you to create a poll. The user can put as an argument how long the poll will be. So I want to refresh every 5s or 10s (or maybe more) the message with the poll editing how much time the user has left. I want to implement a countdown from 3600 seconds for example, and every 5s or 10s execute a function which will edit the message until the time goes to 0. Everything on the bot side I have it under control and more or less I know how to implement it.

So, what I thought is making an interval and stop when the current time is equal to time when it started + duration of the poll. So I can use the rx.interval() for creating the observable and use an operator like .take_while().

This is my code:

import time
import rx
print(rx.__version__) # 3.1.1

started_at = time.time() # Time in seconds
end_at = started_at + 3600 # One hour after
source = rx.interval(5).take_while(time.time() < end_at)

But I get AttributeError: 'Observable' object has no attribute 'take_while'.

I think I should put it in a pipe or something like this:

from rx import operators as op
sub = source.pipe(op.take_while(time.time() < end_at))

But I get TypeError: 'bool' object is not callable

How can I use take_while? Thank you!

Answer 1

You should pipe the source and then subscribe. You have to use operators inside of the pipe() method that the Observable has (https://rxpy.readthedocs.io/en/latest/reference_observable.html#rx.Observable.pipe)

The code should look something like this

import time
import rx
from rx import operators as op

print(rx.__version__) # 3.1.1

started_at = time.time() # Time in seconds
end_at = started_at + 3600 # One hour after
ob = rx.interval(5)
sub = ob.pipe(op.take_while(lambda _: time.time() < end_at))
sub.subscribe(lambda i: print(i))