<s> import unittest <EOL> from rx import Observable <EOL> from rx . abstractobserver import AbstractObserver <EOL> from rx . linq . connectableobservable import ConnectableObservable <EOL> from rx . subjects import Subject <EOL> from rx . testing import TestScheduler , ReactiveTest , is_prime , MockDisposable <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 RxException ( Exception ) : <EOL> pass <EOL> def _raise ( ex ) : <EOL> raise RxException ( ex ) <EOL> class MySubject ( Observable , AbstractObserver ) : <EOL> def __init__ ( self ) : <EOL> def subscribe ( observer ) : <EOL> self . subscribe_count += <NUM_LIT:1> <EOL> self . observer = observer <EOL> class Duck : <EOL> def __init__ ( self , this ) : <EOL> self . this = this <EOL> def dispose ( self ) : <EOL> self . this . disposed = True <EOL> return Duck ( self ) <EOL> super ( MySubject , self ) . __init__ ( subscribe ) <EOL> self . dispose_on_map = { } <EOL> self . subscribe_count = <NUM_LIT:0> <EOL> self . disposed = False <EOL> def dispose_on ( self , value , disposable ) : <EOL> self . dispose_on_map [ value ] = disposable <EOL> def on_next ( self , value ) : <EOL> self . observer . on_next ( value ) <EOL> if value in self . dispose_on_map : <EOL> self . dispose_on_map [ value ] . dispose ( ) <EOL> def on_error ( self , exception ) : <EOL> self . observer . on_error ( exception ) <EOL> def on_completed ( self ) : <EOL> self . observer . on_completed ( ) <EOL> class TestPublish ( unittest . TestCase ) : <EOL> def test_publish_cold_zip ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:0> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_completed ( <NUM_LIT> ) <EOL> ) <EOL> def create ( ) : <EOL> def selector ( ys ) : <EOL> def result_selector ( a , b ) : <EOL> return a + b <EOL> return ys . zip ( ys , result_selector ) <EOL> return xs . publish ( selector = selector ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:10> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:12> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_ref_count_connects_on_first ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_completed ( <NUM_LIT> ) <EOL> ) <EOL> subject = MySubject ( ) <EOL> conn = ConnectableObservable ( xs , subject ) <EOL> def create ( ) : <EOL> return conn . ref_count ( ) <EOL> res = scheduler . start ( create ) <EOL> res . messages . assert_equal ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_completed ( <NUM_LIT> ) <EOL> ) <EOL> assert ( subject . disposed ) <EOL> def test_ref_count_notconnected ( self ) : <EOL> disconnected = [ False ] <EOL> count = [ <NUM_LIT:0> ] <EOL> def factory ( ) : <EOL> count [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> def create ( obs ) : <EOL> def func ( ) : <EOL> disconnected [ <NUM_LIT:0> ] = True <EOL> return func <EOL> return Observable . create ( create ) <EOL> xs = Observable . defer ( factory ) <EOL> subject = MySubject ( ) <EOL> conn = ConnectableObservable ( xs , subject ) <EOL> refd = conn . ref_count ( ) <EOL> dis1 = refd . subscribe ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , count [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , subject . subscribe_count ) <EOL> assert ( not disconnected [ <NUM_LIT:0> ] ) <EOL> dis2 = refd . subscribe ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , count [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( <NUM_LIT:2> , subject . subscribe_count ) <EOL> assert ( not disconnected [ <NUM_LIT:0> ] ) <EOL> dis1 . dispose ( ) <EOL> assert ( not disconnected [ <NUM_LIT:0> ] ) <EOL> dis2 . dispose ( ) <EOL> assert ( disconnected [ <NUM_LIT:0> ] ) <EOL> disconnected [ <NUM_LIT:0> ] = False <EOL> dis3 = refd . subscribe ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , count [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( <NUM_LIT:3> , subject . subscribe_count ) <EOL> assert ( not disconnected [ <NUM_LIT:0> ] ) <EOL> dis3 . dispose ( ) <EOL> assert ( disconnected [ <NUM_LIT:0> ] ) <EOL> def test_publish_basic ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_completed ( <NUM_LIT> ) <EOL> ) <EOL> ys = [ None ] <EOL> subscription = [ None ] <EOL> connection = [ None ] <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . publish ( ) <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 ( subscribed , 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 ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) <EOL> ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_publish_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_error ( <NUM_LIT> , ex ) ) <EOL> ys = [ None ] <EOL> subscription = [ None ] <EOL> connection = [ None ] <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . publish ( ) <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 ( subscribed , 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 ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_publish_complete ( self ) : <EOL> connection = [ None ] <EOL> subscription = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . publish ( ) <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 ( subscribed , 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 ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_publish_dispose ( self ) : <EOL> connection = [ None ] <EOL> subscription = [ None ] <EOL> ys = [ None ] <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_completed ( <NUM_LIT> ) ) <EOL> results = scheduler . create_observer ( ) <EOL> def action0 ( scheduler , state ) : <EOL> ys [ <NUM_LIT:0> ] = xs . publish ( ) <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 ( subscribed , 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:8> ) ) <EOL> xs . subscriptions . assert_equal ( <EOL> subscribe ( <NUM_LIT> , <NUM_LIT> ) , <EOL> subscribe ( <NUM_LIT> , <NUM_LIT> ) , <EOL> subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_publish_multipleconnections ( self ) : <EOL> xs = Observable . never ( ) <EOL> ys = xs . publish ( ) <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> connection3 . dispose ( ) <EOL> def test_publish_lambda_zip_complete ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . zip ( _xs . skip ( <NUM_LIT:1> ) , lambda prev , cur : cur + prev ) <EOL> return xs . publish ( selector ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:15> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_publish_lambda_zip_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . zip ( _xs . skip ( <NUM_LIT:1> ) , lambda prev , cur : cur + prev ) <EOL> return xs . publish ( selector ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:15> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_publish_lambda_zip_dispose ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:3> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:4> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:1> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:8> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:6> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:2> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( _xs ) : <EOL> return _xs . zip ( _xs . skip ( <NUM_LIT:1> ) , lambda prev , cur : cur + prev ) <EOL> return xs . publish ( selector ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( <EOL> on_next ( <NUM_LIT> , <NUM_LIT:7> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:5> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:9> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:20> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:15> ) , <EOL> on_next ( <NUM_LIT> , <NUM_LIT:11> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) </s>
