ACIL FM

Nama	Ukuran	Aksi
io	-	Open
__pycache__	-	Open
base.py	2.02 MB	View DL Edit
ebtables.py	9.19 MB	View DL Edit
fw.py	56 MB	View DL Edit
fw_config.py	48.18 MB	View DL Edit
fw_direct.py	22.46 MB	View DL Edit
fw_helper.py	1.79 MB	View DL Edit
fw_icmptype.py	2.43 MB	View DL Edit
fw_ifcfg.py	2.5 MB	View DL Edit
fw_ipset.py	9.85 MB	View DL Edit
fw_nm.py	7.02 MB	View DL Edit
fw_policies.py	2.74 MB	View DL Edit
fw_policy.py	72.82 MB	View DL Edit
fw_service.py	1.6 MB	View DL Edit
fw_transaction.py	5.73 MB	View DL Edit
fw_zone.py	42.47 MB	View DL Edit
helper.py	804 B	View DL Edit
icmp.py	3.03 MB	View DL Edit
ipset.py	11.93 MB	View DL Edit
ipXtables.py	61.6 MB	View DL Edit
logger.py	30.31 MB	View DL Edit
modules.py	3.73 MB	View DL Edit
nftables.py	97.7 MB	View DL Edit
prog.py	1.47 MB	View DL Edit
rich.py	34.32 MB	View DL Edit
watcher.py	3.14 MB	View DL Edit
__init__.py	0 B	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/firewall/core/watcher.py

# -*- coding: utf-8 -*-
#
# Copyright (C) 2012-2016 Red Hat, Inc.
#
# Authors:
# Thomas Woerner <twoerner@redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

__all__ = [ "Watcher" ]

from gi.repository import Gio, GLib

class Watcher(object):
    def __init__(self, callback, timeout):
        self._callback = callback
        self._timeout = timeout
        self._monitors = { }
        self._timeouts = { }
        self._blocked = [ ]

def add_watch_dir(self, directory):
        gfile = Gio.File.new_for_path(directory)
        self._monitors[directory] = gfile.monitor_directory(\
            Gio.FileMonitorFlags.NONE, None)
        self._monitors[directory].connect("changed", self._file_changed_cb)

def add_watch_file(self, filename):
        gfile = Gio.File.new_for_path(filename)
        self._monitors[filename] = gfile.monitor_file(\
            Gio.FileMonitorFlags.NONE, None)
        self._monitors[filename].connect("changed", self._file_changed_cb)

def get_watches(self):
        return self._monitors.keys()

def has_watch(self, filename):
        return filename in self._monitors

def remove_watch(self, filename):
        del self._monitors[filename]

def block_source(self, filename):
        if filename not in self._blocked:
            self._blocked.append(filename)

def unblock_source(self, filename):
        if filename in self._blocked:
            self._blocked.remove(filename)

def clear_timeouts(self):
        for filename in list(self._timeouts.keys()):
            GLib.source_remove(self._timeouts[filename])
            del self._timeouts[filename]

def _call_callback(self, filename):
        if filename not in self._blocked:
            self._callback(filename)
        del self._timeouts[filename]

def _file_changed_cb(self, monitor, gio_file, gio_other_file, event):
        filename = gio_file.get_parse_name()
        if filename in self._blocked:
            if filename in self._timeouts:
                GLib.source_remove(self._timeouts[filename])
                del self._timeouts[filename]
            return

if event == Gio.FileMonitorEvent.CHANGED or \
                event == Gio.FileMonitorEvent.CREATED or \
                event == Gio.FileMonitorEvent.DELETED or \
                event == Gio.FileMonitorEvent.ATTRIBUTE_CHANGED:
            if filename in self._timeouts:
                GLib.source_remove(self._timeouts[filename])
                del self._timeouts[filename]
            self._timeouts[filename] = GLib.timeout_add_seconds(\
                self._timeout, self._call_callback, filename)

Batal