We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Re-building w/ symbolic links stopped working, regression after #2385
Since a444c43 in master using the local development server via `mkdocs serve` updating files that are symbolically linked is not triggering to rebuild (and therefore not reloading browser tabs).

On first glance this is due to the switch to watchdog for detecting file-system changes which needs more guidance to handle this file-type.

Preparing a PR with a patch.

Ref: a444c43474f91dea089922dd8fb188d1db3a4535

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mkdocs/livereload/__init__.py`
Content:
```
1 import functools
2 import io
3 import logging
4 import mimetypes
5 import os
6 import os.path
7 import pathlib
8 import re
9 import socketserver
10 import threading
11 import time
12 import warnings
13 import wsgiref.simple_server
14 
15 import watchdog.events
16 import watchdog.observers
17 
18 
19 class _LoggerAdapter(logging.LoggerAdapter):
20     def process(self, msg, kwargs):
21         return time.strftime("[%H:%M:%S] ") + msg, kwargs
22 
23 
24 log = _LoggerAdapter(logging.getLogger(__name__), {})
25 
26 
27 class LiveReloadServer(socketserver.ThreadingMixIn, wsgiref.simple_server.WSGIServer):
28     daemon_threads = True
29     poll_response_timeout = 60
30 
31     def __init__(
32         self,
33         builder,
34         host,
35         port,
36         root,
37         mount_path="/",
38         build_delay=0.25,
39         shutdown_delay=0.25,
40         **kwargs,
41     ):
42         self.builder = builder
43         self.server_name = host
44         self.server_port = port
45         self.root = os.path.abspath(root)
46         self.mount_path = ("/" + mount_path.lstrip("/")).rstrip("/") + "/"
47         self.url = f"http://{self.server_name}:{self.server_port}{self.mount_path}"
48         self.build_delay = build_delay
49         self.shutdown_delay = shutdown_delay
50         # To allow custom error pages.
51         self.error_handler = lambda code: None
52 
53         super().__init__((host, port), _Handler, **kwargs)
54         self.set_app(self.serve_request)
55 
56         self._wanted_epoch = _timestamp()  # The version of the site that started building.
57         self._visible_epoch = self._wanted_epoch  # Latest fully built version of the site.
58         self._epoch_cond = threading.Condition()  # Must be held when accessing _visible_epoch.
59 
60         self._to_rebuild = {}  # Used as an ordered set of functions to call.
61         self._rebuild_cond = threading.Condition()  # Must be held when accessing _to_rebuild.
62 
63         self._shutdown = False
64         self.serve_thread = threading.Thread(target=lambda: self.serve_forever(shutdown_delay))
65         self.observer = watchdog.observers.Observer(timeout=shutdown_delay)
66 
67     def watch(self, path, func=None, recursive=True):
68         """Add the 'path' to watched paths, call the function and reload when any file changes under it."""
69         path = os.path.abspath(path)
70         if func in (None, self.builder):
71             func = self.builder
72         else:
73             warnings.warn(
74                 "Plugins should not pass the 'func' parameter of watch(). "
75                 "The ability to execute custom callbacks will be removed soon.",
76                 DeprecationWarning,
77                 stacklevel=2,
78             )
79 
80         def callback(event, allowed_path=None):
81             if isinstance(event, watchdog.events.DirCreatedEvent):
82                 return
83             if allowed_path is not None and event.src_path != allowed_path:
84                 return
85             # Text editors always cause a "file close" event in addition to "modified" when saving
86             # a file. Some editors also have "swap" functionality that keeps writing into another
87             # file that's never closed. Prevent such write events from causing a rebuild.
88             if isinstance(event, watchdog.events.FileModifiedEvent):
89                 # But FileClosedEvent is implemented only on Linux, otherwise we mustn't skip this:
90                 if type(self.observer).__name__ == "InotifyObserver":
91                     return
92             log.debug(str(event))
93             with self._rebuild_cond:
94                 self._to_rebuild[func] = True
95                 self._rebuild_cond.notify_all()
96 
97         dir_handler = watchdog.events.FileSystemEventHandler()
98         dir_handler.on_any_event = callback
99 
100         seen = set()
101 
102         def schedule(path):
103             seen.add(path)
104             if os.path.isfile(path):
105                 # Watchdog doesn't support watching files, so watch its directory and filter by path
106                 handler = watchdog.events.FileSystemEventHandler()
107                 handler.on_any_event = lambda event: callback(event, allowed_path=path)
108 
109                 parent = os.path.dirname(path)
110                 log.debug(f"Watching file '{path}' through directory '{parent}'")
111                 self.observer.schedule(handler, parent)
112             else:
113                 log.debug(f"Watching directory '{path}'")
114                 self.observer.schedule(dir_handler, path, recursive=recursive)
115 
116         schedule(os.path.realpath(path))
117 
118         def watch_symlink_targets(path_obj):  # path is os.DirEntry or pathlib.Path
119             if path_obj.is_symlink():
120                 # The extra `readlink` is needed due to https://bugs.python.org/issue9949
121                 target = os.path.realpath(os.readlink(os.fspath(path_obj)))
122                 if target in seen or not os.path.exists(target):
123                     return
124                 schedule(target)
125 
126                 path_obj = pathlib.Path(target)
127 
128             if path_obj.is_dir() and recursive:
129                 with os.scandir(os.fspath(path_obj)) as scan:
130                     for entry in scan:
131                         watch_symlink_targets(entry)
132 
133         watch_symlink_targets(pathlib.Path(path))
134 
135     def serve(self):
136         self.observer.start()
137 
138         log.info(f"Serving on {self.url}")
139         self.serve_thread.start()
140 
141         self._build_loop()
142 
143     def _build_loop(self):
144         while True:
145             with self._rebuild_cond:
146                 while not self._rebuild_cond.wait_for(
147                     lambda: self._to_rebuild or self._shutdown, timeout=self.shutdown_delay
148                 ):
149                     # We could have used just one wait instead of a loop + timeout, but we need
150                     # occasional breaks, otherwise on Windows we can't receive KeyboardInterrupt.
151                     pass
152                 if self._shutdown:
153                     break
154                 log.info("Detected file changes")
155                 while self._rebuild_cond.wait(timeout=self.build_delay):
156                     log.debug("Waiting for file changes to stop happening")
157 
158                 self._wanted_epoch = _timestamp()
159                 funcs = list(self._to_rebuild)
160                 self._to_rebuild.clear()
161 
162             for func in funcs:
163                 func()
164 
165             with self._epoch_cond:
166                 log.info("Reloading browsers")
167                 self._visible_epoch = self._wanted_epoch
168                 self._epoch_cond.notify_all()
169 
170     def shutdown(self):
171         self.observer.stop()
172         with self._rebuild_cond:
173             self._shutdown = True
174             self._rebuild_cond.notify_all()
175 
176         if self.serve_thread.is_alive():
177             super().shutdown()
178             self.serve_thread.join()
179             self.observer.join()
180 
181     def serve_request(self, environ, start_response):
182         try:
183             result = self._serve_request(environ, start_response)
184         except Exception:
185             code = 500
186             msg = "500 Internal Server Error"
187             log.exception(msg)
188         else:
189             if result is not None:
190                 return result
191             code = 404
192             msg = "404 Not Found"
193 
194         error_content = None
195         try:
196             error_content = self.error_handler(code)
197         except Exception:
198             log.exception("Failed to render an error message!")
199         if error_content is None:
200             error_content = msg.encode()
201 
202         start_response(msg, [("Content-Type", "text/html")])
203         return [error_content]
204 
205     def _serve_request(self, environ, start_response):
206         path = environ["PATH_INFO"]
207 
208         m = re.fullmatch(r"/livereload/([0-9]+)/[0-9]+", path)
209         if m:
210             epoch = int(m[1])
211             start_response("200 OK", [("Content-Type", "text/plain")])
212 
213             def condition():
214                 return self._visible_epoch > epoch
215 
216             with self._epoch_cond:
217                 if not condition():
218                     # Stall the browser, respond as soon as there's something new.
219                     # If there's not, respond anyway after a minute.
220                     self._log_poll_request(environ.get("HTTP_REFERER"), request_id=path)
221                     self._epoch_cond.wait_for(condition, timeout=self.poll_response_timeout)
222                 return [b"%d" % self._visible_epoch]
223 
224         if path == "/js/livereload.js":
225             file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "livereload.js")
226         elif path.startswith(self.mount_path):
227             if path.endswith("/"):
228                 path += "index.html"
229             path = path[len(self.mount_path):]
230             file_path = os.path.join(self.root, path.lstrip("/"))
231         elif path == "/":
232             start_response("302 Found", [("Location", self.mount_path)])
233             return []
234         else:
235             return None  # Not found
236 
237         # Wait until the ongoing rebuild (if any) finishes, so we're not serving a half-built site.
238         with self._epoch_cond:
239             self._epoch_cond.wait_for(lambda: self._visible_epoch == self._wanted_epoch)
240             epoch = self._visible_epoch
241 
242         try:
243             file = open(file_path, "rb")
244         except OSError:
245             return None  # Not found
246 
247         if path.endswith(".html"):
248             with file:
249                 content = file.read()
250             content = self._inject_js_into_html(content, epoch)
251             file = io.BytesIO(content)
252             content_length = len(content)
253         else:
254             content_length = os.path.getsize(file_path)
255 
256         content_type = self._guess_type(file_path)
257         start_response(
258             "200 OK", [("Content-Type", content_type), ("Content-Length", str(content_length))]
259         )
260         return wsgiref.util.FileWrapper(file)
261 
262     @classmethod
263     def _inject_js_into_html(cls, content, epoch):
264         try:
265             body_end = content.rindex(b"</body>")
266         except ValueError:
267             body_end = len(content)
268         # The page will reload if the livereload poller returns a newer epoch than what it knows.
269         # The other timestamp becomes just a unique identifier for the initiating page.
270         return (
271             b'%b<script src="/js/livereload.js"></script><script>livereload(%d, %d);</script>%b'
272             % (content[:body_end], epoch, _timestamp(), content[body_end:])
273         )
274 
275     @classmethod
276     @functools.lru_cache()  # "Cache" to not repeat the same message for the same browser tab.
277     def _log_poll_request(cls, url, request_id):
278         log.info(f"Browser connected: {url}")
279 
280     def _guess_type(cls, path):
281         # MkDocs only ensures a few common types (as seen in livereload_tests.py::test_mime_types).
282         # Other uncommon types will not be accepted.
283         if path.endswith((".js", ".JS")):
284             return "application/javascript"
285         if path.endswith(".gz"):
286             return "application/gzip"
287 
288         guess, _ = mimetypes.guess_type(path)
289         if guess:
290             return guess
291         return "application/octet-stream"
292 
293 
294 class _Handler(wsgiref.simple_server.WSGIRequestHandler):
295     def log_request(self, code="-", size="-"):
296         level = logging.DEBUG if str(code) == "200" else logging.WARNING
297         log.log(level, f'"{self.requestline}" code {code}')
298 
299     def log_message(self, format, *args):
300         log.debug(format, *args)
301 
302 
303 def _timestamp():
304     return round(time.monotonic() * 1000)
305
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mkdocs/livereload/__init__.py b/mkdocs/livereload/__init__.py
--- a/mkdocs/livereload/__init__.py
+++ b/mkdocs/livereload/__init__.py
@@ -101,29 +101,26 @@
 
         def schedule(path):
             seen.add(path)
