强制coroutine异步结束

Question

我正在尝试理解python中的协同机制，但我很难理解如何结束其中一个。

我试图理解以下代码：

async def send_recieve():
    async with websockets.connect(*parameters*) as _ws:
            async def send():
                while True:
                    #function send...
            async def recieve():
                while True:
                    #function recieve...
                    if #condition met:
                        break
            send_result, receive_result = await asyncio.gather(send(), receive())
asyncio.run(send_receive())

当条件满足时，接收函数就结束了，但是发送函数继续工作，我不能结束整个send_recieve异步。我试着把代码总结得更清楚，如果更容易理解的话，我可以分享整个版本。当接收函数结束时，我发现我错过了send函数中的一个条件，但我不知道如何编写它。

如果我试图在满足条件的情况下添加loop.stop()，它会引发错误"RuntimeError: Event循环在未来完成之前停止“。

整个守则如下：

async def send_receive():
    print(f'Connecting websocket to url ${URL}')
    async with websockets.connect(
        URL,
        extra_headers=(("Authorization", auth_key),),
        ping_interval=5,
        ping_timeout=20
    ) as _ws:
        await asyncio.sleep(0.3)
        print("Receiving SessionBegins ...")
        session_begins = await _ws.recv()
        print(session_begins)
        print("Sending messages ...")

        async def send():
            while True:
                try:
                    data = stream.read(FRAMES_PER_BUFFER)
                    data = base64.b64encode(data).decode("utf-8")
                    json_data = json.dumps({"audio_data":str(data)})
                    await _ws.send(json_data)
                except websockets.exceptions.ConnectionClosedError as e:
                    print(e)
                    assert e.code == 4008
                    break
                except Exception as e:
                    assert False, "Not a websocket 4008 error"
                await asyncio.sleep(0.01)
          
            return True
      
        async def receive():
            while True:
                try:
                    result_str = await _ws.recv()
                    majtext = json.loads(result_str)['text']
                    print(majtext)

                except websockets.exceptions.ConnectionClosedError as e:
                    print(e)
                    assert e.code == 4008
                    return
                except Exception as e:
                    assert False, "Not a websocket 4008 error"

                if json.loads(result_str)['message_type'] == 'FinalTranscript':
                    break   
      
        send_result, receive_result = await asyncio.gather(send(), receive())

loop = asyncio.get_event_loop()
loop.run_until_complete(send_receive())

Answer 1

如何告诉send函数在receive函数结束时结束

asyncio.gather()等待这两个函数完成。相反，可以通过替换以下两个函数来等待任何一个函数完成：

send_result, receive_result = await asyncio.gather(send(), receive())

通过以下方式：

await asyncio.wait(
    [asyncio.create_task(send()), asyncio.create_task(receive())],
    return_when=asyncio.FIRST_COMPLETED
)

(请注意，从send和receive检索到的“结果”没有意义，因为这两个函数都没有返回有用的值。)