Как завершить вычитывание генератора из другого экземпляра программы?

Question

[Илья Рябыкин]

Всем здравствуйте.

Работаю с paramiko и channels (web-socket) в Django.

Реализован класс для создания подключения и выполнения комманд на сервере через paramiko.Transport.
Экземляр этого класса в данном контексте - executer.
Создаю экземпляр класса - executer, выполняю команду, результат выполнения которой поступает в self.session.recv (который я собираюсь вычитывать через следующий метод-генератор).
Метод-генератор из класса executer:

def gen(self):
    while not self.session.exit_status_ready():
        yield self.session.recv(self.RECV_BYTES).decode("utf-8") \
              or self.session.recv_stderr(self.RECV_BYTES).decode("utf-8")

Выполняю команду "tail -f textfile".
Вычитываю из генератора постоянно поступающие данные (никогда не прервётся) и отправляю через web-socket результат:
Экземпляр класса WebsocketConsumer:

for message in self.executer.gen:
            self.send(text_data=json.dumps({
                'message': str(message)
            }))

Проблема в том, что данный генератор всегда будет находиться в ожидании, т.к. self.session.exit_status_ready() никогда не вернёт True (в случает с командой "tail -f textfile").
ВОПРОС:
Нахожусь в поиске решения, как завершить вычитывание из генератора из вне (например, через веб-сокет, вызовом какой-то функции которая подключится к self.session и вызовет метод close())

Какие ещё мысли?
Худшая:
В генераторе сделать вычитывание из базы (по какому-либо уникальному ключу выполняемой команды и если там "done") - завершать выполнение генератора.
Минусы: много запросов в базу; если выполняемая команда ничего не возвращает - не срабатывает yield и в базу мы не смотрим -> завершение так же не произойдёт
Не реализуемая:
Вызывать цикл вычитывания из генератора асинхронно через asyncio.Future и завершать выполнение через close. С использованием веб-сокета так же невозможно, т.к. невозможно будет обратиться к выполняемому контексту.
и т.п.
Недобитая:
Сохранять через pickle экземпляр объекта paramiko.Transport для выполняемой команды, в случае необходимости подгружать его и завершать сессию (не получилось сохранить объект из-за объектов с lock'ами) да и вообще не пробовал, возможно ли это..

Answer 1

[santaatnas]

Вам нужен второй поток, который будет слушать соединение (TCP, HTTP, Web Socket - не важно) и по обращению к нему - вызывайте close()

Comments

[Илья Рябыкин]

Вы имеете в виду асинхронно вызывать вычитывание из генератора?
Чтобы оно там себе выполнялось, а слушалка при получении сообщения будет закрывать paramiko.Transport?
Вполне решение, спасибо.

[Илья Рябыкин]

К сожалению не получается реализовать, несколько потоков работают с разными контекстами класса.
Перешёл на асинхронный сокет, всё работает отлично с коммандами, возвращающими ответ сразу, так же можно завершать сессию когда угодно из вне.
НО, для "долгоиграющих" комманд (выполнение комплексного скрипта с множественными echo, и т.п.) возникает следующее: await self.send <...> ждёт когда асинхронный цикл завершится и только потом отдаёт все сообщения из генератора. Это уже связано с внутрянкой channels, но как быть?)

async def receive(self, text_data):
    text_data_json = json.loads(text_data)
    self.executer.func.execute(text_data_json['command'])
    async for message in self.executer.func.gen:
        print(message) # выводит сообщения
        await self.send(text_data=json.dumps({
            'message': str(message)
        }))