Python неблокирующий Asyncio - PullRequest
Новая
       24

Python неблокирующий Asyncio

0 голосов
/ 12 июля 2020

Я пытаюсь создать класс, который позволяет мне помещать в него данные, пока открыто соединение через веб-сокет, но я не могу понять, как сделать основной вызов класса неблокирующим. Может ли кто-нибудь указать мне в правильном направлении?

Вот что у меня есть (удален посторонний код): 

class Audio_Sender:

    def __init__(self,IP_Address):
        self.Remote_IP_Address = IP_Address
        self.audio_queue = queue.Queue(10) #Max of 10 items

    async def Connect(self):
        uri = "ws://127.0.0.1:8765"
        async with websockets.connect(uri) as websocket:

            await websocket.send(json_voice_start)    
            
            while self.status == "Run":
                if not self.audio_queue.empty():
                    audio_data = self.audio_queue.get()
                    await websocket.send(audio_data)

            #pull any remaing data out:
            while not self.audio_queue.empty():
                audio_data = self.audio_queue.get()
                await websocket.send(audio_data)

            await websocket.send(json_voice_stop)

            voice_response = await websocket.recv()
            message = json.loads(voice_response)
            print("\t- " + message["result"])


    async def run_connect(self):
        task = asyncio.create_task(self.Connect())
        while not task.done():
            print("Task is not Done")
            await asyncio.sleep(1)
            
    def go(self):
        asyncio.run(self.run_connect())

#Create the Audio Sender
A = Audio_Sender("127.0.0.1","r")

#Put some data into it's queue
A.audio_queue.put(b"abc")
A.audio_queue.put(b"abc")
A.audio_queue.put(b"abc")

#Finished putting data in
A.status = "Done"

#Now send the data
#Ideally I would like to have the go part way through the queuing of
# data (above), but A.go() is blocking.. how to make it not blocking?
A.go()

1 Ответ

0 голосов
/ 15 июля 2020

Итак, поиграв с этим еще немного, я понял, что могу использовать поток для go (), который работает:

Обновленный код выглядит следующим образом: 

#Create the Audio Sender
A = Audio_Sender("127.0.0.1","r")

s = threading.Thread(target=A.go,args=())
s.start()

i=0
while i<10:
    #Put some data into it's queue
    A.audio_queue.put(b"abc")
    i = i +1
    print(i)



#Finished putting data in
A.status = "Done"
...