byp4a55 professor6T9

Linux das102.truehost.cloud 4.18.0-553.8.1.lve.el8.x86_64 #1 SMP Thu Jul 4 16:24:39 UTC 2024 x86_64
LiteSpeed
156.232.88.2

File Name : hotobservable.py

from typing import Any, List, Optional, TypeVar

from reactivex import Observable, abc
from reactivex.disposable import Disposable
from reactivex.notification import Notification
from reactivex.scheduler import VirtualTimeScheduler

from .recorded import Recorded
from .subscription import Subscription

_T = TypeVar("_T")

class HotObservable(Observable[_T]):
    def __init__(
        self, scheduler: VirtualTimeScheduler, messages: List[Recorded[_T]]
    ) -> None:
        super().__init__()

self.scheduler = scheduler
        self.messages = messages
        self.subscriptions: List[Subscription] = []
        self.observers: List[abc.ObserverBase[_T]] = []

observable = self

def get_action(notification: Notification[_T]) -> abc.ScheduledAction[_T]:
            def action(scheduler: abc.SchedulerBase, state: Any) -> abc.DisposableBase:
                for observer in observable.observers[:]:
                    notification.accept(observer)
                return Disposable()

return action

for message in self.messages:
            notification = message.value
            if not isinstance(notification, Notification):
                raise ValueError("Must be notification")

# Warning: Don't make closures within a loop
            action = get_action(notification)
            scheduler.schedule_absolute(message.time, action)

def _subscribe_core(
        self,
        observer: Optional[abc.ObserverBase[_T]] = None,
        scheduler: Optional[abc.SchedulerBase] = None,
    ) -> abc.DisposableBase:
        if observer:
            self.observers.append(observer)
        self.subscriptions.append(Subscription(self.scheduler.clock))
        index = len(self.subscriptions) - 1

def dispose_action() -> None:
            if observer:
                self.observers.remove(observer)
            start = self.subscriptions[index].subscribe
            end = self.scheduler.clock
            self.subscriptions[index] = Subscription(start, end)

return Disposable(dispose_action)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2026-02-16 08:57:59
..	dir	-	root/root	0755	2026-02-16 08:57:59
__pycache__	dir	-	root/root	0755	2026-02-16 08:57:59
__init__.py	text/x-script.python	356 B	root/root	0644	2026-02-16 08:57:59
coldobservable.py	text/x-script.python	1.9 KB	root/root	0644	2026-02-16 08:57:59
hotobservable.py	text/x-script.python	2.03 KB	root/root	0644	2026-02-16 08:57:59
marbles.py	text/x-script.python	5.63 KB	root/root	0644	2026-02-16 08:57:59
mockdisposable.py	text/x-script.python	436 B	root/root	0644	2026-02-16 08:57:59
mockobserver.py	text/x-script.python	780 B	root/root	0644	2026-02-16 08:57:59
reactivetest.py	text/x-script.python	2.07 KB	root/root	0644	2026-02-16 08:57:59
recorded.py	text/x-script.python	1.07 KB	root/root	0644	2026-02-16 08:57:59
subscription.py	text/x-script.python	701 B	root/root	0644	2026-02-16 08:57:59
testscheduler.py	text/x-script.python	5.74 KB	root/root	0644	2026-02-16 08:57:59

Team Anon Force

https://t.me/Professor6T9x