New transport and reader type in Twisted

Question

I'm trying to add a new transport to Twisted, which will read data from a stream - either a file in a tail -f way, or from a pipe, but I have some problems with Twisted architecture.

I've got the transport itself (implements ITransport) ready - it handles all file opening. I've got streaming functions/deferreds ready. How do I put it together now? I'd like to report the new data back to some protocol's dataReceived().

I could of course create a new object that will set up the I/O monitors with proper callbacks, register a callback on reactor shutting down (to close the files / protocols) and start everything up manually - but is that "the right way"? Is there any nicer abstraction I could use? I've seen reactor.connectWith(), but it doesn't really provide much of an abstraction...

Also - how am I supposed to pass the data from my reader to the protocol? ITransport doesn't define any interface for it, even though it seems like exactly the transport's responsibility.

Answer 1

It sounds like you've mostly figured out how to do this. You might be interested in twisted.internet.fdesc.readFromFD, but it's only a few lines long and it's not doing anything particularly complicated (it's a few lines you don't have to maintain, though). Aside from that - yes, you have to do the I/O monitoring in this case, because regular file descriptors aren't supported by select/poll/epoll (they always get reported as ready, not what you want).

Some work has been done on supporting inotify in Twisted (http://twistedmatrix.com/trac/ticket/972) but this isn't complete yet, so it's not going to be directly useful to you now (unless you want to help finish it and then use it). Assuming you just use time-based polling, much of what's in the reactor isn't going to help you out much, since that code is focused on using a system-provided readiness API (ie, select/poll/epoll) to trigger events.

For the pipe case, though, you should be able to use and benefit from IReactorFDSet's methods - addReader et al.

Your time-based polling transport may still benefit from implementing ITransport - although I'm not sure how you would implement write for a tail -f-like transport. You will definitely benefit from having your transport deliver data via the IProtocol interface, since this simplifies code-reuse. IProtocol.dataReceived is exactly how you want to pass data from your reader (I think that's the same as your transport, isn't it?). This isn't defined on ITransport because it's a method you call on some other object which is not the transport.

reactor.connectWith probably isn't going to buy you anything. As you say, it's not much of an abstraction; I'd say it's more of a mistake. :)

Don't worry too much about not being able to add methods directly to the reactor. A free-function which accepts a reactor as a parameter is just as easy to use.

For the shutdown callback, addReader should actually get you most of the way there. Any reader in the reactor at shutdown time will have connectionLost called on it (part of IFileDescriptor). You should implement this to clean up the files and protocol.