Новый транспортный и читательский тип в Twisted

Question

Я пытаюсь добавить новый транспорт в Twisted, который будет считывать данные из потока - либо из файла tail -f, либо из канала, но у меня есть некоторые проблемы с архитектурой Twisted.

У меня есть готовый транспорт (реализует ITransport) - он обрабатывает все открытия файлов. У меня есть готовые функции потоковой передачи / отложенные. Как мне собрать это сейчас? Я хотел бы сообщить о новых данных обратно в какой-то протокол dataReceived().

Конечно, я мог бы создать новый объект, который будет настраивать мониторы ввода / вывода с соответствующими обратными вызовами, регистрировать обратный вызов при остановке реактора (для закрытия файлов / протоколов) и запускать все вручную - но это " правильно"? Есть ли какая-нибудь более приятная абстракция, которую я мог бы использовать? Я видел reactor.connectWith(), но на самом деле он не дает много абстракции ...

Кроме того, как мне передать данные из моего считывателя в протокол? ITransport не определяет какой-либо интерфейс для него, хотя кажется, что именно ответственность за него несет транспорт.

Answer 1

Похоже, вы в основном разобрались, как это сделать. Возможно, вас заинтересует twisted.internet.fdesc.readFromFD, но это всего лишь несколько строк, и он не делает ничего особенно сложного (хотя вам и не нужно поддерживать несколько строк). Кроме того - да, в этом случае вы должны выполнять мониторинг ввода-вывода, потому что обычные файловые дескрипторы не поддерживаются select / poll / epoll (они всегда сообщаются как готовые, а не то, что вы хотите).

Была проделана некоторая работа по поддержке inotify в Twisted (http://twistedmatrix.com/trac/ticket/972), но она еще не завершена, поэтому она сейчас не будет вам непосредственно полезна (если вы не хотите помочь закончить ее и затем использовать это). Предполагая, что вы просто используете основанный на времени опрос, многое из того, что находится в реакторе, не сильно вам поможет, поскольку этот код ориентирован на использование предоставляемого системой API готовности (т. е. select / poll / epoll) для запуска событий.

Тем не менее, для случая с трубой вы должны быть в состоянии использовать и использовать методы IReactorFDSet - addReader и др.

Ваш транспорт для опроса по времени может по-прежнему выиграть от реализации ITransport - хотя я не уверен, как бы вы реализовали write для tail -f -подобного транспорта. Вы определенно выиграете от доставки данных через интерфейс IProtocol, поскольку это упрощает повторное использование кода. IProtocol.dataReceived - это именно то, как вы хотите передать данные из вашего считывателя (я думаю, это то же самое, что ваш транспорт , не так ли?). Это не определено в ITransport, потому что это метод, который вы вызываете для другого объекта, который не является транспортным.

reactor.connectWith, вероятно, ничего не купит. Как вы говорите, это не большая абстракция; Я бы сказал, что это скорее ошибка. :)

Не беспокойтесь слишком сильно о невозможности добавлять методы непосредственно в реактор. Свободная функция, которая принимает реактор в качестве параметра, также проста в использовании.

Для обратного вызова выключения, addReader на самом деле должно помочь вам в этом. Любой считыватель в реакторе во время останова будет вызывать connectionLost (часть IFileDescriptor). Вы должны реализовать это для очистки файлов и протокола.