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
restore re-building with symbolic-links, closes #2425
previously (1.1.2 + master at 23e2051) building was triggered by changes
of file-content that was symbolically linked within docs_dir while
`mkdocs serve` was running.

since migrating from livereload>=2.6.1 to watchdog>=2.0.0 to detect
file-system changes (triggering the re-build) it stopped working.

this is because watchdog does not support symbolic links out of the box,
e.g. see [1].

change is to provide additional observe instructions on the realpath [2]
for the following cases:

1. docs_dir & config_file_path path deviation:

   when the absolute path to either the `docs_dir` or the `config_file` is
   different from its realpath, the realpath is added for observing (as
   well).

2. symbolic links within docs_dir:

   if a file within docs_dir is a symbolic link, the files real path
   is added for observing. sub-directories (that are not symbolically
   linked) are traversed up to a depth of nine levels (only if the
   recursive flag is enabled, otherwise no traversal into sub-directories).

Ref: 23e205153f01d24d50fe9ba18e5186cdbc2c2dbe
[1]: https://github.com/gorakhargosh/watchdog/issues/365
[2]: <https://docs.python.org/3.8/library/os.path.html#os.path.realpath>

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

{"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@@ -4,6 +4,7 @@\n import mimetypes\n import os\n import os.path\n+import pathlib\n import re\n import socketserver\n import threading\n@@ -76,8 +77,10 @@\n                 stacklevel=2,\n             )\n \n-        def callback(event):\n-            if event.is_directory:\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@@ -91,9 +94,43 @@\n                 self._to_rebuild[func] = True\n                 self._rebuild_cond.notify_all()\n \n-        handler = watchdog.events.FileSystemEventHandler()\n-        handler.on_any_event = callback\n-        self.observer.schedule(handler, path, recursive=recursive)\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", "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\nrestore re-building with symbolic-links, closes #2425\npreviously (1.1.2 + master at 23e2051) building was triggered by changes\r\nof file-content that was symbolically linked within docs_dir while\r\n`mkdocs serve` was running.\r\n\r\nsince migrating from livereload>=2.6.1 to watchdog>=2.0.0 to detect\r\nfile-system changes (triggering the re-build) it stopped working.\r\n\r\nthis is because watchdog does not support symbolic links out of the box,\r\ne.g. see [1].\r\n\r\nchange is to provide additional observe instructions on the realpath [2]\r\nfor the following cases:\r\n\r\n1. docs_dir & config_file_path path deviation:\r\n\r\n   when the absolute path to either the `docs_dir` or the `config_file` is\r\n   different from its realpath, the realpath is added for observing (as\r\n   well).\r\n\r\n2. symbolic links within docs_dir:\r\n\r\n   if a file within docs_dir is a symbolic link, the files real path\r\n   is added for observing. sub-directories (that are not symbolically\r\n   linked) are traversed up to a depth of nine levels (only if the\r\n   recursive flag is enabled, otherwise no traversal into sub-directories).\r\n\r\nRef: 23e205153f01d24d50fe9ba18e5186cdbc2c2dbe\r\n[1]: https://github.com/gorakhargosh/watchdog/issues/365\r\n[2]: <https://docs.python.org/3.8/library/os.path.html#os.path.realpath>\n", "before_files": [{"content": "import functools\nimport io\nimport logging\nimport mimetypes\nimport os\nimport os.path\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):\n            if event.is_directory:\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        handler = watchdog.events.FileSystemEventHandler()\n        handler.on_any_event = callback\n        self.observer.schedule(handler, path, recursive=recursive)\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 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"}]}