Why I can not close channel?

[olivick salent]

### What version of gRPC and what language are you using?

python3.10.10


### What operating system (Linux, Windows,...) and version?

deepin v20.8(like ubuntu)

### What runtime / compiler are you using (e.g. python version or version of gcc)

python3.10.10


### What did you do?

My unit test :
proto( python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. example.proto ):

```protobuf

syntax = "proto3";

package example;

service BidirectionalService {
  rpc StreamMessages(stream Request) returns (stream Response);
}

message Request {
  string message = 1;
}

message Response {
  string message = 1;
}



```
my server:

```python

import grpc
from concurrent import futures
import time
import example_pb2
import example_pb2_grpc

class BidirectionalService(example_pb2_grpc.BidirectionalServiceServicer):
    def StreamMessages(self, request_iterator, context):
        for request in request_iterator:
            message = request.message
            response = f"Server received: {message}"
            yield example_pb2.Response(message=response)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    example_pb2_grpc.add_BidirectionalServiceServicer_to_server(BidirectionalService(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server started on port 50051")
    try:
        while True:
            time.sleep(86400)
    except KeyboardInterrupt:
        server.stop(0)

if __name__ == '__main__':
    serve()



```

client:

```python


import asyncio

import grpc
import example_pb2
import example_pb2_grpc

channel_options = [  # keep_alive
    ("grpc.keepalive_time_ms", 8000),
    ("grpc.keepalive_timeout_ms", 5000),
    ("grpc.http2.max_pings_without_data", 5),
    ("grpc.keepalive_permit_without_calls", 1),
]


async def send_msg(stub, st):
    await  st.write(example_pb2.Request(message="MESSAGE sub"))


async def stream(stub: example_pb2_grpc.BidirectionalServiceStub):
    st = stub.StreamMessages()
    await st.write(example_pb2.Request(message="MESSAGE - 10"))
    async for res in st: # it would not stop when channel.close.
        print(res)
        asyncio.create_task(send_msg(stub, st))
        await asyncio.sleep(1)


async def start(ip_addr):
    async with grpc.aio.insecure_channel(
            target=ip_addr, options=channel_options
    ) as channel:
        stub = example_pb2_grpc.BidirectionalServiceStub(channel)

        async def close_channel():
            await asyncio.sleep(5)
            await channel.close(None)

        asyncio.create_task(close_channel())
        await stream(stub)


def main():
    asyncio.run(start("localhost:50051"))


if __name__ == '__main__':
    main()


```


### What did you expect to see?

I want the program to end when close_channel executes.

### What did you see instead?

Channel was not be closed and go on receiving message.