🔐 Sid Gifari File Manager Pro
v8.0.5 | 2026-08-09 00:46:30 | PHP 8.2.33
📂
/ (Root)
/
opt
/
alt
/
python38
/
lib
/
python3.8
/
site-packages
/
reactivex
/
testing
📍 /opt/alt/python38/lib/python3.8/site-packages/reactivex/testing
🔄 Refresh
✏️
Editing: mockobserver.py
Read Only
from typing import List, TypeVar from reactivex import abc from reactivex.notification import OnCompleted, OnError, OnNext from reactivex.scheduler import VirtualTimeScheduler from .recorded import Recorded _T = TypeVar("_T") class MockObserver(abc.ObserverBase[_T]): def __init__(self, scheduler: VirtualTimeScheduler) -> None: self.scheduler = scheduler self.messages: List[Recorded[_T]] = [] def on_next(self, value: _T) -> None: self.messages.append(Recorded(self.scheduler.clock, OnNext(value))) def on_error(self, error: Exception) -> None: self.messages.append(Recorded(self.scheduler.clock, OnError(error))) def on_completed(self) -> None: self.messages.append(Recorded(self.scheduler.clock, OnCompleted()))
💾 Save Changes
❌ Cancel