add tests and update *.md files

Author: lesnik512

CLAUDE.md

@@ -64,3 +64,6 @@ Runs as a background asyncio task (spawned via `spawn()`). Collects `KafkaCommit
 - Always use `auto_offset_reset="earliest"` on test subscribers. The default `"latest"` causes the consumer to miss messages published before it gets its partition assignment.
 - Pre-create topics with `AIOKafkaAdminClient` before starting the broker. Auto-creation on first publish triggers a `NotLeaderForPartitionError` retry loop that can outlast short sleeps.
 - After `await broker.start()`, sleep ~1.5 s before publishing to let the consumer join the group and receive partition assignments.
+- `AsgiFastStream` lifespan tests must use `async with app.start_lifespan_context()` — calling `app.start()` / `app.stop()` bypasses the `lifespan` context manager entirely.
+- `AsgiFastStream` injects its own app-level `ContextRepo` into the lifespan, separate from `broker.context`. Pass `broker.context` explicitly to `initialize_concurrent_processing` and `stop_concurrent_processing`.
+- Subscriber-level `middlewares` on `@broker.subscriber(...)` takes `SubscriberMiddleware` (a plain `(call_next, msg)` callable), not `BaseMiddleware` subclasses. To scope `KafkaConcurrentProcessingMiddleware` to a subset of subscribers, use `KafkaRouter(middlewares=[KafkaConcurrentProcessingMiddleware])` and `broker.include_router(router)`.

README.md

@@ -35,40 +35,50 @@ pip install faststream-concurrent-aiokafka
 `ack_policy=AckPolicy.MANUAL` is **required** on every subscriber — the middleware enforces this at runtime.
 Without it, aiokafka's auto-commit timer would commit offsets before processing tasks complete, causing silent message loss on crash.
 
+> **`AsgiFastStream` note**: its lifespan receives an app-level `ContextRepo` separate from `broker.context`. Pass `broker.context` explicitly instead of the injected argument.
+
 ```python
-from faststream import FastStream, ContextRepo
-from faststream.kafka import KafkaBroker
+from contextlib import asynccontextmanager
+from faststream import ContextRepo
+from faststream.asgi import AsgiFastStream
+from faststream.kafka import KafkaBroker, KafkaRouter
 from faststream.middlewares import AckPolicy
 from faststream_concurrent_aiokafka import (
     KafkaConcurrentProcessingMiddleware,
     initialize_concurrent_processing,
-    is_kafka_handler_healthy,
     stop_concurrent_processing,
 )
 
+# Middleware applied globally to all subscribers
 broker = KafkaBroker(middlewares=[KafkaConcurrentProcessingMiddleware])
-app = FastStream(broker)
 
+# Or scope it to specific subscribers via a router
+router = KafkaRouter(middlewares=[KafkaConcurrentProcessingMiddleware])
 
-@app.on_startup
-async def on_startup(context: ContextRepo) -> None:
+@asynccontextmanager
+async def lifespan(_context: ContextRepo):
     await initialize_concurrent_processing(
-        context=context,
-        concurrency_limit=20,       # max concurrent tasks (minimum: 1)
-        commit_batch_size=100,      # commit after this many completed tasks
-        commit_batch_timeout_sec=5.0,  # or after this many seconds
+        context=broker.context,
+        concurrency_limit=20,         # max concurrent tasks (minimum: 1)
+        commit_batch_size=100,        # commit after this many completed tasks
+        commit_batch_timeout_sec=5.0, # or after this many seconds
     )
+    try:
+        yield
+    finally:
+        await stop_concurrent_processing(broker.context)
 
-
-@app.on_shutdown
-async def on_shutdown(context: ContextRepo) -> None:
-    await stop_concurrent_processing(context)
-
+app = AsgiFastStream(broker, lifespan=lifespan)
 
 @broker.subscriber("my-topic", group_id="my-group", ack_policy=AckPolicy.MANUAL)
 async def handle(msg: str) -> None:
-    # runs concurrently with other messages
     ...
+
+@router.subscriber("other-topic", group_id="other-group", ack_policy=AckPolicy.MANUAL)
+async def handle_other(msg: str) -> None:
+    ...
+
+broker.include_router(router)
 ```
 
 ## Core Concepts
@@ -114,7 +124,7 @@ Returns `True` if the `KafkaConcurrentHandler` stored in `context` is running an
 
 ### `KafkaConcurrentProcessingMiddleware`
 
-FastStream middleware class. Pass it to `KafkaBroker(middlewares=[...])` or `broker.add_middleware(...)`.
+FastStream middleware class. Pass it to `KafkaBroker(middlewares=[...])`, `broker.add_middleware(...)`, or scope it to a subset of subscribers via `KafkaRouter`. See Quick Start for usage examples.
 
 ## How It Works
 

tests/test_integration.py

@@ -1,9 +1,12 @@
 import asyncio
+import contextlib
 import typing
 import uuid
 
 from aiokafka.admin import AIOKafkaAdminClient, NewTopic
-from faststream.kafka import KafkaBroker
+from faststream import ContextRepo
+from faststream.asgi import AsgiFastStream
+from faststream.kafka import KafkaBroker, KafkaRouter
 from faststream.middlewares import AckPolicy
 
 from faststream_concurrent_aiokafka import (
@@ -360,3 +363,69 @@ async def handler_b2(msg: dict[str, int]) -> None:  # pragma: no cover
 
     assert replayed_a == [], f"topic_a messages replayed after clean stop: {replayed_a}"
     assert replayed_b == [], f"topic_b messages replayed after clean stop: {replayed_b}"
+
+
+async def test_middleware_on_router(kafka_bootstrap_servers: str) -> None:
+    """Middleware set on a KafkaRouter (not broker-level) routes messages through concurrent handler."""
+    processed: typing.Final[list[dict[str, int]]] = []
+    topic: typing.Final = _topic("router-mw")
+    # Plain broker — no broker-level middleware; middleware is scoped to the router
+    broker: typing.Final = KafkaBroker(kafka_bootstrap_servers)
+    router: typing.Final = KafkaRouter(middlewares=[KafkaConcurrentProcessingMiddleware])
+
+    @router.subscriber(topic, group_id="router-mw-group", auto_offset_reset="earliest", ack_policy=AckPolicy.MANUAL)
+    async def handler(msg: dict[str, int]) -> None:
+        processed.append(msg)
+
+    broker.include_router(router)
+
+    await _create_topic(kafka_bootstrap_servers, topic)
+    async with broker:
+        await broker.start()
+        await initialize_concurrent_processing(
+            context=broker.context, commit_batch_size=10, commit_batch_timeout_sec=5, concurrency_limit=5
+        )
+        await asyncio.sleep(CONSUMER_READY_SLEEP)
+        try:
+            await broker.publish({"id": 7}, topic=topic)
+            await asyncio.sleep(POLL_SLEEP)
+        finally:
+            await stop_concurrent_processing(broker.context)
+
+    assert len(processed) == 1
+    assert processed[0]["id"] == 7
+
+
+async def test_asgi_faststream_basic_processing(kafka_bootstrap_servers: str) -> None:
+    """AsgiFastStream lifespan initialises and stops concurrent processing correctly."""
+    processed: typing.Final[list[dict[str, int]]] = []
+    topic: typing.Final = _topic("asgi")
+    broker: typing.Final = _broker(kafka_bootstrap_servers)
+
+    @broker.subscriber(topic, group_id="asgi-group", auto_offset_reset="earliest", ack_policy=AckPolicy.MANUAL)
+    async def handler(msg: dict[str, int]) -> None:
+        processed.append(msg)
+
+    @contextlib.asynccontextmanager
+    async def lifespan(_context: ContextRepo) -> typing.AsyncIterator[None]:
+        # AsgiFastStream injects its own app-level context, which is separate from
+        # broker.context. Use broker.context explicitly so the middleware can find
+        # the handler via self.context.
+        await initialize_concurrent_processing(
+            context=broker.context, commit_batch_size=10, commit_batch_timeout_sec=5, concurrency_limit=5
+        )
+        try:
+            yield
+        finally:
+            await stop_concurrent_processing(broker.context)
+
+    app: typing.Final = AsgiFastStream(broker, lifespan=lifespan)
+
+    await _create_topic(kafka_bootstrap_servers, topic)
+    async with app.start_lifespan_context():
+        await asyncio.sleep(CONSUMER_READY_SLEEP)
+        await broker.publish({"id": 42}, topic=topic)
+        await asyncio.sleep(POLL_SLEEP)
+
+    assert len(processed) == 1
+    assert processed[0]["id"] == 42