同步模式：Thread per Request
Thread-1: [──gRPC Call──][──Wait──][──Process──]
Thread-2: [──gRPC Call──][──Wait──][──Process──]
Thread-3: [──gRPC Call──][──Wait──][──Process──]
→ 3 线程，3 请求

异步模式：Event Loop
Event Loop: [gRPC-1][gRPC-2][gRPC-3][Wait...][Process-1][Process-2][Process-3]
→ 1 线程，3 请求，更高吞吐

class AsyncOntoPlatform:
    """异步 Ontology 平台客户端"""

    def __init__(self, config: SDKConfig):
        self._config = config
        self._connection_manager = AsyncConnectionManager(config)
        self._interceptor_chain = InterceptorChain([
            AsyncAuthInterceptor(config.auth_provider),
            AsyncRetryInterceptor(config.max_retries),
            AsyncTracingInterceptor(),
            AsyncMetricsInterceptor(),
        ])

        # 领域客户端
        self.objects = AsyncObjectClient(self._connection_manager, self._interceptor_chain)
        self.actions = AsyncActionClient(self._connection_manager, self._interceptor_chain)
        self.search = AsyncSearchClient(self._connection_manager, self._interceptor_chain)
        self.links = AsyncLinkClient(self._connection_manager, self._interceptor_chain)

    @classmethod
    async def connect(cls, url: str, **kwargs) -> "AsyncOntoPlatform":
        """异步工厂方法"""
        config = SDKConfig(platform_url=url, **kwargs)
        client = cls(config)
        await client._connection_manager.initialize()
        return client

    async def close(self) -> None:
        """关闭连接"""
        await self._connection_manager.close()

    async def __aenter__(self) -> "AsyncOntoPlatform":
        return self

    async def __aexit__(self, *args) -> None:
        await self.close()

class AsyncConnectionManager:
    """异步连接管理器"""

    def __init__(self, config: SDKConfig):
        self._config = config
        self._channel: grpc.aio.Channel | None = None
        self._semaphore = asyncio.Semaphore(config.max_connections)
        self._health_check_task: asyncio.Task | None = None

    async def initialize(self) -> None:
        self._channel = grpc.aio.insecure_channel(
            self._config.platform_url,
            options=[
                ("grpc.max_receive_message_length", 50 * 1024 * 1024),
                ("grpc.keepalive_time_ms", 10000),
                ("grpc.keepalive_timeout_ms", 5000),
                ("grpc.http2.max_pings_without_data", 0),
            ],
        )
        self._health_check_task = asyncio.create_task(self._health_check_loop())

    async def acquire(self) -> grpc.aio.Channel:
        """获取连接（受信号量控制）"""
        await self._semaphore.acquire()
        return self._channel

    def release(self) -> None:
        """释放连接"""
        self._semaphore.release()

    @asynccontextmanager
    async def connection(self):
        """连接上下文管理器"""
        channel = await self.acquire()
        try:
            yield channel
        finally:
            self.release()

    async def _health_check_loop(self) -> None:
        while True:
            try:
                state = self._channel.get_state(try_to_connect=True)
                if state == grpc.ChannelConnectivity.TRANSIENT_FAILURE:
                    logger.warning("gRPC channel in transient failure state")
            except Exception as e:
                logger.error(f"Health check failed: {e}")
            await asyncio.sleep(30)

