code: def test_lots_of_concurrent_messages(
      self, ManagedServerLoop: MSL
  ) -> None:
      application = Application()
  
      def setup_stuff(doc):
          m1 = AnotherModelInTestClientServer(bar=43, name="m1")
          m2 = SomeModelInTestClientServer(
              foo=42, name="m2", data=bytes([0x00, 0x01, 0xFE, 0xFF])
          )
          m3 = SomeModelInTestClientServer(
              foo=68, name="m3", data=bytes([0x00, 0x01, 0xFE, 0xFF])
          )
          doc.add_root(m1)
          doc.add_root(m2)
          doc.add_root(m3)
  
          def timeout1():
              m1.bar += 1
  
          timeout1_cb_id = doc.add_timeout_callback(timeout1, 1)
  
          def timeout2():
              m2.foo += 1
  
          timeout2_cb_id = doc.add_timeout_callback(timeout2, 3)
  
          def periodic1():
              m1.bar += 1
              doc.remove_timeout_callback(timeout1_cb_id)
              doc.add_timeout_callback(timeout1, m1.bar % 7)
  
          doc.add_periodic_callback(periodic1, 3)
  
          def periodic2():
              m2.foo += 1
              doc.remove_timeout_callback(timeout2_cb_id)
              doc.add_timeout_callback(timeout2, m2.foo % 7)
  
          doc.add_periodic_callback(periodic2, 1)
  
          def server_on_change(event):
              if (
                  isinstance(event, ModelChangedEvent)
                  and event.model is m3
              ):
                  return
              m3.foo += 1
  
          doc.on_change(server_on_change)
  
      handler = FunctionHandler(setup_stuff)
      application.add(handler)
  
      # keep_alive_milliseconds=1 sends pings as fast as the OS will let us
      with ManagedServerLoop(
          application, keep_alive_milliseconds=1
      ) as server:
          session = pull_session(
              session_id=ID("test_lots_of_concurrent_messages"),
              url=url(server),
              io_loop=server.io_loop,
          )
          assert session.connected
  
          server_session = server.get_session("/", session.id)
  
          def client_timeout():
              m = session.document.roots[0]
              m.name = m.name[::-1]
  
          cb_id = session.document.add_timeout_callback(
              client_timeout, 3
          )
  
          def client_periodic():
              m = session.document.roots[1]
              m.name = m.name[::-1]
              session.document.remove_timeout_callback(cb_id)
              session.document.add_timeout_callback(client_timeout, 3)
  
          session.document.add_periodic_callback(client_periodic, 1)
  
          result = {}
  
          def end_test():
              result["connected"] = session.connected
              result[
                  "server_connection_count"
              ] = server_session.connection_count
              result["server_close_code"] = next(
                  iter(server._tornado._clients)
              )._socket.close_code
              result["doc"] = session.document.to_json()
              session.close()
  
          # making this longer is more likely to trigger bugs, but it also
          # makes the test take however much time you put here
          session.document.add_timeout_callback(end_test, 250)
  
          def client_on_change(event):
              if not isinstance(event, TitleChangedEvent):
                  session.document.title = session.document.title[::-1]
  
          session.document.on_change(client_on_change)
  
          session._loop_until_closed()
  
          assert not session.connected
  
          # we should have still been connected at the end,
          # if we didn't have any crazy protocol errors
          assert "connected" in result
          assert result["connected"]
  
          # server should also still have been connected
          assert result["server_connection_count"] == 1
          assert result["server_close_code"] is None
  
imports: from __future__ import annotations
  import pytest
  import asyncio
  import logging
  import sys
  from unittest.mock import MagicMock, patch
  from tornado.httpclient import HTTPError
  import bokeh.document
  from bokeh.application import Application
  from bokeh.application.handlers import FunctionHandler
  from bokeh.client import ClientSession, pull_session, push_session
  from bokeh.core.properties import AngleSpec, Any, Bytes, Dict, DistanceSpec, Instance, Int, Nullable, String
  from bokeh.core.types import ID
  from bokeh.document import Document
  from bokeh.document.events import ModelChangedEvent, TitleChangedEvent
  from bokeh.model import Model
  from bokeh.models import ColumnDataSource, Plot
  from bokeh.util.token import generate_jwt_token
  from tests.support.plugins.managed_server_loop import MSL
  from tests.support.util.env import envset
  from server._util_server import http_get, url, websocket_open, ws_url
label_desc: Use image node_id 21 for calling the TestClientServer obj's underlying member method code with example usage: obj.test_lots_of_concurrent_messages(ManagedServerLoop) without return types