| | |
- builtins.object
-
- runtimepy.net.arbiter.factory.connection.ConnectionFactory
- runtimepy.net.arbiter.base.BaseConnectionArbiter(vcorelib.namespace.mixin.NamespaceMixin, vcorelib.logging.LoggerMixin, runtimepy.tui.mixin.TuiMixin)
-
- runtimepy.net.arbiter.factory.connection.FactoryConnectionArbiter
class ConnectionFactory(builtins.object) |
| |
An interface for creating client connections.
|
| |
Methods defined here:
- async client(self, name: str, *args, **kwargs) -> runtimepy.net.connection.Connection
- Create a client connection.
- async server_task(self, stop_sig: asyncio.locks.Event, manager: runtimepy.net.manager.ConnectionManager, started_sem: asyncio.locks.Semaphore, *args, **kwargs) -> Awaitable[NoneType]
- Create a task that will run a connection server.
Data descriptors defined here:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
|
class FactoryConnectionArbiter(runtimepy.net.arbiter.base.BaseConnectionArbiter) |
| |
FactoryConnectionArbiter(manager: runtimepy.net.manager.ConnectionManager = None, stop_sig: asyncio.locks.Event = None, namespace: vcorelib.namespace.base.Namespace = None, logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, app: Union[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]], list[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]]]] = None, config: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, window: Optional[Any] = None) -> None
A class implementing an interface to allow connection factories to be
used for creating new connections (that can then be registered).
|
| |
- Method resolution order:
- FactoryConnectionArbiter
- runtimepy.net.arbiter.base.BaseConnectionArbiter
- vcorelib.namespace.mixin.NamespaceMixin
- vcorelib.logging.LoggerMixin
- runtimepy.tui.mixin.TuiMixin
- builtins.object
Methods defined here:
- async factory_client(self, factory: str, name: str, *args, defer: bool = False, **kwargs) -> bool
- Attempt to register a client connection using a registered factory.
- async factory_server(self, factory: str, *args, **kwargs) -> bool
- Attempt to create a server task using a registered factory.
- register_connection_factory(self, factory: runtimepy.net.arbiter.factory.connection.ConnectionFactory, *namespaces: str) -> bool
- Attempt to register a connection factory.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from runtimepy.net.arbiter.base.BaseConnectionArbiter:
- __init__(self, manager: runtimepy.net.manager.ConnectionManager = None, stop_sig: asyncio.locks.Event = None, namespace: vcorelib.namespace.base.Namespace = None, logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, app: Union[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]], list[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]]]] = None, config: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, window: Optional[Any] = None) -> None
- Initialize this connection arbiter.
- async app(self, app: Union[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]], list[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]]]] = None, check_connections: bool = True, config: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None) -> int
- Run the application alongside the connection manager and server tasks.
- register_connection(self, connection: Union[runtimepy.net.connection.Connection, Awaitable[runtimepy.net.connection.Connection]], *names: str, delim: str = None) -> bool
- Attempt to register a connection object.
- run(self, app: Union[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]], list[Callable[[runtimepy.net.arbiter.info.AppInfo], Awaitable[int]]]] = None, eloop: asyncio.events.AbstractEventLoop = None, signals: Iterable[int] = None, check_connections: bool = True, config: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]] = None, enable_uvloop: bool = True) -> int
- Run the application until the stop signal is set.
Methods inherited from vcorelib.namespace.mixin.NamespaceMixin:
- child_namespace(self, *names: str, namespace: vcorelib.namespace.base.Namespace = None) -> vcorelib.namespace.base.Namespace
- Obtain a child namespace.
- names_pushed(self, *names: str, namespace: vcorelib.namespace.base.Namespace = None) -> Iterator[NoneType]
- Apply some names to this object's namespace as a managed context.
- namespace(self, name: str = None, delim: str = None, namespace: vcorelib.namespace.base.Namespace = None) -> str
- Get a namespace string for this object.
- namespace_search(self, *names: str, pattern: str = '.*', namespace: vcorelib.namespace.base.Namespace = None) -> Iterator[str]
- Perform a search on the namespace.
- namespace_suggest(self, data: str, delta: bool = True, namespace: vcorelib.namespace.base.Namespace = None) -> Optional[str]
- Find the shortest name suggestion.
- pop_name(self, name: str = None, namespace: vcorelib.namespace.base.Namespace = None) -> str
- Pop the latest name off the stack.
- push_name(self, name: str, namespace: vcorelib.namespace.base.Namespace = None) -> None
- Push a name onto the stack.
Readonly properties inherited from vcorelib.namespace.mixin.NamespaceMixin:
- ns
- Get this instance's namespace.
Data descriptors inherited from vcorelib.namespace.mixin.NamespaceMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Methods inherited from runtimepy.tui.mixin.TuiMixin:
- async handle_char(self, char: int) -> bool
- Handle character input.
- init(self, window: Optional[Any]) -> bool
- Initialize this interface's window.
- tui_update(self) -> None
- Re-draw the screen.
- update_dimensions(self) -> Any
- Handle an update to the window's dimensions.
Readonly properties inherited from runtimepy.tui.mixin.TuiMixin:
- window
- Get the window for this instance.
| |