-            if os.path.isfile(path):
+            if path.is_file():
                 # Watchdog doesn't support watching files, so watch its directory and filter by path
                 handler = watchdog.events.FileSystemEventHandler()
-                handler.on_any_event = lambda event: callback(event, allowed_path=path)
+                handler.on_any_event = lambda event: callback(event, allowed_path=os.fspath(path))
 
-                parent = os.path.dirname(path)
+                parent = path.parent
                 log.debug(f"Watching file '{path}' through directory '{parent}'")
                 self.observer.schedule(handler, parent)
             else:
                 log.debug(f"Watching directory '{path}'")
                 self.observer.schedule(dir_handler, path, recursive=recursive)
 
-        schedule(os.path.realpath(path))
+        schedule(pathlib.Path(path).resolve())
 
         def watch_symlink_targets(path_obj):  # path is os.DirEntry or pathlib.Path
             if path_obj.is_symlink():
-                # The extra `readlink` is needed due to https://bugs.python.org/issue9949
-                target = os.path.realpath(os.readlink(os.fspath(path_obj)))
-                if target in seen or not os.path.exists(target):
+                path_obj = pathlib.Path(path_obj).resolve()
+                if path_obj in seen or not path_obj.exists():
                     return
-                schedule(target)
-
-                path_obj = pathlib.Path(target)
+                schedule(path_obj)
 
             if path_obj.is_dir() and recursive:
                 with os.scandir(os.fspath(path_obj)) as scan:

