SubscriptionRef

__init__(value) -> None

def __init__(self, value: A) -> None:
    super().__init__(value)
    self._pubsub = PubSub(Atomic[A].unbounded(replay=1), "dropping")
    self._pubsub.unsafe_publish(value)

make(value) -> Effect[SubscriptionRef[T]]

@staticmethod
def make[T](value: T) -> Effect[SubscriptionRef[T]]:
    return _Sync(lambda: SubscriptionRef(value))

subscribe() -> Effect[Subscription[A], Never, Scope]

def subscribe(self) -> Effect[Subscription[A], Never, Scope]:
    """Subscribe until the current scope closes: the current value, then
    every later one."""
    return self._pubsub.subscribe()

changes() -> Stream[A]

def changes(self) -> Stream[A]:
    """A stream of the current value, then every later one (Effect's
    ``SubscriptionRef.changes``). Each run subscribes anew."""
    from pyfect.stream import from_pubsub

    return from_pubsub(self._pubsub)

unsafe_get() -> A

def unsafe_get(self) -> A:
    return self._value

get() -> Effect[A]

def get(self) -> Effect[A]:
    return _Sync(lambda: self._value)

set(value) -> Effect[None]

def set(self, value: A) -> Effect[None]:
    return self._locked(lambda _: (None, Some(value)))

set_and_get(value) -> Effect[A]

def set_and_get(self, value: A) -> Effect[A]:
    return self._locked(lambda _: (value, Some(value)))

get_and_set(value) -> Effect[A]

def get_and_set(self, value: A) -> Effect[A]:
    return self._locked(lambda old: (old, Some(value)))

update(f) -> Effect[None]

def update(self, f: Callable[[A], A]) -> Effect[None]:
    return self._locked(lambda old: (None, Some(f(old))))

update_and_get(f) -> Effect[A]

def update_and_get(self, f: Callable[[A], A]) -> Effect[A]:
    return self._locked(lambda old: _new_and_get(f(old)))

get_and_update(f) -> Effect[A]

def get_and_update(self, f: Callable[[A], A]) -> Effect[A]:
    return self._locked(lambda old: (old, Some(f(old))))

update_some(f) -> Effect[None]

def update_some(self, f: Callable[[A], Option[A]]) -> Effect[None]:
    return self._locked(lambda old: (None, f(old)))

update_some_and_get(f) -> Effect[A]

def update_some_and_get(self, f: Callable[[A], Option[A]]) -> Effect[A]:
    return self._locked(lambda old: _some_and_get(old, f(old)))

get_and_update_some(f) -> Effect[A]

def get_and_update_some(self, f: Callable[[A], Option[A]]) -> Effect[A]:
    return self._locked(lambda old: (old, f(old)))

modify(f) -> Effect[B]

def modify[B](self, f: Callable[[A], tuple[B, A]]) -> Effect[B]:
    return self._locked(lambda old: _modified(f(old)))

modify_some(f) -> Effect[B]

def modify_some[B](self, f: Callable[[A], tuple[B, Option[A]]]) -> Effect[B]:
    return self._locked(f)

update_effect(f) -> Effect[None, E, R]

def update_effect[E = Never, R = Never](
    self, f: Callable[[A], Effect[A, E, R]]
) -> Effect[None, E, R]:
    return self._locked_effect(lambda old: f(old).map(lambda new: (None, Some(new))))

update_and_get_effect(f) -> Effect[A, E, R]

def update_and_get_effect[E = Never, R = Never](
    self, f: Callable[[A], Effect[A, E, R]]
) -> Effect[A, E, R]:
    return self._locked_effect(lambda old: f(old).map(_new_and_get))

get_and_update_effect(f) -> Effect[A, E, R]

def get_and_update_effect[E = Never, R = Never](
    self, f: Callable[[A], Effect[A, E, R]]
) -> Effect[A, E, R]:
    return self._locked_effect(lambda old: f(old).map(lambda new: (old, Some(new))))

update_some_effect(f) -> Effect[None, E, R]

def update_some_effect[E = Never, R = Never](
    self, f: Callable[[A], Effect[Option[A], E, R]]
) -> Effect[None, E, R]:
    return self._locked_effect(lambda old: f(old).map(lambda new: (None, new)))

update_some_and_get_effect(f) -> Effect[A, E, R]

def update_some_and_get_effect[E = Never, R = Never](
    self, f: Callable[[A], Effect[Option[A], E, R]]
) -> Effect[A, E, R]:
    return self._locked_effect(lambda old: f(old).map(lambda new: _some_and_get(old, new)))

get_and_update_some_effect(f) -> Effect[A, E, R]

def get_and_update_some_effect[E = Never, R = Never](
    self, f: Callable[[A], Effect[Option[A], E, R]]
) -> Effect[A, E, R]:
    return self._locked_effect(lambda old: f(old).map(lambda new: (old, new)))

modify_effect(f) -> Effect[B, E, R]

def modify_effect[B, E = Never, R = Never](
    self, f: Callable[[A], Effect[tuple[B, A], E, R]]
) -> Effect[B, E, R]:
    return self._locked_effect(lambda old: f(old).map(_modified))

modify_some_effect(f) -> Effect[B, E, R]

def modify_some_effect[B, E = Never, R = Never](
    self, f: Callable[[A], Effect[tuple[B, Option[A]], E, R]]
) -> Effect[B, E, R]:
    return self._locked_effect(f)