code: def coalescing_reply(
      self, raw_msg, msg, original_msg_id, outgoing_id, idents
  ):
      # accumulate buffers
      self.accumulated_replies[original_msg_id][outgoing_id] = msg[
          "buffers"
      ]
      if all(
          msg_buffers is not None
          for msg_buffers in self.accumulated_replies[
              original_msg_id
          ].values()
      ):
          replies = self.accumulated_replies.pop(original_msg_id)
          self.log.debug(
              f"Coalescing {len(replies)} reply to {original_msg_id}"
          )
          targets = self.accumulated_targets.pop(original_msg_id)
  
          new_msg = msg.copy()
          # begin rebuilding message
          # metadata['targets']
          if self.is_leaf:
              new_msg["metadata"]["broadcast_targets"] = targets
          else:
              new_msg["metadata"]["broadcast_targets"] = []
  
          # avoid duplicated msg buffers
          buffers = []
          for sub_target, msg_buffers in replies.items():
              buffers.extend(msg_buffers)
              if not self.is_leaf:
                  new_msg["metadata"]["broadcast_targets"].extend(
                      targets[sub_target]
                  )
  
          new_raw_msg = self.session.serialize(new_msg)
          self.client_stream.send_multipart(
              idents + new_raw_msg + buffers, copy=False
          )
  
file_path: ipyparallel/ipyparallel/controller/broadcast_scheduler.py
imports: import logging
  import zmq
  from traitlets import Bool, Bytes, Integer, List, Unicode
  from ipyparallel import util
  from ipyparallel.controller.scheduler import Scheduler, ZMQStream, get_common_scheduler_streams
label_desc: Use image node_id 4 for calling the BroadcastScheduler obj's underlying member method code with example usage: obj.coalescing_reply(raw_msg, msg, original_msg_id, outgoing_id, idents) without return types