md.subscription = duration.take(1).subscribe(
              lambda {|_| },
              lambda {|e|
                left_map.values.each {|v| v.on_error(e) }
                observer.on_error(e)