Новый тип транспортировки и считывателя в скрученном виде

Question

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

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

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

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

Answer 1

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

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

Однако в случае с трубой вы должны быть в состоянии использовать и извлекать выгоду из IReactorFDSetметоды - addReader et al.

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

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

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

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