class AsyncObjectClient:
    """异步对象客户端"""

    async def get_many(
        self,
        object_type: str,
        object_ids: list[str],
        concurrency: int = 10,
    ) -> list[OntologyObject]:
        """并发获取多个对象"""
        semaphore = asyncio.Semaphore(concurrency)

        async def fetch_one(oid: str) -> OntologyObject:
            async with semaphore:
                return await self.get(object_type, oid)

        tasks = [fetch_one(oid) for oid in object_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        objects = []
        errors = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                errors.append((object_ids[i], result))
            else:
                objects.append(result)

        if errors:
            logger.warning(f"Failed to fetch {len(errors)} objects: {errors[:5]}")

        return objects

    async def parallel_query(
        self,
        queries: list[QuerySpec],
        concurrency: int = 5,
    ) -> list[QueryResult]:
        """并行执行多个查询"""
        semaphore = asyncio.Semaphore(concurrency)

        async def run_query(spec: QuerySpec) -> QueryResult:
            async with semaphore:
                return await self._execute_query(spec)

        return await asyncio.gather(*[run_query(q) for q in queries])

class AsyncBatchWriter:
    """异步批量写入器"""

    def __init__(
        self,
        client: AsyncObjectClient,
        batch_size: int = 100,
        max_concurrent_batches: int = 5,
        flush_interval: float = 1.0,
    ):
        self._client = client
        self._batch_size = batch_size
        self._semaphore = asyncio.Semaphore(max_concurrent_batches)
        self._buffer: list[WriteOperation] = []
        self._flush_interval = flush_interval
        self._flush_task: asyncio.Task | None = None

    async def add(self, operation: WriteOperation) -> None:
        """添加写入操作到缓冲区"""
        self._buffer.append(operation)
        if len(self._buffer) >= self._batch_size:
            await self.flush()

    async def flush(self) -> BatchResult:
        """刷新缓冲区"""
        if not self._buffer:
            return BatchResult(total=0, succeeded=0, failed=0)

        batch = self._buffer[:self._batch_size]
        self._buffer = self._buffer[self._batch_size:]

        async with self._semaphore:
            return await self._client.batch_write(batch)

    async def __aenter__(self):
        self._flush_task = asyncio.create_task(self._auto_flush())
        return self

    async def __aexit__(self, *args):
        if self._flush_task:
            self._flush_task.cancel()
        while self._buffer:
            await self.flush()

    async def _auto_flush(self):
        while True:
            await asyncio.sleep(self._flush_interval)
            if self._buffer:
                await self.flush()

class AsyncResultStream:
    """异步结果流 - 支持背压控制"""

    def __init__(
        self,
        grpc_stream: grpc.aio.UnaryStreamCall,
        buffer_size: int = 100,
    ):
        self._stream = grpc_stream
        self._buffer: asyncio.Queue = asyncio.Queue(maxsize=buffer_size)
        self._exhausted = False
        self._fetch_task: asyncio.Task | None = None

    def __aiter__(self):
        return self

    async def __anext__(self) -> OntologyObject:
        if self._fetch_task is None:
            self._fetch_task = asyncio.create_task(self._fetch_loop())

        try:
            item = await asyncio.wait_for(self._buffer.get(), timeout=30)
            if item is _SENTINEL:
                raise StopAsyncIteration
            return item
        except asyncio.TimeoutError:
            raise StopAsyncIteration

    async def _fetch_loop(self):
        try:
            async for response in self._stream:
                obj = self._deserialize(response)
                await self._buffer.put(obj)  # 背压：缓冲区满时等待
        except grpc.aio.AioRpcError as e:
            logger.error(f"Stream error: {e}")
        finally:
            await self._buffer.put(_SENTINEL)


# 使用示例
async for employee in client.objects.Employee.stream(
    filters=[Filter("department", "eq", "Engineering")],
    page_size=500,
):
    await process_employee(employee)

class AsyncPipeline:
    """异步数据处理管道"""

    def __init__(self):
        self._stages: list[PipelineStage] = []

    def map(self, func: Callable) -> "AsyncPipeline":
        self._stages.append(MapStage(func))
        return self

    def filter(self, predicate: Callable) -> "AsyncPipeline":
        self._stages.append(FilterStage(predicate))
        return self

    def batch(self, size: int) -> "AsyncPipeline":
        self._stages.append(BatchStage(size))
        return self

    async def execute(self, source: AsyncIterable) -> list:
        results = []
        async for item in self._apply_stages(source):
            results.append(item)
        return results

    async def _apply_stages(self, source: AsyncIterable) -> AsyncIterable:
        current = source
        for stage in self._stages:
            current = stage.process(current)
        async for item in current:
            yield item


# 使用示例
results = await (
    AsyncPipeline()
    .filter(lambda emp: emp.salary > 50000)
    .map(lambda emp: {"name": emp.name, "salary": emp.salary})
    .batch(100)
    .execute(client.objects.Employee.stream())
)

class SyncOntoPlatform:
    """同步包装器 - 为非异步场景提供同步 API"""

    def __init__(self, async_client: AsyncOntoPlatform):
        self._async_client = async_client
        self._loop = asyncio.new_event_loop()

    @classmethod
    def connect(cls, url: str, **kwargs) -> "SyncOntoPlatform":
        loop = asyncio.new_event_loop()
        async_client = loop.run_until_complete(
            AsyncOntoPlatform.connect(url, **kwargs)
        )
        wrapper = cls(async_client)
        wrapper._loop = loop
        return wrapper

    def _run(self, coro):
        return self._loop.run_until_complete(coro)

    @property
    def objects(self) -> "SyncObjectClient":
        return SyncObjectClient(self._async_client.objects, self._run)

    def close(self) -> None:
        self._run(self._async_client.close())
        self._loop.close()

# 使用示例
client = SyncOntoPlatform.connect("https://platform.example.com", token="...")
employee = client.objects.get("Employee", "emp-001")
client.close()

class GracefulShutdown:
    """优雅关闭管理"""

    def __init__(self, client: AsyncOntoPlatform):
        self._client = client
        self._pending_tasks: set[asyncio.Task] = set()

    async def shutdown(self, timeout: float = 30.0) -> None:
        # 1. 停止接受新请求
        self._client._accepting_requests = False

        # 2. 等待进行中的请求完成
        if self._pending_tasks:
            done, pending = await asyncio.wait(
                self._pending_tasks, timeout=timeout,
            )
            for task in pending:
                task.cancel()

        # 3. 关闭连接
        await self._client.close()

class TestAsyncSDK:
    @pytest.mark.asyncio
    async def test_concurrent_get_many(self):
        async with AsyncOntoPlatform.connect(test_url, token=token) as client:
            ids = [f"emp-{i:03d}" for i in range(100)]
            results = await client.objects.get_many("Employee", ids, concurrency=10)
            assert len(results) == 100

    @pytest.mark.asyncio
    async def test_batch_writer(self):
        async with AsyncOntoPlatform.connect(test_url, token=token) as client:
            async with AsyncBatchWriter(client.objects, batch_size=50) as writer:
                for i in range(200):
                    await writer.add(WriteOperation(
                        type="create",
                        object_type="Employee",
                        properties={"name": f"Employee {i}", "department": "Test"},
                    ))
            # All 200 records should be written

    @pytest.mark.asyncio
    async def test_streaming_with_backpressure(self):
        async with AsyncOntoPlatform.connect(test_url, token=token) as client:
            count = 0
            async for emp in client.objects.Employee.stream():
                count += 1
                if count >= 1000:
                    break
            assert count == 1000

    @pytest.mark.asyncio
    async def test_graceful_shutdown(self):
        client = await AsyncOntoPlatform.connect(test_url, token=token)
        # Start some operations
        task = asyncio.create_task(client.objects.get_many("Employee", ids))
        # Shutdown should wait for task
        await GracefulShutdown(client).shutdown(timeout=10)