#-----------------------------------------------------------------------------# Copyright (c) 2012 - 2022, Anaconda, Inc., and Bokeh Contributors.# All rights reserved.## The full license is in the file LICENSE.txt, distributed with this software.#-----------------------------------------------------------------------------''' Provides the ``ServerSession`` class.'''#-----------------------------------------------------------------------------# Boilerplate#-----------------------------------------------------------------------------from__future__importannotationsimportlogging# isort:skiplog=logging.getLogger(__name__)#-----------------------------------------------------------------------------# Imports#-----------------------------------------------------------------------------# Standard library importsfromtypingimportTYPE_CHECKING,Any,Awaitable## Bokeh importsifTYPE_CHECKING:from..document.eventsimportDocumentPatchedEventfrom..protocolimportProtocol,messagesasmsgfrom..protocol.messageimportMessagefrom.contextsimportApplicationContextfrom.sessionimportServerSessionfrom.views.wsimportWSHandler#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------__all__=('ServerConnection',)#-----------------------------------------------------------------------------# General API#-----------------------------------------------------------------------------
[docs]classServerConnection:''' Wraps a websocket connection to a client. .. autoclasstoc:: '''_session:ServerSession|Nonedef__init__(self,protocol:Protocol,socket:WSHandler,application_context:ApplicationContext,session:ServerSession)->None:self._protocol=protocolself._socket=socketself._application_context=application_contextself._session=sessionself._session.subscribe(self)self._ping_count=0@propertydefsession(self)->ServerSession:assertself._sessionisnotNonereturnself._session@propertydefapplication_context(self)->ApplicationContext:returnself._application_context
[docs]defdetach_session(self)->None:"""Allow the session to be discarded and don't get change notifications from it anymore"""ifself._sessionisnotNone:self._session.unsubscribe(self)self._session=None
[docs]defsend_patch_document(self,event:DocumentPatchedEvent)->Awaitable[None]:""" Sends a PATCH-DOC message, returning a Future that's completed when it's written out. """msg=self.protocol.create('PATCH-DOC',[event])# yes, *return* the awaitable, it will be awaited when pending writes are processedreturnself._socket.send_message(msg)