<s> import unittest <EOL> from rx . observable import Observable <EOL> from rx . testing import TestScheduler , ReactiveTest <EOL> from rx . disposables import Disposable , SerialDisposable <EOL> on_next = ReactiveTest . on_next <EOL> on_completed = ReactiveTest . on_completed <EOL> on_error = ReactiveTest . on_error <EOL> subscribe = ReactiveTest . subscribe <EOL> subscribed = ReactiveTest . subscribed <EOL> disposed = ReactiveTest . disposed <EOL> created = ReactiveTest . created <EOL> class TestReplay ( unittest . TestCase ) : <EOL> def test_replay_count_basic ( self ) : <EOL> connection = [ None ] <EOL> subscription = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , <NUM_LIT:3> , None , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( disposed , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> def action7 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action7 ) <EOL> def action8 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action8 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_count_error ( self ) : <EOL> connection = [ None ] <EOL> subscription = [ None ] <EOL> ys = [ None ] <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , <NUM_LIT:3> , None , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( disposed , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_count_complete ( self ) : <EOL> connection = [ None ] <EOL> subscription = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , <NUM_LIT:3> , None , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scehduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( disposed , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_count_dispose ( self ) : <EOL> connection = [ None ] <EOL> subscription = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , <NUM_LIT:3> , None , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> def action7 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action7 ) <EOL> def action8 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action8 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_count_multiple_connections ( self ) : <EOL> xs = Observable . never ( ) <EOL> ys = xs . replay ( None , <NUM_LIT:3> ) <EOL> connection1 = ys . connect ( ) <EOL> connection2 = ys . connect ( ) <EOL> assert ( connection1 == connection2 ) <EOL> connection1 . dispose ( ) <EOL> connection2 . dispose ( ) <EOL> connection3 = ys . connect ( ) <EOL> assert ( connection1 != connection3 ) <EOL> def test_replay_count_lambda_zip_complete ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> def action ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . take ( <NUM_LIT:6> ) . repeat ( ) <EOL> return xs . replay ( selector , <NUM_LIT:3> , None , scheduler ) <EOL> results = scheduler . start ( action , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_replay_count_lambda_zip_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . take ( <NUM_LIT:6> ) . repeat ( ) <EOL> return xs . replay ( selector , <NUM_LIT:3> , None , scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_replay_count_lambda_zip_dispose ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . take ( <NUM_LIT:6> ) . repeat ( ) <EOL> return xs . replay ( selector , <NUM_LIT:3> , None , scheduler ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_replay_time_basic ( self ) : <EOL> subscription = [ None ] <EOL> connection = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , None , <NUM_LIT> , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( disposed , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> def action7 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action7 ) <EOL> def action8 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action8 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_time_error ( self ) : <EOL> subscription = [ None ] <EOL> connection = [ None ] <EOL> ys = [ None ] <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , None , <NUM_LIT> , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( disposed , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_time_complete ( self ) : <EOL> subscription = [ None ] <EOL> connection = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , None , <NUM_LIT> , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( disposed , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_time_dispose ( self ) : <EOL> subscription = [ None ] <EOL> connection = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . replay ( None , None , <NUM_LIT:100> , scheduler ) <EOL> scheduler . schedule_absolute ( created , action0 ) <EOL> def action1 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . subscribe ( results ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action1 ) <EOL> def action2 ( scheduler , state ) : <EOL> subscription [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action2 ) <EOL> def action3 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action3 ) <EOL> def action4 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action4 ) <EOL> def action5 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action5 ) <EOL> def action6 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action6 ) <EOL> def action7 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] = ys [ <NUM_LIT:0> ] . connect ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action7 ) <EOL> def action8 ( scheduler , state ) : <EOL> connection [ <NUM_LIT:0> ] . dispose ( ) <EOL> scheduler . schedule_absolute ( <NUM_LIT> , action8 ) <EOL> scheduler . start ( ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_replay_time_multiple_connections ( self ) : <EOL> xs = Observable . never ( ) <EOL> ys = xs . replay ( None , None , <NUM_LIT:100> ) <EOL> connection1 = ys . connect ( ) <EOL> connection2 = ys . connect ( ) <EOL> assert ( connection1 == connection2 ) <EOL> connection1 . dispose ( ) <EOL> connection2 . dispose ( ) <EOL> connection3 = ys . connect ( ) <EOL> assert ( connection1 != connection3 ) <EOL> def test_replay_time_lambda_zip_complete ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . take ( <NUM_LIT:6> ) . repeat ( ) <EOL> return xs . replay ( selector , None , <NUM_LIT:50> , scheduler ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_replay_time_lambda_zip_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . take ( <NUM_LIT:6> ) . repeat ( ) <EOL> return xs . replay ( selector , None , <NUM_LIT:50> , scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_replay_time_lambda_zip_dispose ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) , on_next ( <NUM_LIT> , <NUM_LIT:11> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . take ( <NUM_LIT:6> ) . repeat ( ) <EOL> return xs . replay ( selector , None , <NUM_LIT:50> , scheduler ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:8> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:7> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:9> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) </s>
