Commit 263f5f0d authored by Olivier Duchateau's avatar Olivier Duchateau

Update import of socket/ directory

parent d6e3ac52
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: 0BSD
# Olivier Duchateau <duchateau.olivier@gmail.com>
#
'''
Simple script which monitors events from devd daemon. It is
useful with removable media (such USB key, µ SD and so on).
It requires GObject Introspection.
'''
import sys
import gi
gi.require_version('GLib', '2.0')
gi.require_version('Gio', '2.0')
from gi.repository import GLib, Gio
def read_ready_cb(src, result, user_data):
main_loop = user_data
res = src.read_finish(result)
if res == -1:
print('Error')
main_loop.quit()
elif res == 0:
print('End')
main_loop.quit()
else:
buf = GLib.ByteArray.new_take(bytes(4096))
src.read_async(buf, GLib.PRIORITY_DEFAULT, None,
read_ready_cb, main_loop)
def request_read_cb(src, result, user_data):
main_loop = user_data
conn = src.connect_finish(result)
if conn is not None:
# It is an Gio.InputStream object
stream = conn.get_input_stream()
buf = GLib.ByteArray.new_take(bytes(4096))
stream.read_async(buf, GLib.PRIORITY_DEFAULT, None,
read_ready_cb, main_loop)
print(buf)
else:
print('Error')
main_loop.quit()
def main(main_loop):
if sys.platform.startswith('freebsd'):
p = '/var/run/devd.seqpacket.pipe'
socket_type = Gio.SocketType.SEQPACKET
addr = Gio.UnixSocketAddress.new(p)
client = Gio.SocketClient.new()
client.set_socket_type(socket_type)
client.connect_async(addr, None, request_read_cb, main_loop)
else:
print('Platform not supported')
if __name__ == '__main__':
try:
loop = GLib.MainLoop.new(None, True)
main(loop)
loop.run()
except KeyboardInterrupt:
print('Keyboard has been caught')
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment