test-subprocess-source.py

#!/usr/bin/env python
# -*- coding: utf-8 vi:noet
# Example of data source using subprocess, python 2 & 3 compatible

"""
This illustrates in a short example, a solution to a few caveats:

- read() that can hang if the subprocess closes
- select() that knows nothing about the subprocess closing
"""

import sys, io, os, re, time, socket, subprocess, fcntl, select

class Source(object):
	def __init__(self):
		self._pipe = os.pipe()

		cmd = [
		 sys.executable, sys.argv[0], "test", "%d" % (self._pipe[1]),
		]

		fl = fcntl.fcntl(self._pipe[0], fcntl.F_GETFL)
		fcntl.fcntl(self._pipe[0], fcntl.F_SETFL, fl | os.O_NONBLOCK)

		kw = dict()
		if sys.hexversion >= 0x03000000:
			kw["pass_fds"] = (self._pipe[1],)
		else:
			kw["close_fds"] = False

		self._proc = subprocess.Popen(cmd, **kw)

	def __del__(self):
		try:
			self._proc.kill()
		except OSError:
			pass

	def read(self, bufsize=4096):
		#print("Select")
		i, o, e = select.select([self._pipe[0]], [], [], 1e-3)
		if not i:
			a = os.waitpid(self._proc.pid, os.WNOHANG) 
			pid, err = a
			if pid != 0:
				return None
			else:
				return b""
		return os.read(self._pipe[0], bufsize)

if __name__ == "__main__":
	if len(sys.argv) == 3 and sys.argv[1] == "test":
		fd = int(sys.argv[2])
		for i in range(30):
			print("pushed")
			os.write(fd, b" "*4096)
		raise SystemExit(0)

	src = Source()
	while True:
		data = src.read()
		if data is None:
			break
		if len(data) == 0:
			continue
		print("pulled %d" % len(data))