numa_hit 261780419 numa_miss 0 numa_foreign 0 interleave_hit 938 local_node 261780419 other_node 0  ˆ¾Á ƒh¹À¿Ã numa_hit 261780945 numa_miss 0 numa_foreign 0 interleave_hit 938 local_node 261780945 other_node 0 $ ˆl–ßi~” jn-jqA àAq·Tţ „„'ÁÀÄ?÷ # # Copyright (C) 2022 Red Hat, Inc. # SPDX-License-Identifier: GPL-3.0-or-later import asyncio import codecs import json import logging import traceback import typing from typing import BinaryIO, Callable, ClassVar, Collection, Generator, Mapping, Sequence, Type from .jsonutil import JsonError, JsonObject, JsonValue, create_object, get_bool, get_enum, get_str from .protocol import CockpitProblem from .router import Endpoint, Router, RoutingRule logger = logging.getLogger(__name__) if typing.TYPE_CHECKING: _T = typing.TypeVar('_T') _P = typing.ParamSpec("_P") class ChannelRoutingRule(RoutingRule): table: 'dict[str, list[Type[Channel]]]' def __init__(self, router: Router, channel_types: 'Collection[Type[Channel]]'): super().__init__(router) self.table = {} # Sort the channels into buckets by payload type for cls in channel_types: entry = self.table.setdefault(cls.payload, []) entry.append(cls) # Within each bucket, sort the channels so those with more # restrictions are considered first. for entry in self.table.values(): entry.sort(key=lambda cls: len(cls.restrictions), reverse=True) def capabilities(self) -> JsonObject: result: 'dict[str, list[str]]' = {} for payload, impls in self.table.items(): caps: 'list[str]' = [] for impl in impls: caps.extend(impl.capabilities) result[payload] = caps return result def check_restrictions(self, restrictions: 'Collection[tuple[str, object]]', options: JsonObject) -> bool: for key, expected_value in restrictions: our_value = options.get(key) # If the match rule specifies that a value must be present and # we don't have it, then fail. if our_value is None: return False # If the match rule specified a specific expected value, and # our value doesn't match it, then fail. if expected_value is not None and our_value != expected_value: return False # Everything checked out return True def apply_rule(self, options: JsonObject) -> 'Channel | None': assert self.router is not None payload = options.get('payload') if not isinstance(payload, str): return None for cls in self.table.get(payload, []): if self.check_restrictions(cls.restrictions, options): return cls(self.router) else: return None def shutdown(self) -> None: pass # we don't hold any state class ChannelError(CockpitProblem): pass class Channel(Endpoint): # Values borrowed from C implementation BLOCK_SIZE = 16 * 1024 SEND_WINDOW = 2 * 1024 * 1024 # Flow control book-keeping _send_pings: bool = False _out_sequence: int = 0 _out_window: int = SEND_WINDOW _ack_bytes: bool # Task management _tasks: 'set[asyncio.Task]' _close_args: 'JsonObject | None' = None # Must be filled in by the channel implementation payload: 'ClassVar[str]' restrictions: 'ClassVar[Sequence[tuple[str, object]]]' = () capabilities: 'ClassVar[Sequence[str]]' = () # These get filled in from .do_open() channel = '' group = '' is_binary: bool decoder: 'codecs.IncrementalDecoder | None' # input def do_control(self, command: str, message: JsonObject) -> None: # Break the various different kinds of control messages out into the # things that our subclass may be interested in handling. We drop the # 'message' field for handlers that don't need it. if command == 'open': self._tasks = set() se