1 import time
   2 import re
   3 
   4 def follow(fname):
   5     f = file(fname)
   6     f.seek(0,2) # go to the end
   7     while True:
   8         l = f.readline()
   9         if not l: # no data
  10             time.sleep(.1)
  11         else:
  12             yield l
  13 
  14 def grep(lines, pattern):
  15     regex = re.compile(pattern)
  16     for l in lines:
  17         if regex.match(l):
  18             yield l
  19 
  20 def printer(lines):
  21     for l in lines:
  22         print l.strip()
  23 
  24 f = follow('/var/log/system.log')
  25 g = grep(f, ".*pants.*")
  26 p = printer(g)
  27 
  28 for i in p:
  29     pass

   1 import time
   2 import re
   3 from functools import wraps
   4 
   5 
   6 def coroutine(func):
   7     @wraps(func)
   8     def thing(*args, **kwargs):
   9         gen = func(*args, **kwargs)
  10         gen.next() # advance to the first yield
  11         return gen
  12     return thing
  13 
  14 @coroutine
  15 def follow(fname, next):
  16     f = file(fname)
  17     f.seek(0,2) # go to the end
  18     while True:
  19         l = f.readline()
  20         if not l: # no data
  21             time.sleep(.1)
  22         else:
  23             next.send(l)
  24 
  25 @coroutine
  26 def grep(pattern, next):
  27     regex = re.compile(pattern)
  28     while True:
  29         l = yield
  30         if regex.match(l):
  31             next.send(l)
  32 
  33 @coroutine
  34 def printer():
  35     while True:
  36         l = yield
  37         print l.strip()
  38 
  39 
  40 p = printer()
  41 g = grep('.*pants.*', p)
  42 f = follow('/var/log/system.log', g)

   1 import greenlet
   2 import time
   3 import re
   4 
   5 def follow(fname, next):
   6     # setup
   7     f = file(fname)
   8     f.seek(0,2) # go to the end
   9     # do stuff
  10     while True:
  11         l = f.readline()
  12         if not l: # no data
  13             time.sleep(.1)
  14         else:
  15             next.switch(l)
  16 
  17 def grep(pattern, next):
  18     # setup
  19     regex = re.compile(pattern)
  20 
  21     def do_stuff(l):
  22         parent = greenlet.getcurrent().parent
  23         while True:
  24             if regex.match(l):
  25                 l = next.switch(l)
  26             else:
  27                 l = parent.switch() # subtle!
  28 
  29     return do_stuff
  30 
  31 def printer(l):
  32     # no setup
  33     parent = greenlet.getcurrent().parent
  34     # do stuff
  35     while True:
  36         print l.strip()
  37         l = parent.switch()
  38 
  39 p = greenlet.greenlet(printer)
  40 g = greenlet.greenlet(grep(".*pants.*", p))
  41 follow("/var/log/system.log", g)

   1 import time
   2 import re
   3 
   4 import Axon
   5 from Kamaelia.Chassis.Pipeline import Pipeline
   6 
   7 # threaded due to the time.sleep() call
   8 # No yield since a threaded component
   9 class Follow(Axon.ThreadedComponent.threadedcomponent):
  10     def __init__(self, fname, **argv):
  11         self.fname = fname
  12         super(Follow,self).__init__(**argv)
  13     def main(self):
  14         f = file(self.fname)
  15         f.seek(0,2) # go to the end
  16         while not self.dataReady("control"):
  17             l = f.readline()
  18             if not l: # no data
  19                 time.sleep(.1)
  20             else:
  21                 self.send(l, "outbox")
  22 
  23         self.send(self.recv("control"), "signal")
  24 
  25 class Grep(Axon.Component.component):
  26     # Default pattern, override in constructor with pattern="some pattern"
  27     # See below
  28     pattern = "."
  29     def main(self):
  30         regex = re.compile(self.pattern)
  31         while not self.dataReady("control"):
  32            for l in self.Inbox("inbox"):
  33                if regex.match(l):
  34                    self.send(l, "outbox")
  35            self.pause()
  36            yield 1
  37         self.send(self.recv("control"), "signal")
  38 
  39 class Printer(Axon.Component.component):
  40     def main(self):
  41         while not self.dataReady("control"):
  42             for l in self.Inbox("inbox"):
  43                 print l.strip()
  44             self.pause()
  45             yield 1
  46         self.send(self.recv("control"), "signal")
  47 
  48 Pipeline(
  49     Follow('/var/log/system.log'),
  50     Grep(".*pants.*"),
  51     Printer(),
  52 ).run()

   1 from twisted.protocols.basic import LineReceiver
   2 from twisted.python import log
   3 
   4 SLOW_INTERVAL = 1.0
   5 FAST_INTERVAL = 0.001
   6 SEEK_END = 2
   7 BLOCKSIZE = 8192
   8 
   9 class TailTransport(object):
  10   def __init__(self, fileobj, protocol):
  11       self.fileobj = fileobj
  12       self.protocol = protocol
  13       self.disconnecting = False
  14 
  15   def start(self, clock):
  16       self.clock = clock
  17       self.fileobj.seek(0, SEEK_END)
  18       self.protocol.makeConnection(self)
  19       self.tick()
  20 
  21   def tick(self):
  22       anyData = self.fileobj.read(BLOCKSIZE)
  23       try:
  24           self.protocol.dataReceived(anyData)
  25       except:
  26           log.err()
  27       if anyData:
  28           interval = FAST_INTERVAL
  29       else:
  30           interval = SLOW_INTERVAL
  31       self.clock.callLater(interval, self.tick)
  32 
  33 class Grep(LineReceiver):
  34   delimiter = '\n'
  35   def __init__(self, term):
  36       self.term = term
  37 
  38   def lineReceived(self, line):
  39       if self.term in line:
  40           print line.rstrip("\n")
  41 
  42 def main():
  43   from twisted.internet import reactor
  44   TailTransport(file("/var/log/syslog", "rb"),
  45                 Grep("pants")).start(reactor)
  46   reactor.run()
  47 
  48 main()

   1 import fibra
   2 import re
   3 
   4 def tail(f, output):
   5     f.seek(0,2)
   6     while True:
   7         line = f.readline()
   8         yield output.push(line) if line else 0.1 #push line, or sleep.
   9 
  10 def grep(pattern, input, output):
  11     regex = re.compile(pattern)
  12     while True:
  13         line = yield input.pop()
  14         if regex.match(line):
  15             yield output.push(line) 
  16 
  17 def printer(input):
  18     while True:
  19         line = yield input.pop()
  20         print line.strip()
  21     
  22 schedule = fibra.schedule()
  23 schedule.install(tail(open("/var/log/syslog.log","r"), fibra.Tube("T2G")))
  24 schedule.install(grep(".*pants.*", fibra.Tube("T2G"), fibra.Tube("G2P")))
  25 schedule.install(printer(fibra.Tube("G2P")))
  26 schedule.run()

   1 import stackless
   2 import time
   3 import re
   4 
   5 @stackless.tasklet
   6 def tail(f, output):
   7     f.seek(0,2)
   8     while True:
   9         line = f.readline()
  10         if line:
  11             output.send(line)
  12         else:
  13             time.sleep(0.1)
  14 
  15 @stackless.tasklet
  16 def grep(pattern, input, output):
  17     regex = re.compile(pattern)
  18     while True:
  19         line = input.receive()
  20         if regex.match(line):
  21             output.send(line)
  22 
  23 @stackless.tasklet
  24 def printer(input):
  25     while True:
  26         line = input.receive()
  27         print line.strip()
  28 
  29 T2G = stackless.channel()
  30 G2P = stackless.channel()
  31 tail(open("/var/log/syslog.log","r"), T2G)
  32 grep(".*pants.*", T2G, G2P)
  33 printer(G2P)
  34 stackless.run()

   1 import re
   2 from time import sleep
   3 
   4 from circuits import Event, Component, Manager, Thread
   5 
   6 SEEK_END = 2
   7 BLOCKSIZE = 8192
   8 
   9 LINESEP = re.compile("\r?\n")
  10 
  11 def splitLines(s, buffer):
  12     lines = LINESEP.split(buffer + s)
  13     return lines[:-1], lines[-1]
  14 
  15 class Follow(Thread):
  16 
  17     def __init__(self, fd):
  18         super(Follow, self).__init__()
  19         self.fd = fd
  20         self.fd.seek(0, SEEK_END)
  21 
  22     def stopped(self, manager):
  23         if not manager == self:
  24             self.stop()
  25 
  26     def run(self):
  27         while self.alive:
  28             data = self.fd.read(BLOCKSIZE)
  29             if data:
  30                 self.push(Event(data), "read", self.channel)
  31                 sleep(0.01)
  32             else:
  33                 sleep(0.1)
  34 
  35 class LineBuffer(Component):
  36 
  37     def __init__(self):
  38         super(LineBuffer, self).__init__()
  39         self._data = ""
  40 
  41     def read(self, data):
  42         lines, self._data = splitLines(data, self._data)
  43         for line in lines:
  44             self.push(Event(line), "line", self.channel)
  45 
  46 class Grep(Component):
  47 
  48     def __init__(self, pattern):
  49         super(Grep, self).__init__()
  50         self._pattern = pattern
  51 
  52     def line(self, line):
  53         if self._pattern in line:
  54             print line
  55 
  56 m = Manager()
  57 f = Follow(file("/tmp/foo"))
  58 f.register(m)
  59 LineBuffer().register(m)
  60 Grep("pants").register(m)
  61 
  62 f.start()
  63 m.run()
  64 
  65     * A Thread(ed) Component (Follow) is used to continuously read from the given file and when data is available propagates an event containing the newly read data to a "read" channel.
  66     * A LineBuffer? Component listener for events on the "read" channel, takes the newly read data from the Follow Component and splits this buffer up one line at a time propagating a single line at a time onto a "line" channel.
  67     * Finally the Grep Component listens for events on the "line" channel and searches for the given pattern, if found will print the line. 
  68 
  69 Asynchronous I/O
  70 
  71 This second implementation uses the standard asynchronous File Component available in circuits.io.
  72 
  73 import re
  74 
  75 from circuits.io import File
  76 from circuits import Event, Component
  77 
  78 SEEK_END = 2
  79 
  80 LINESEP = re.compile("\r?\n")
  81 
  82 def splitLines(s, buffer):
  83     lines = LINESEP.split(buffer + s)
  84     return lines[:-1], lines[-1]
  85 
  86 class Follow(Component):
  87 
  88     def __init__(self, filename):
  89         super(Follow, self).__init__()
  90         fd = File(filename, "r")
  91         fd.seek(0, SEEK_END)
  92         fd.register(self)
  93 
  94 class LineBuffer(Component):
  95 
  96     def __init__(self):
  97         super(LineBuffer, self).__init__()
  98         self._data = ""
  99 
 100     def read(self, data):
 101         lines, self._data = splitLines(data, self._data)
 102         for line in lines:
 103             self.push(Event(line), "line", self.channel)
 104 
 105 class Grep(Component):
 106 
 107     def __init__(self, pattern):
 108         super(Grep, self).__init__()
 109         self._pattern = pattern
 110 
 111     def line(self, line):
 112         if self._pattern in line:
 113             print line
 114 
 115 (Follow("/tmp/foo") + LineBuffer() + Grep("pants")).run()