def _recv_thread(self):
for msg, metadata in self._source:<EOL><INDENT>if msg.msg_type:<EOL><INDENT>self._call(msg, **metadata)<EOL><DEDENT><DEDENT>for sink in self._sinks:<EOL><INDENT>i = sink()<EOL>if i is not None:<EOL><INDENT>i.breakiter()<EOL><DEDENT><DEDENT>self._dead = True<EOL>
Internal thread to iterate over source messages and dispatch callbacks.