code: def test_server_shutdown() -> None:
      def run_test(client_ops: Callable) -> None:
          server = LogStreamingServer()
          server.start()
          time.sleep(1)
          client = LogStreamingClient("localhost", server.port)
          client_ops(client)
          server.shutdown()
          client.close()
  
      def client_ops_close(client: Any) -> None:
          client.close()
  
      def client_ops_send_half_msg(client: Any) -> None:
          # Test server only recv incomplete message from client can exit.
          client._connect()
          client.sock.sendall(b"msg part1 ")
          time.sleep(_SERVER_POLL_INTERVAL + 1)
  
      def client_ops_send_a_msg(client: Any) -> None:
          client.send("msg1")
          time.sleep(_SERVER_POLL_INTERVAL + 1)
  
      def client_ops_send_a_msg_and_close(client: Any) -> None:
          client.send("msg1")
          client.close()
          time.sleep(_SERVER_POLL_INTERVAL + 1)
  
      run_test(client_ops_close)
      run_test(client_ops_send_half_msg)
      run_test(client_ops_send_a_msg)
      run_test(client_ops_send_a_msg_and_close)
  
file_path: spark/python/pyspark/ml/torch/tests/test_log_communication.py
imports: from __future__ import absolute_import, division, print_function
  import contextlib
  from io import StringIO
  import sys
  import time
  from typing import Any, Callable
  import unittest
  import pyspark.ml.torch.log_communication
  from pyspark.ml.torch.log_communication import LogStreamingServer, LogStreamingClient, LogStreamingClientBase, _SERVER_POLL_INTERVAL
label_desc: Use image node_id 8 for calling the LogStreamingServiceTestCase obj's underlying member method code with example usage: obj.test_server_shutdown() without return types