We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
session could auto-no-op any callbacks invoked after the session is destroyed
Right now, if you do something like this:

```
def update_all_sessions(server_context):
   for session_context in server_context.sessions:
      yield session_context.with_locked_document(update_document)
```

One of the sessions could expire and be destroyed before your code gets to it.

So you need to write a check for that:

```
def update_all_sessions(server_context):
   for session_context in server_context.sessions:
      if not session_context.destroyed:
        yield session_context.with_locked_document(update_document)
```

I think it would be better if `with_locked_document` did this automatically (just became a no-op on destroyed sessions). This could be done in session.py, `_needs_document_lock_wrapper` perhaps.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `bokeh/server/session.py`
Content:
```
1 ''' Provides the ``ServerSession`` class.
2 
3 '''
4 from __future__ import absolute_import
5 
6 import logging
7 log = logging.getLogger(__name__)
8 
9 import time
10 
11 from tornado import gen, locks
12 
13 from ..util.tornado import yield_for_all_futures
14 
15 from .callbacks import _DocumentCallbackGroup
16 
17 def current_time():
18     '''Return the time in milliseconds since the epoch as a floating
19        point number.
20     '''
21     try:
22         # python >=3.3 only
23         return time.monotonic() * 1000
24     except:
25         # if your python is old, don't set your clock backward!
26         return time.time() * 1000
27 
28 def _needs_document_lock(func):
29     '''Decorator that adds the necessary locking and post-processing
30        to manipulate the session's document. Expects to decorate a
31        method on ServerSession and transforms it into a coroutine
32        if it wasn't already.
33     '''
34     @gen.coroutine
35     def _needs_document_lock_wrapper(self, *args, **kwargs):
36         # while we wait for and hold the lock, prevent the session
37         # from being discarded. This avoids potential weirdness
38         # with the session vanishing in the middle of some async
39         # task.
40         self.block_expiration()
41         try:
42             with (yield self._lock.acquire()):
43                 if self._pending_writes is not None:
44                     raise RuntimeError("internal class invariant violated: _pending_writes " + \
45                                        "should be None if lock is not held")
46                 self._pending_writes = []
47                 try:
48                     result = yield yield_for_all_futures(func(self, *args, **kwargs))
49                 finally:
50                     # we want to be very sure we reset this or we'll
51                     # keep hitting the RuntimeError above as soon as
52                     # any callback goes wrong
53                     pending_writes = self._pending_writes
54                     self._pending_writes = None
55                 for p in pending_writes:
56                     yield p
57             raise gen.Return(result)
58         finally:
59             self.unblock_expiration()
60     return _needs_document_lock_wrapper
61 
62 class ServerSession(object):
63     ''' Hosts an application "instance" (an instantiated Document) for one or more connections.
64 
65     '''
66 
67     def __init__(self, session_id, document, io_loop=None):
68         if session_id is None:
69             raise ValueError("Sessions must have an id")
70         if document is None:
71             raise ValueError("Sessions must have a document")
72         self._id = session_id
73         self._document = document
74         self._loop = io_loop
75         self._subscribed_connections = set()
76         self._last_unsubscribe_time = current_time()
77         self._lock = locks.Lock()
78         self._current_patch_connection = None
79         self._document.on_change_dispatch_to(self)
80         self._callbacks = _DocumentCallbackGroup(io_loop)
81         self._pending_writes = None
82         self._destroyed = False
83         self._expiration_requested = False
84         self._expiration_blocked_count = 0
85 
86         wrapped_callbacks = self._wrap_session_callbacks(self._document.session_callbacks)
87         self._callbacks.add_session_callbacks(wrapped_callbacks)
88 
89     @property
90     def document(self):
91         return self._document
92 
93     @property
94     def id(self):
95         return self._id
96 
97     @property
98     def destroyed(self):
99         return self._destroyed
100 
101     @property
102     def expiration_requested(self):
103         return self._expiration_requested
104 
105     @property
106     def expiration_blocked(self):
107         return self._expiration_blocked_count > 0
108 
109     @property
110     def expiration_blocked_count(self):
111         return self._expiration_blocked_count
112 
113     def destroy(self):
114         self._destroyed = True
115         self._document.delete_modules()
116         self._document.remove_on_change(self)
117         self._callbacks.remove_all_callbacks()
118 
119     def request_expiration(self):
120         """ Used in test suite for now. Forces immediate expiration if no connections."""
121         self._expiration_requested = True
122 
123     def block_expiration(self):
124         self._expiration_blocked_count += 1
125 
126     def unblock_expiration(self):
127         if self._expiration_blocked_count <= 0:
128             raise RuntimeError("mismatched block_expiration / unblock_expiration")
129         self._expiration_blocked_count -= 1
130 
131     def subscribe(self, connection):
132         """This should only be called by ServerConnection.subscribe_session or our book-keeping will be broken"""
133         self._subscribed_connections.add(connection)
134 
135     def unsubscribe(self, connection):
136         """This should only be called by ServerConnection.unsubscribe_session or our book-keeping will be broken"""
137         self._subscribed_connections.discard(connection)
138         self._last_unsubscribe_time = current_time()
139 
140     @property
141     def connection_count(self):
142         return len(self._subscribed_connections)
143 
144     @property
145     def milliseconds_since_last_unsubscribe(self):
146         return current_time() - self._last_unsubscribe_time
147 
148     @_needs_document_lock
149     def with_document_locked(self, func, *args, **kwargs):
150         ''' Asynchronously locks the document and runs the function with it locked.'''
151         return func(*args, **kwargs)
152 
153     def _wrap_document_callback(self, callback):
154         if getattr(callback, "nolock", False):
155             return callback
156         def wrapped_callback(*args, **kwargs):
157             return self.with_document_locked(callback, *args, **kwargs)
158         return wrapped_callback
159 
160     def _wrap_session_callback(self, callback):
161         wrapped = self._wrap_document_callback(callback.callback)
162         return callback._copy_with_changed_callback(wrapped)
163 
164     def _wrap_session_callbacks(self, callbacks):
165         wrapped = []
166         for cb in callbacks:
167             wrapped.append(self._wrap_session_callback(cb))
168         return wrapped
169 
170     def _document_patched(self, event):
171         may_suppress = event.setter is self
172 
173         if self._pending_writes is None:
174             raise RuntimeError("_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes")
175 
176         # TODO (havocp): our "change sync" protocol is flawed because if both
177         # sides change the same attribute at the same time, they will each end
178         # up with the state of the other and their final states will differ.
179         for connection in self._subscribed_connections:
180             if may_suppress and connection is self._current_patch_connection:
181                 log.trace("Not sending notification back to client %r for a change it requested", connection)
182             else:
183                 self._pending_writes.append(connection.send_patch_document(event))
184 
185     @_needs_document_lock
186     def _handle_pull(self, message, connection):
187         log.debug("Sending pull-doc-reply from session %r", self.id)
188         return connection.protocol.create('PULL-DOC-REPLY', message.header['msgid'], self.document)
189 
190     def _session_callback_added(self, event):
191         wrapped = self._wrap_session_callback(event.callback)
192         self._callbacks.add_session_callback(wrapped)
193 
194     def _session_callback_removed(self, event):
195         self._callbacks.remove_session_callback(event.callback)
196 
197     @classmethod
198     def pull(cls, message, connection):
199         ''' Handle a PULL-DOC, return a Future with work to be scheduled. '''
200         return connection.session._handle_pull(message, connection)
201 
202     @_needs_document_lock
203     def _handle_push(self, message, connection):
204         log.debug("pushing doc to session %r", self.id)
205         message.push_to_document(self.document)
206         return connection.ok(message)
207 
208     @classmethod
209     def push(cls, message, connection):
210         ''' Handle a PUSH-DOC, return a Future with work to be scheduled. '''
211         return connection.session._handle_push(message, connection)
212 
213     @_needs_document_lock
214     def _handle_patch(self, message, connection):
215         self._current_patch_connection = connection
216         try:
217             message.apply_to_document(self.document, self)
218         finally:
219             self._current_patch_connection = None
220 
221         return connection.ok(message)
222 
223     @_needs_document_lock
224     def _handle_event(self, message, connection):
225         message.notify_event(self.document)
226         return connection.ok(message)
227 
228     @classmethod
229     def event(cls, message, connection):
230         return connection.session._handle_event(message, connection)
231 
232 
233     @classmethod
234     def patch(cls, message, connection):
235         ''' Handle a PATCH-DOC, return a Future with work to be scheduled. '''
236         return connection.session._handle_patch(message, connection)
237
```
--- 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/bokeh/server/session.py b/bokeh/server/session.py
--- a/bokeh/server/session.py
+++ b/bokeh/server/session.py
@@ -37,6 +37,9 @@
         # from being discarded. This avoids potential weirdness
         # with the session vanishing in the middle of some async
         # task.
+        if self.destroyed:
+            log.debug("Ignoring locked callback on already-destroyed session.")
+            raise gen.Return(None)
         self.block_expiration()
         try:
             with (yield self._lock.acquire()):

{"golden_diff": "diff --git a/bokeh/server/session.py b/bokeh/server/session.py\n--- a/bokeh/server/session.py\n+++ b/bokeh/server/session.py\n@@ -37,6 +37,9 @@\n         # from being discarded. This avoids potential weirdness\n         # with the session vanishing in the middle of some async\n         # task.\n+        if self.destroyed:\n+            log.debug(\"Ignoring locked callback on already-destroyed session.\")\n+            raise gen.Return(None)\n         self.block_expiration()\n         try:\n             with (yield self._lock.acquire()):\n", "issue": "session could auto-no-op any callbacks invoked after the session is destroyed\nRight now, if you do something like this:\n\n```\ndef update_all_sessions(server_context):\n   for session_context in server_context.sessions:\n      yield session_context.with_locked_document(update_document)\n```\n\nOne of the sessions could expire and be destroyed before your code gets to it.\n\nSo you need to write a check for that:\n\n```\ndef update_all_sessions(server_context):\n   for session_context in server_context.sessions:\n      if not session_context.destroyed:\n        yield session_context.with_locked_document(update_document)\n```\n\nI think it would be better if `with_locked_document` did this automatically (just became a no-op on destroyed sessions). This could be done in session.py, `_needs_document_lock_wrapper` perhaps.\n\n", "before_files": [{"content": "''' Provides the ``ServerSession`` class.\n\n'''\nfrom __future__ import absolute_import\n\nimport logging\nlog = logging.getLogger(__name__)\n\nimport time\n\nfrom tornado import gen, locks\n\nfrom ..util.tornado import yield_for_all_futures\n\nfrom .callbacks import _DocumentCallbackGroup\n\ndef current_time():\n    '''Return the time in milliseconds since the epoch as a floating\n       point number.\n    '''\n    try:\n        # python >=3.3 only\n        return time.monotonic() * 1000\n    except:\n        # if your python is old, don't set your clock backward!\n        return time.time() * 1000\n\ndef _needs_document_lock(func):\n    '''Decorator that adds the necessary locking and post-processing\n       to manipulate the session's document. Expects to decorate a\n       method on ServerSession and transforms it into a coroutine\n       if it wasn't already.\n    '''\n    @gen.coroutine\n    def _needs_document_lock_wrapper(self, *args, **kwargs):\n        # while we wait for and hold the lock, prevent the session\n        # from being discarded. This avoids potential weirdness\n        # with the session vanishing in the middle of some async\n        # task.\n        self.block_expiration()\n        try:\n            with (yield self._lock.acquire()):\n                if self._pending_writes is not None:\n                    raise RuntimeError(\"internal class invariant violated: _pending_writes \" + \\\n                                       \"should be None if lock is not held\")\n                self._pending_writes = []\n                try:\n                    result = yield yield_for_all_futures(func(self, *args, **kwargs))\n                finally:\n                    # we want to be very sure we reset this or we'll\n                    # keep hitting the RuntimeError above as soon as\n                    # any callback goes wrong\n                    pending_writes = self._pending_writes\n                    self._pending_writes = None\n                for p in pending_writes:\n                    yield p\n            raise gen.Return(result)\n        finally:\n            self.unblock_expiration()\n    return _needs_document_lock_wrapper\n\nclass ServerSession(object):\n    ''' Hosts an application \"instance\" (an instantiated Document) for one or more connections.\n\n    '''\n\n    def __init__(self, session_id, document, io_loop=None):\n        if session_id is None:\n            raise ValueError(\"Sessions must have an id\")\n        if document is None:\n            raise ValueError(\"Sessions must have a document\")\n        self._id = session_id\n        self._document = document\n        self._loop = io_loop\n        self._subscribed_connections = set()\n        self._last_unsubscribe_time = current_time()\n        self._lock = locks.Lock()\n        self._current_patch_connection = None\n        self._document.on_change_dispatch_to(self)\n        self._callbacks = _DocumentCallbackGroup(io_loop)\n        self._pending_writes = None\n        self._destroyed = False\n        self._expiration_requested = False\n        self._expiration_blocked_count = 0\n\n        wrapped_callbacks = self._wrap_session_callbacks(self._document.session_callbacks)\n        self._callbacks.add_session_callbacks(wrapped_callbacks)\n\n    @property\n    def document(self):\n        return self._document\n\n    @property\n    def id(self):\n        return self._id\n\n    @property\n    def destroyed(self):\n        return self._destroyed\n\n    @property\n    def expiration_requested(self):\n        return self._expiration_requested\n\n    @property\n    def expiration_blocked(self):\n        return self._expiration_blocked_count > 0\n\n    @property\n    def expiration_blocked_count(self):\n        return self._expiration_blocked_count\n\n    def destroy(self):\n        self._destroyed = True\n        self._document.delete_modules()\n        self._document.remove_on_change(self)\n        self._callbacks.remove_all_callbacks()\n\n    def request_expiration(self):\n        \"\"\" Used in test suite for now. Forces immediate expiration if no connections.\"\"\"\n        self._expiration_requested = True\n\n    def block_expiration(self):\n        self._expiration_blocked_count += 1\n\n    def unblock_expiration(self):\n        if self._expiration_blocked_count <= 0:\n            raise RuntimeError(\"mismatched block_expiration / unblock_expiration\")\n        self._expiration_blocked_count -= 1\n\n    def subscribe(self, connection):\n        \"\"\"This should only be called by ServerConnection.subscribe_session or our book-keeping will be broken\"\"\"\n        self._subscribed_connections.add(connection)\n\n    def unsubscribe(self, connection):\n        \"\"\"This should only be called by ServerConnection.unsubscribe_session or our book-keeping will be broken\"\"\"\n        self._subscribed_connections.discard(connection)\n        self._last_unsubscribe_time = current_time()\n\n    @property\n    def connection_count(self):\n        return len(self._subscribed_connections)\n\n    @property\n    def milliseconds_since_last_unsubscribe(self):\n        return current_time() - self._last_unsubscribe_time\n\n    @_needs_document_lock\n    def with_document_locked(self, func, *args, **kwargs):\n        ''' Asynchronously locks the document and runs the function with it locked.'''\n        return func(*args, **kwargs)\n\n    def _wrap_document_callback(self, callback):\n        if getattr(callback, \"nolock\", False):\n            return callback\n        def wrapped_callback(*args, **kwargs):\n            return self.with_document_locked(callback, *args, **kwargs)\n        return wrapped_callback\n\n    def _wrap_session_callback(self, callback):\n        wrapped = self._wrap_document_callback(callback.callback)\n        return callback._copy_with_changed_callback(wrapped)\n\n    def _wrap_session_callbacks(self, callbacks):\n        wrapped = []\n        for cb in callbacks:\n            wrapped.append(self._wrap_session_callback(cb))\n        return wrapped\n\n    def _document_patched(self, event):\n        may_suppress = event.setter is self\n\n        if self._pending_writes is None:\n            raise RuntimeError(\"_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes\")\n\n        # TODO (havocp): our \"change sync\" protocol is flawed because if both\n        # sides change the same attribute at the same time, they will each end\n        # up with the state of the other and their final states will differ.\n        for connection in self._subscribed_connections:\n            if may_suppress and connection is self._current_patch_connection:\n                log.trace(\"Not sending notification back to client %r for a change it requested\", connection)\n            else:\n                self._pending_writes.append(connection.send_patch_document(event))\n\n    @_needs_document_lock\n    def _handle_pull(self, message, connection):\n        log.debug(\"Sending pull-doc-reply from session %r\", self.id)\n        return connection.protocol.create('PULL-DOC-REPLY', message.header['msgid'], self.document)\n\n    def _session_callback_added(self, event):\n        wrapped = self._wrap_session_callback(event.callback)\n        self._callbacks.add_session_callback(wrapped)\n\n    def _session_callback_removed(self, event):\n        self._callbacks.remove_session_callback(event.callback)\n\n    @classmethod\n    def pull(cls, message, connection):\n        ''' Handle a PULL-DOC, return a Future with work to be scheduled. '''\n        return connection.session._handle_pull(message, connection)\n\n    @_needs_document_lock\n    def _handle_push(self, message, connection):\n        log.debug(\"pushing doc to session %r\", self.id)\n        message.push_to_document(self.document)\n        return connection.ok(message)\n\n    @classmethod\n    def push(cls, message, connection):\n        ''' Handle a PUSH-DOC, return a Future with work to be scheduled. '''\n        return connection.session._handle_push(message, connection)\n\n    @_needs_document_lock\n    def _handle_patch(self, message, connection):\n        self._current_patch_connection = connection\n        try:\n            message.apply_to_document(self.document, self)\n        finally:\n            self._current_patch_connection = None\n\n        return connection.ok(message)\n\n    @_needs_document_lock\n    def _handle_event(self, message, connection):\n        message.notify_event(self.document)\n        return connection.ok(message)\n\n    @classmethod\n    def event(cls, message, connection):\n        return connection.session._handle_event(message, connection)\n\n\n    @classmethod\n    def patch(cls, message, connection):\n        ''' Handle a PATCH-DOC, return a Future with work to be scheduled. '''\n        return connection.session._handle_patch(message, connection)\n", "path": "bokeh/server/session.py"}], "after_files": [{"content": "''' Provides the ``ServerSession`` class.\n\n'''\nfrom __future__ import absolute_import\n\nimport logging\nlog = logging.getLogger(__name__)\n\nimport time\n\nfrom tornado import gen, locks\n\nfrom ..util.tornado import yield_for_all_futures\n\nfrom .callbacks import _DocumentCallbackGroup\n\ndef current_time():\n    '''Return the time in milliseconds since the epoch as a floating\n       point number.\n    '''\n    try:\n        # python >=3.3 only\n        return time.monotonic() * 1000\n    except:\n        # if your python is old, don't set your clock backward!\n        return time.time() * 1000\n\ndef _needs_document_lock(func):\n    '''Decorator that adds the necessary locking and post-processing\n       to manipulate the session's document. Expects to decorate a\n       method on ServerSession and transforms it into a coroutine\n       if it wasn't already.\n    '''\n    @gen.coroutine\n    def _needs_document_lock_wrapper(self, *args, **kwargs):\n        # while we wait for and hold the lock, prevent the session\n        # from being discarded. This avoids potential weirdness\n        # with the session vanishing in the middle of some async\n        # task.\n        if self.destroyed:\n            log.debug(\"Ignoring locked callback on already-destroyed session.\")\n            raise gen.Return(None)\n        self.block_expiration()\n        try:\n            with (yield self._lock.acquire()):\n                if self._pending_writes is not None:\n                    raise RuntimeError(\"internal class invariant violated: _pending_writes \" + \\\n                                       \"should be None if lock is not held\")\n                self._pending_writes = []\n                try:\n                    result = yield yield_for_all_futures(func(self, *args, **kwargs))\n                finally:\n                    # we want to be very sure we reset this or we'll\n                    # keep hitting the RuntimeError above as soon as\n                    # any callback goes wrong\n                    pending_writes = self._pending_writes\n                    self._pending_writes = None\n                for p in pending_writes:\n                    yield p\n            raise gen.Return(result)\n        finally:\n            self.unblock_expiration()\n    return _needs_document_lock_wrapper\n\nclass ServerSession(object):\n    ''' Hosts an application \"instance\" (an instantiated Document) for one or more connections.\n\n    '''\n\n    def __init__(self, session_id, document, io_loop=None):\n        if session_id is None:\n            raise ValueError(\"Sessions must have an id\")\n        if document is None:\n            raise ValueError(\"Sessions must have a document\")\n        self._id = session_id\n        self._document = document\n        self._loop = io_loop\n        self._subscribed_connections = set()\n        self._last_unsubscribe_time = current_time()\n        self._lock = locks.Lock()\n        self._current_patch_connection = None\n        self._document.on_change_dispatch_to(self)\n        self._callbacks = _DocumentCallbackGroup(io_loop)\n        self._pending_writes = None\n        self._destroyed = False\n        self._expiration_requested = False\n        self._expiration_blocked_count = 0\n\n        wrapped_callbacks = self._wrap_session_callbacks(self._document.session_callbacks)\n        self._callbacks.add_session_callbacks(wrapped_callbacks)\n\n    @property\n    def document(self):\n        return self._document\n\n    @property\n    def id(self):\n        return self._id\n\n    @property\n    def destroyed(self):\n        return self._destroyed\n\n    @property\n    def expiration_requested(self):\n        return self._expiration_requested\n\n    @property\n    def expiration_blocked(self):\n        return self._expiration_blocked_count > 0\n\n    @property\n    def expiration_blocked_count(self):\n        return self._expiration_blocked_count\n\n    def destroy(self):\n        self._destroyed = True\n        self._document.delete_modules()\n        self._document.remove_on_change(self)\n        self._callbacks.remove_all_callbacks()\n\n    def request_expiration(self):\n        \"\"\" Used in test suite for now. Forces immediate expiration if no connections.\"\"\"\n        self._expiration_requested = True\n\n    def block_expiration(self):\n        self._expiration_blocked_count += 1\n\n    def unblock_expiration(self):\n        if self._expiration_blocked_count <= 0:\n            raise RuntimeError(\"mismatched block_expiration / unblock_expiration\")\n        self._expiration_blocked_count -= 1\n\n    def subscribe(self, connection):\n        \"\"\"This should only be called by ServerConnection.subscribe_session or our book-keeping will be broken\"\"\"\n        self._subscribed_connections.add(connection)\n\n    def unsubscribe(self, connection):\n        \"\"\"This should only be called by ServerConnection.unsubscribe_session or our book-keeping will be broken\"\"\"\n        self._subscribed_connections.discard(connection)\n        self._last_unsubscribe_time = current_time()\n\n    @property\n    def connection_count(self):\n        return len(self._subscribed_connections)\n\n    @property\n    def milliseconds_since_last_unsubscribe(self):\n        return current_time() - self._last_unsubscribe_time\n\n    @_needs_document_lock\n    def with_document_locked(self, func, *args, **kwargs):\n        ''' Asynchronously locks the document and runs the function with it locked.'''\n        return func(*args, **kwargs)\n\n    def _wrap_document_callback(self, callback):\n        if getattr(callback, \"nolock\", False):\n            return callback\n        def wrapped_callback(*args, **kwargs):\n            return self.with_document_locked(callback, *args, **kwargs)\n        return wrapped_callback\n\n    def _wrap_session_callback(self, callback):\n        wrapped = self._wrap_document_callback(callback.callback)\n        return callback._copy_with_changed_callback(wrapped)\n\n    def _wrap_session_callbacks(self, callbacks):\n        wrapped = []\n        for cb in callbacks:\n            wrapped.append(self._wrap_session_callback(cb))\n        return wrapped\n\n    def _document_patched(self, event):\n        may_suppress = event.setter is self\n\n        if self._pending_writes is None:\n            raise RuntimeError(\"_pending_writes should be non-None when we have a document lock, and we should have the lock when the document changes\")\n\n        # TODO (havocp): our \"change sync\" protocol is flawed because if both\n        # sides change the same attribute at the same time, they will each end\n        # up with the state of the other and their final states will differ.\n        for connection in self._subscribed_connections:\n            if may_suppress and connection is self._current_patch_connection:\n                log.trace(\"Not sending notification back to client %r for a change it requested\", connection)\n            else:\n                self._pending_writes.append(connection.send_patch_document(event))\n\n    @_needs_document_lock\n    def _handle_pull(self, message, connection):\n        log.debug(\"Sending pull-doc-reply from session %r\", self.id)\n        return connection.protocol.create('PULL-DOC-REPLY', message.header['msgid'], self.document)\n\n    def _session_callback_added(self, event):\n        wrapped = self._wrap_session_callback(event.callback)\n        self._callbacks.add_session_callback(wrapped)\n\n    def _session_callback_removed(self, event):\n        self._callbacks.remove_session_callback(event.callback)\n\n    @classmethod\n    def pull(cls, message, connection):\n        ''' Handle a PULL-DOC, return a Future with work to be scheduled. '''\n        return connection.session._handle_pull(message, connection)\n\n    @_needs_document_lock\n    def _handle_push(self, message, connection):\n        log.debug(\"pushing doc to session %r\", self.id)\n        message.push_to_document(self.document)\n        return connection.ok(message)\n\n    @classmethod\n    def push(cls, message, connection):\n        ''' Handle a PUSH-DOC, return a Future with work to be scheduled. '''\n        return connection.session._handle_push(message, connection)\n\n    @_needs_document_lock\n    def _handle_patch(self, message, connection):\n        self._current_patch_connection = connection\n        try:\n            message.apply_to_document(self.document, self)\n        finally:\n            self._current_patch_connection = None\n\n        return connection.ok(message)\n\n    @_needs_document_lock\n    def _handle_event(self, message, connection):\n        message.notify_event(self.document)\n        return connection.ok(message)\n\n    @classmethod\n    def event(cls, message, connection):\n        return connection.session._handle_event(message, connection)\n\n\n    @classmethod\n    def patch(cls, message, connection):\n        ''' Handle a PATCH-DOC, return a Future with work to be scheduled. '''\n        return connection.session._handle_patch(message, connection)\n", "path": "bokeh/server/session.py"}]}