TakeLastTimed with backpressure support

Implemented #1488. I moved `OperatorTakeLast.QueueProducer` to `TakeLastQueueProducer` to reuse it.