code: def _recv_thread(self):
          
          for msg, metadata in self._source:
              if msg.msg_type:
                  self._call(msg, **metadata)
          # Break any upstream iterators
          for sink in self._sinks:
              i = sink()
              if i is not None:
                  i.breakiter()
          self._dead = True
docs: Internal thread to iterate over source messages and dispatch callbacks.