{"golden_diff": "diff --git a/mkdocs/livereload/__init__.py b/mkdocs/livereload/__init__.py\n--- a/mkdocs/livereload/__init__.py\n+++ b/mkdocs/livereload/__init__.py\n@@ -101,29 +101,26 @@\n \n         def schedule(path):\n             seen.add(path)\n-            if os.path.isfile(path):\n+            if path.is_file():\n                 # Watchdog doesn't support watching files, so watch its directory and filter by path\n                 handler = watchdog.events.FileSystemEventHandler()\n-                handler.on_any_event = lambda event: callback(event, allowed_path=path)\n+                handler.on_any_event = lambda event: callback(event, allowed_path=os.fspath(path))\n \n-                parent = os.path.dirname(path)\n+                parent = path.parent\n                 log.debug(f\"Watching file '{path}' through directory '{parent}'\")\n                 self.observer.schedule(handler, parent)\n             else:\n                 log.debug(f\"Watching directory '{path}'\")\n                 self.observer.schedule(dir_handler, path, recursive=recursive)\n \n-        schedule(os.path.realpath(path))\n+        schedule(pathlib.Path(path).resolve())\n \n         def watch_symlink_targets(path_obj):  # path is os.DirEntry or pathlib.Path\n             if path_obj.is_symlink():\n-                # The extra `readlink` is needed due to https://bugs.python.org/issue9949\n-                target = os.path.realpath(os.readlink(os.fspath(path_obj)))\n-                if target in seen or not os.path.exists(target):\n+                path_obj = pathlib.Path(path_obj).resolve()\n+                if path_obj in seen or not path_obj.exists():\n                     return\n-                schedule(target)\n-\n-                path_obj = pathlib.Path(target)\n+                schedule(path_obj)\n \n             if path_obj.is_dir() and recursive:\n                 with os.scandir(os.fspath(path_obj)) as scan:\n", "issue": "Re-building w/ symbolic links stopped working, regression after #2385\nSince a444c43 in master using the local development server via `mkdocs serve` updating files that are symbolically linked is not triggering to rebuild (and therefore not reloading browser tabs).\r\n\r\nOn first glance this is due to the switch to watchdog for detecting file-system changes which needs more guidance to handle this file-type.\r\n\r\nPreparing a PR with a patch.\r\n\r\nRef: a444c43474f91dea089922dd8fb188d1db3a4535\n", "before_files": [{"content": "import functools\nimport io\nimport logging\nimport mimetypes\nimport os\nimport os.path\nimport pathlib\nimport re\nimport socketserver\nimport threading\nimport time\nimport warnings\nimport wsgiref.simple_server\n\nimport watchdog.events\nimport watchdog.observers\n\n\nclass _LoggerAdapter(logging.LoggerAdapter):\n    def process(self, msg, kwargs):\n        return time.strftime(\"[%H:%M:%S] \") + msg, kwargs\n\n\nlog = _LoggerAdapter(logging.getLogger(__name__), {})\n\n\nclass LiveReloadServer(socketserver.ThreadingMixIn, wsgiref.simple_server.WSGIServer):\n    daemon_threads = True\n    poll_response_timeout = 60\n\n    def __init__(\n        self,\n        builder,\n        host,\n        port,\n        root,\n        mount_path=\"/\",\n        build_delay=0.25,\n        shutdown_delay=0.25,\n        **kwargs,\n    ):\n        self.builder = builder\n        self.server_name = host\n        self.server_port = port\n        self.root = os.path.abspath(root)\n        self.mount_path = (\"/\" + mount_path.lstrip(\"/\")).rstrip(\"/\") + \"/\"\n        self.url = f\"http://{self.server_name}:{self.server_port}{self.mount_path}\"\n        self.build_delay = build_delay\n        self.shutdown_delay = shutdown_delay\n        # To allow custom error pages.\n        self.error_handler = lambda code: None\n\n        super().__init__((host, port), _Handler, **kwargs)\n        self.set_app(self.serve_request)\n\n        self._wanted_epoch = _timestamp()  # The version of the site that started building.\n        self._visible_epoch = self._wanted_epoch  # Latest fully built version of the site.\n        self._epoch_cond = threading.Condition()  # Must be held when accessing _visible_epoch.\n\n        self._to_rebuild = {}  # Used as an ordered set of functions to call.\n        self._rebuild_cond = threading.Condition()  # Must be held when accessing _to_rebuild.\n\n        self._shutdown = False\n        self.serve_thread = threading.Thread(target=lambda: self.serve_forever(shutdown_delay))\n        self.observer = watchdog.observers.Observer(timeout=shutdown_delay)\n\n    def watch(self, path, func=None, recursive=True):\n        \"\"\"Add the 'path' to watched paths, call the function and reload when any file changes under it.\"\"\"\n        path = os.path.abspath(path)\n        if func in (None, self.builder):\n            func = self.builder\n        else:\n            warnings.warn(\n                \"Plugins should not pass the 'func' parameter of watch(). \"\n                \"The ability to execute custom callbacks will be removed soon.\",\n                DeprecationWarning,\n                stacklevel=2,\n            )\n\n        def callback(event, allowed_path=None):\n            if isinstance(event, watchdog.events.DirCreatedEvent):\n                return\n            if allowed_path is not None and event.src_path != allowed_path:\n                return\n            # Text editors always cause a \"file close\" event in addition to \"modified\" when saving\n            # a file. Some editors also have \"swap\" functionality that keeps writing into another\n            # file that's never closed. Prevent such write events from causing a rebuild.\n            if isinstance(event, watchdog.events.FileModifiedEvent):\n                # But FileClosedEvent is implemented only on Linux, otherwise we mustn't skip this:\n                if type(self.observer).__name__ == \"InotifyObserver\":\n                    return\n            log.debug(str(event))\n            with self._rebuild_cond:\n                self._to_rebuild[func] = True\n                self._rebuild_cond.notify_all()\n\n        dir_handler = watchdog.events.FileSystemEventHandler()\n        dir_handler.on_any_event = callback\n\n        seen = set()\n\n        def schedule(path):\n            seen.add(path)\n            if os.path.isfile(path):\n                # Watchdog doesn't support watching files, so watch its directory and filter by path\n                handler = watchdog.events.FileSystemEventHandler()\n                handler.on_any_event = lambda event: callback(event, allowed_path=path)\n\n                parent = os.path.dirname(path)\n                log.debug(f\"Watching file '{path}' through directory '{parent}'\")\n                self.observer.schedule(handler, parent)\n            else:\n                log.debug(f\"Watching directory '{path}'\")\n                self.observer.schedule(dir_handler, path, recursive=recursive)\n\n        schedule(os.path.realpath(path))\n\n        def watch_symlink_targets(path_obj):  # path is os.DirEntry or pathlib.Path\n            if path_obj.is_symlink():\n                # The extra `readlink` is needed due to https://bugs.python.org/issue9949\n                target = os.path.realpath(os.readlink(os.fspath(path_obj)))\n                if target in seen or not os.path.exists(target):\n                    return\n                schedule(target)\n\n                path_obj = pathlib.Path(target)\n\n            if path_obj.is_dir() and recursive:\n                with os.scandir(os.fspath(path_obj)) as scan:\n                    for entry in scan:\n                        watch_symlink_targets(entry)\n\n        watch_symlink_targets(pathlib.Path(path))\n\n    def serve(self):\n        self.observer.start()\n\n        log.info(f\"Serving on {self.url}\")\n        self.serve_thread.start()\n\n        self._build_loop()\n\n    def _build_loop(self):\n        while True:\n            with self._rebuild_cond:\n                while not self._rebuild_cond.wait_for(\n                    lambda: self._to_rebuild or self._shutdown, timeout=self.shutdown_delay\n                ):\n                    # We could have used just one wait instead of a loop + timeout, but we need\n                    # occasional breaks, otherwise on Windows we can't receive KeyboardInterrupt.\n                    pass\n                if self._shutdown:\n                    break\n                log.info(\"Detected file changes\")\n                while self._rebuild_cond.wait(timeout=self.build_delay):\n                    log.debug(\"Waiting for file changes to stop happening\")\n\n                self._wanted_epoch = _timestamp()\n                funcs = list(self._to_rebuild)\n                self._to_rebuild.clear()\n\n            for func in funcs:\n                func()\n\n            with self._epoch_cond:\n                log.info(\"Reloading browsers\")\n                self._visible_epoch = self._wanted_epoch\n                self._epoch_cond.notify_all()\n\n    def shutdown(self):\n        self.observer.stop()\n        with self._rebuild_cond:\n            self._shutdown = True\n            self._rebuild_cond.notify_all()\n\n        if self.serve_thread.is_alive():\n            super().shutdown()\n            self.serve_thread.join()\n            self.observer.join()\n\n    def serve_request(self, environ, start_response):\n        try:\n            result = self._serve_request(environ, start_response)\n        except Exception:\n            code = 500\n            msg = \"500 Internal Server Error\"\n            log.exception(msg)\n        else:\n            if result is not None:\n                return result\n            code = 404\n            msg = \"404 Not Found\"\n\n        error_content = None\n        try:\n            error_content = self.error_handler(code)\n        except Exception:\n            log.exception(\"Failed to render an error message!\")\n        if error_content is None:\n            error_content = msg.encode()\n\n        start_response(msg, [(\"Content-Type\", \"text/html\")])\n        return [error_content]\n\n    def _serve_request(self, environ, start_response):\n        path = environ[\"PATH_INFO\"]\n\n        m = re.fullmatch(r\"/livereload/([0-9]+)/[0-9]+\", path)\n        if m:\n            epoch = int(m[1])\n            start_response(\"200 OK\", [(\"Content-Type\", \"text/plain\")])\n\n            def condition():\n                return self._visible_epoch > epoch\n\n            with self._epoch_cond:\n                if not condition():\n                    # Stall the browser, respond as soon as there's something new.\n                    # If there's not, respond anyway after a minute.\n                    self._log_poll_request(environ.get(\"HTTP_REFERER\"), request_id=path)\n                    self._epoch_cond.wait_for(condition, timeout=self.poll_response_timeout)\n                return [b\"%d\" % self._visible_epoch]\n\n        if path == \"/js/livereload.js\":\n            file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), \"livereload.js\")\n        elif path.startswith(self.mount_path):\n            if path.endswith(\"/\"):\n                path += \"index.html\"\n            path = path[len(self.mount_path):]\n            file_path = os.path.join(self.root, path.lstrip(\"/\"))\n        elif path == \"/\":\n            start_response(\"302 Found\", [(\"Location\", self.mount_path)])\n            return []\n        else:\n            return None  # Not found\n\n        # Wait until the ongoing rebuild (if any) finishes, so we're not serving a half-built site.\n        with self._epoch_cond:\n            self._epoch_cond.wait_for(lambda: self._visible_epoch == self._wanted_epoch)\n            epoch = self._visible_epoch\n\n        try:\n            file = open(file_path, \"rb\")\n        except OSError:\n            return None  # Not found\n\n        if path.endswith(\".html\"):\n            with file:\n                content = file.read()\n            content = self._inject_js_into_html(content, epoch)\n            file = io.BytesIO(content)\n            content_length = len(content)\n        else:\n            content_length = os.path.getsize(file_path)\n\n        content_type = self._guess_type(file_path)\n        start_response(\n            \"200 OK\", [(\"Content-Type\", content_type), (\"Content-Length\", str(content_length))]\n        )\n        return wsgiref.util.FileWrapper(file)\n\n    @classmethod\n    def _inject_js_into_html(cls, content, epoch):\n        try:\n            body_end = content.rindex(b\"</body>\")\n        except ValueError:\n            body_end = len(content)\n        # The page will reload if the livereload poller returns a newer epoch than what it knows.\n        # The other timestamp becomes just a unique identifier for the initiating page.\n        return (\n            b'%b<script src=\"/js/livereload.js\"></script><script>livereload(%d, %d);</script>%b'\n            % (content[:body_end], epoch, _timestamp(), content[body_end:])\n        )\n\n    @classmethod\n    @functools.lru_cache()  # \"Cache\" to not repeat the same message for the same browser tab.\n    def _log_poll_request(cls, url, request_id):\n        log.info(f\"Browser connected: {url}\")\n\n    def _guess_type(cls, path):\n        # MkDocs only ensures a few common types (as seen in livereload_tests.py::test_mime_types).\n        # Other uncommon types will not be accepted.\n        if path.endswith((\".js\", \".JS\")):\n            return \"application/javascript\"\n        if path.endswith(\".gz\"):\n            return \"application/gzip\"\n\n        guess, _ = mimetypes.guess_type(path)\n        if guess:\n            return guess\n        return \"application/octet-stream\"\n\n\nclass _Handler(wsgiref.simple_server.WSGIRequestHandler):\n    def log_request(self, code=\"-\", size=\"-\"):\n        level = logging.DEBUG if str(code) == \"200\" else logging.WARNING\n        log.log(level, f'\"{self.requestline}\" code {code}')\n\n    def log_message(self, format, *args):\n        log.debug(format, *args)\n\n\ndef _timestamp():\n    return round(time.monotonic() * 1000)\n", "path": "mkdocs/livereload/__init__.py"}], "after_files": [{"content": "import functools\nimport io\nimport logging\nimport mimetypes\nimport os\nimport os.path\nimport pathlib\nimport re\nimport socketserver\nimport threading\nimport time\nimport warnings\nimport wsgiref.simple_server\n\nimport watchdog.events\nimport watchdog.observers\n\n\nclass _LoggerAdapter(logging.LoggerAdapter):\n    def process(self, msg, kwargs):\n        return time.strftime(\"[%H:%M:%S] \") + msg, kwargs\n\n\nlog = _LoggerAdapter(logging.getLogger(__name__), {})\n\n\nclass LiveReloadServer(socketserver.ThreadingMixIn, wsgiref.simple_server.WSGIServer):\n    daemon_threads = True\n    poll_response_timeout = 60\n\n    def __init__(\n        self,\n        builder,\n        host,\n        port,\n        root,\n        mount_path=\"/\",\n        build_delay=0.25,\n        shutdown_delay=0.25,\n        **kwargs,\n    ):\n        self.builder = builder\n        self.server_name = host\n        self.server_port = port\n        self.root = os.path.abspath(root)\n        self.mount_path = (\"/\" + mount_path.lstrip(\"/\")).rstrip(\"/\") + \"/\"\n        self.url = f\"http://{self.server_name}:{self.server_port}{self.mount_path}\"\n        self.build_delay = build_delay\n        self.shutdown_delay = shutdown_delay\n        # To allow custom error pages.\n        self.error_handler = lambda code: None\n\n        super().__init__((host, port), _Handler, **kwargs)\n        self.set_app(self.serve_request)\n\n        self._wanted_epoch = _timestamp()  # The version of the site that started building.\n        self._visible_epoch = self._wanted_epoch  # Latest fully built version of the site.\n        self._epoch_cond = threading.Condition()  # Must be held when accessing _visible_epoch.\n\n        self._to_rebuild = {}  # Used as an ordered set of functions to call.\n        self._rebuild_cond = threading.Condition()  # Must be held when accessing _to_rebuild.\n\n        self._shutdown = False\n        self.serve_thread = threading.Thread(target=lambda: self.serve_forever(shutdown_delay))\n        self.observer = watchdog.observers.Observer(timeout=shutdown_delay)\n\n    def watch(self, path, func=None, recursive=True):\n        \"\"\"Add the 'path' to watched paths, call the function and reload when any file changes under it.\"\"\"\n        path = os.path.abspath(path)\n        if func in (None, self.builder):\n            func = self.builder\n        else:\n            warnings.warn(\n                \"Plugins should not pass the 'func' parameter of watch(). \"\n                \"The ability to execute custom callbacks will be removed soon.\",\n                DeprecationWarning,\n                stacklevel=2,\n            )\n\n        def callback(event, allowed_path=None):\n            if isinstance(event, watchdog.events.DirCreatedEvent):\n                return\n            if allowed_path is not None and event.src_path != allowed_path:\n                return\n            # Text editors always cause a \"file close\" event in addition to \"modified\" when saving\n            # a file. Some editors also have \"swap\" functionality that keeps writing into another\n            # file that's never closed. Prevent such write events from causing a rebuild.\n            if isinstance(event, watchdog.events.FileModifiedEvent):\n                # But FileClosedEvent is implemented only on Linux, otherwise we mustn't skip this:\n                if type(self.observer).__name__ == \"InotifyObserver\":\n                    return\n            log.debug(str(event))\n            with self._rebuild_cond:\n                self._to_rebuild[func] = True\n                self._rebuild_cond.notify_all()\n\n        dir_handler = watchdog.events.FileSystemEventHandler()\n        dir_handler.on_any_event = callback\n\n        seen = set()\n\n        def schedule(path):\n            seen.add(path)\n            if path.is_file():\n                # Watchdog doesn't support watching files, so watch its directory and filter by path\n                handler = watchdog.events.FileSystemEventHandler()\n                handler.on_any_event = lambda event: callback(event, allowed_path=os.fspath(path))\n\n                parent = path.parent\n                log.debug(f\"Watching file '{path}' through directory '{parent}'\")\n                self.observer.schedule(handler, parent)\n            else:\n                log.debug(f\"Watching directory '{path}'\")\n                self.observer.schedule(dir_handler, path, recursive=recursive)\n\n        schedule(pathlib.Path(path).resolve())\n\n        def watch_symlink_targets(path_obj):  # path is os.DirEntry or pathlib.Path\n            if path_obj.is_symlink():\n                path_obj = pathlib.Path(path_obj).resolve()\n                if path_obj in seen or not path_obj.exists():\n                    return\n                schedule(path_obj)\n\n            if path_obj.is_dir() and recursive:\n                with os.scandir(os.fspath(path_obj)) as scan:\n                    for entry in scan:\n                        watch_symlink_targets(entry)\n\n        watch_symlink_targets(pathlib.Path(path))\n\n    def serve(self):\n        self.observer.start()\n\n        log.info(f\"Serving on {self.url}\")\n        self.serve_thread.start()\n\n        self._build_loop()\n\n    def _build_loop(self):\n        while True:\n            with self._rebuild_cond:\n                while not self._rebuild_cond.wait_for(\n                    lambda: self._to_rebuild or self._shutdown, timeout=self.shutdown_delay\n                ):\n                    # We could have used just one wait instead of a loop + timeout, but we need\n                    # occasional breaks, otherwise on Windows we can't receive KeyboardInterrupt.\n                    pass\n                if self._shutdown:\n                    break\n                log.info(\"Detected file changes\")\n                while self._rebuild_cond.wait(timeout=self.build_delay):\n                    log.debug(\"Waiting for file changes to stop happening\")\n\n                self._wanted_epoch = _timestamp()\n                funcs = list(self._to_rebuild)\n                self._to_rebuild.clear()\n\n            for func in funcs:\n                func()\n\n            with self._epoch_cond:\n                log.info(\"Reloading browsers\")\n                self._visible_epoch = self._wanted_epoch\n                self._epoch_cond.notify_all()\n\n    def shutdown(self):\n        self.observer.stop()\n        with self._rebuild_cond:\n            self._shutdown = True\n            self._rebuild_cond.notify_all()\n\n        if self.serve_thread.is_alive():\n            super().shutdown()\n            self.serve_thread.join()\n            self.observer.join()\n\n    def serve_request(self, environ, start_response):\n        try:\n            result = self._serve_request(environ, start_response)\n        except Exception:\n            code = 500\n            msg = \"500 Internal Server Error\"\n            log.exception(msg)\n        else:\n            if result is not None:\n                return result\n            code = 404\n            msg = \"404 Not Found\"\n\n        error_content = None\n        try:\n            error_content = self.error_handler(code)\n        except Exception:\n            log.exception(\"Failed to render an error message!\")\n        if error_content is None:\n            error_content = msg.encode()\n\n        start_response(msg, [(\"Content-Type\", \"text/html\")])\n        return [error_content]\n\n    def _serve_request(self, environ, start_response):\n        path = environ[\"PATH_INFO\"]\n\n        m = re.fullmatch(r\"/livereload/([0-9]+)/[0-9]+\", path)\n        if m:\n            epoch = int(m[1])\n            start_response(\"200 OK\", [(\"Content-Type\", \"text/plain\")])\n\n            def condition():\n                return self._visible_epoch > epoch\n\n            with self._epoch_cond:\n                if not condition():\n                    # Stall the browser, respond as soon as there's something new.\n                    # If there's not, respond anyway after a minute.\n                    self._log_poll_request(environ.get(\"HTTP_REFERER\"), request_id=path)\n                    self._epoch_cond.wait_for(condition, timeout=self.poll_response_timeout)\n                return [b\"%d\" % self._visible_epoch]\n\n        if path == \"/js/livereload.js\":\n            file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), \"livereload.js\")\n        elif path.startswith(self.mount_path):\n            if path.endswith(\"/\"):\n                path += \"index.html\"\n            path = path[len(self.mount_path):]\n            file_path = os.path.join(self.root, path.lstrip(\"/\"))\n        elif path == \"/\":\n            start_response(\"302 Found\", [(\"Location\", self.mount_path)])\n            return []\n        else:\n            return None  # Not found\n\n        # Wait until the ongoing rebuild (if any) finishes, so we're not serving a half-built site.\n        with self._epoch_cond:\n            self._epoch_cond.wait_for(lambda: self._visible_epoch == self._wanted_epoch)\n            epoch = self._visible_epoch\n\n        try:\n            file = open(file_path, \"rb\")\n        except OSError:\n            return None  # Not found\n\n        if path.endswith(\".html\"):\n            with file:\n                content = file.read()\n            content = self._inject_js_into_html(content, epoch)\n            file = io.BytesIO(content)\n            content_length = len(content)\n        else:\n            content_length = os.path.getsize(file_path)\n\n        content_type = self._guess_type(file_path)\n        start_response(\n            \"200 OK\", [(\"Content-Type\", content_type), (\"Content-Length\", str(content_length))]\n        )\n        return wsgiref.util.FileWrapper(file)\n\n    @classmethod\n    def _inject_js_into_html(cls, content, epoch):\n        try:\n            body_end = content.rindex(b\"</body>\")\n        except ValueError:\n            body_end = len(content)\n        # The page will reload if the livereload poller returns a newer epoch than what it knows.\n        # The other timestamp becomes just a unique identifier for the initiating page.\n        return (\n            b'%b<script src=\"/js/livereload.js\"></script><script>livereload(%d, %d);</script>%b'\n            % (content[:body_end], epoch, _timestamp(), content[body_end:])\n        )\n\n    @classmethod\n    @functools.lru_cache()  # \"Cache\" to not repeat the same message for the same browser tab.\n    def _log_poll_request(cls, url, request_id):\n        log.info(f\"Browser connected: {url}\")\n\n    def _guess_type(cls, path):\n        # MkDocs only ensures a few common types (as seen in livereload_tests.py::test_mime_types).\n        # Other uncommon types will not be accepted.\n        if path.endswith((\".js\", \".JS\")):\n            return \"application/javascript\"\n        if path.endswith(\".gz\"):\n            return \"application/gzip\"\n\n        guess, _ = mimetypes.guess_type(path)\n        if guess:\n            return guess\n        return \"application/octet-stream\"\n\n\nclass _Handler(wsgiref.simple_server.WSGIRequestHandler):\n    def log_request(self, code=\"-\", size=\"-\"):\n        level = logging.DEBUG if str(code) == \"200\" else logging.WARNING\n        log.log(level, f'\"{self.requestline}\" code {code}')\n\n    def log_message(self, format, *args):\n        log.debug(format, *args)\n\n\ndef _timestamp():\n    return round(time.monotonic() * 1000)\n", "path": "mkdocs/livereload/__init__.py"}]}