aiokq/misc/kqtest.py

#
# Test to figure out how to trigger the event.
#
# Key points:
#	CLEAR is needed if you want to reset things
#	fp.flush is needed otherwise kernel does not see the write
#

import select

kq = select.kqueue()

print('begining')

with open('somefile.txt', 'w+') as fp, open('somefile.txt', 'r') as rfp:
	print('fno:', fp.fileno())
	print('rfno:', rfp.fileno())

	kevent = select.kevent(rfp, select.KQ_FILTER_VNODE, flags=select.KQ_EV_ADD|select.KQ_EV_ENABLE|select.KQ_EV_CLEAR|0x40, fflags=select.KQ_NOTE_EXTEND)

	r = kq.control([ kevent ], 1, 0)
	print(repr(r))

	fp.write('foo!')
	fp.flush()

	r = kq.control([], 1, 0)

	print('r:', repr(r))

	print('rfp.tell:', rfp.tell())
	print('rfp.read:', repr(rfp.read()))

	r = kq.control([], 1, 0)

	print(repr(r))

	fp.write('foo!')
	fp.flush()

	r = kq.control([], 1, 0)

	print(repr(r))

	print('fp.tell:', fp.tell())