How to use session_cleanup_callback method in Slash

Best Python code snippet using slash

jupyter_kernel.py

Source:jupyter_kernel.py Github

copy

Full Screen

...629 #630 # now a couple of things are connected, call the session_cleanup_callback631 #632 if self.task_cnt > 1 and self.session_cleanup_callback:633 self.session_cleanup_callback()634 self.session_cleanup_callback = None635 elif msg[0] == "unregister":636 if msg[1] in self.tasks:637 self.tasks[msg[1]].discard(msg[2])638 self.task_cnt -= 1639 #640 # if there are no connection tasks left, then shutdown the kernel641 #642 if self.task_cnt == 0 and self.task_cnt_max >= 4:643 asyncio.create_task(self.session_shutdown())644 await asyncio.sleep(10000)645 elif msg[0] == "shutdown":646 asyncio.create_task(self.session_shutdown())647 await asyncio.sleep(10000)648 except asyncio.CancelledError:649 raise650 except Exception:651 _LOGGER.error("housekeep task exception: %s", traceback.format_exc(-1))652 async def startup_timeout(self):653 """Shut down the session if nothing connects after 30 seconds."""654 await self.housekeep_q.put(["register", "startup_timeout", asyncio.current_task()])655 await asyncio.sleep(self.no_connect_timeout)656 if self.task_cnt_max <= 1:657 #658 # nothing started other than us, so shut down the session659 #660 _LOGGER.error("No connections to session %s; shutting down", self.global_ctx_name)661 if self.session_cleanup_callback:662 self.session_cleanup_callback()663 self.session_cleanup_callback = None664 await self.housekeep_q.put(["shutdown"])665 await self.housekeep_q.put(["unregister", "startup_timeout", asyncio.current_task()])666 async def start_one_server(self, callback):667 """Start a server by finding an available port."""668 first_port = self.avail_port669 for _ in range(2048):670 try:671 server = await asyncio.start_server(callback, "0.0.0.0", self.avail_port)672 return server, self.avail_port673 except OSError:674 self.avail_port += 1675 _LOGGER.error(676 "unable to find an available port from %d to %d", first_port, self.avail_port - 1,677 )678 return None, None679 def get_ports(self):680 """Return a dict of the port numbers this kernel session is listening to."""681 return {682 "iopub_port": self.iopub_port,683 "hb_port": self.heartbeat_port,684 "control_port": self.control_port,685 "stdin_port": self.stdin_port,686 "shell_port": self.shell_port,687 }688 def set_session_cleanup_callback(self, callback):689 """Set a cleanup callback which is called right after the session has started."""690 self.session_cleanup_callback = callback691 async def session_start(self):692 """Start the kernel session."""693 self.ast_ctx.add_logger_handler(self.console)694 _LOGGER.info("Starting session %s", self.global_ctx_name)695 self.tasks["housekeep"] = {asyncio.create_task(self.housekeep_run())}696 self.tasks["startup_timeout"] = {asyncio.create_task(self.startup_timeout())}697 self.iopub_server, self.iopub_port = await self.start_one_server(self.iopub_listen)698 self.heartbeat_server, self.heartbeat_port = await self.start_one_server(self.heartbeat_listen)699 self.control_server, self.control_port = await self.start_one_server(self.control_listen)700 self.stdin_server, self.stdin_port = await self.start_one_server(self.stdin_listen)701 self.shell_server, self.shell_port = await self.start_one_server(self.shell_listen)702 #...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Slash automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful