<s> import logging <EOL> from rx import Observable , Observer <EOL> from rx . abstractobserver import AbstractObserver <EOL> from rx . disposables import Disposable <EOL> from . subscription import Subscription <EOL> from . reactive_assert import AssertList <EOL> log = logging . getLogger ( "<STR_LIT>" ) <EOL> class HotObservable ( Observable ) : <EOL> def __init__ ( self , scheduler , messages ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> Observable . __init__ ( self , self . _subscribe ) <EOL> self . scheduler = scheduler <EOL> self . messages = messages <EOL> self . subscriptions = AssertList ( ) <EOL> self . observers = [ ] <EOL> observable = self <EOL> def get_action ( notification ) : <EOL> def action ( scheduler , state ) : <EOL> for observer in observable . observers [ : ] : <EOL> notification . accept ( observer ) <EOL> return Disposable . empty ( ) <EOL> return action <EOL> for message in self . messages : <EOL> notification = message . value <EOL> action = get_action ( notification ) <EOL> scheduler . schedule_absolute ( message . time , action ) <EOL> def _subscribe ( self , observer ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> observable = self <EOL> self . observers . append ( observer ) <EOL> self . subscriptions . append ( Subscription ( self . scheduler . clock ) ) <EOL> index = len ( self . subscriptions ) - <NUM_LIT:1> <EOL> def dispose_action ( ) : <EOL> log . debug ( "<STR_LIT>" % self . scheduler . clock ) <EOL> observable . observers . remove ( observer ) <EOL> start = observable . subscriptions [ index ] . subscribe <EOL> end = observable . scheduler . clock <EOL> observable . subscriptions [ index ] = Subscription ( start , end ) <EOL> return Disposable ( dispose_action ) </s>
