def pipe(self, transformer):
if self.next:<EOL><INDENT>return<EOL><DEDENT>stream = Stream()<EOL>self.next = stream<EOL>stream.prev = self<EOL>self.transformer = transformer<EOL>transformer.stream = self<EOL>transformer.piped()<EOL>for file in self.files:<EOL><INDENT>future = asyncio.ensure_future(self.transformer.transform(file))<EOL>future.add_done_callback(self.handle_transform) <EOL><DEDENT>self.onpiped.set_result(None)<EOL>self.flush_if_ended()<EOL>return stream<EOL>
Pipe this stream to another.