output and are used in error messages
Sources must be file handles, open pipes or open sockets (or anything
- else that gives a reasonable fileno(), so help of the select module)
+ else that gives a reasonable fileno(), so help of the :py:mod:`select`
+ module)
Sources are not closed!
warn('poll replied "invalid request" err during polling '
'for {0}'
.format(desc))
+
# read line from source
line = source.readline()
:param proc_descs: None or list/tuple of same length as procs
:param dict subprocess_args: dict or args that are merged with
:py:data:`DEFAULT_SUBPROCESS_ARGS`
+
+ ..seealso:: could have tried to implement this as a nesting of many
+ individual context managers, see :py:mod:`contextlib`
"""
print('__init__')
# #############################################################################
# TESTING
+# (hard to create unittest -- would need separate threads to write to some
+# file /pipe/socket so can test the interactivity...)
# #############################################################################
from datetime import date, datetime as dt, timedelta as td
flwr = Follower(file_handle, 'syslog file', proc.stdout, 'tail syslog',
proc.stderr, 'tail stderr')
for n_lines, (source, desc, line, flags) in enumerate(flwr):
+ if flags:
+ print('received flags {0}'.format(flags_to_str(flags)))
test_syslog_line(n_lines, source, desc, line,
today_str, start_time)
start_time = dt.now() - td(seconds=time_diff_seconds)
with follow(syslog_file) as flwr:
for n_lines, (source, desc, line, flags) in enumerate(flwr):
+ if flags:
+ print('received flags {0}'.format(flags_to_str(flags)))
test_syslog_line(n_lines, source, desc, line,
today_str, start_time)
start_time = dt.now() - td(seconds=time_diff_seconds)
with follow(procs=(['tail', '-f', syslog_file], )) as flwr:
for n_lines, (source, desc, line, flags) in enumerate(flwr):
+ if flags:
+ print('received flags {0}'.format(flags_to_str(flags)))
test_syslog_line(n_lines, source, desc, line,
today_str, start_time)