def _dispatch(self, message):
        if LOG.isEnabledFor(logging.TRACE):
            LOG.trace("Dispatching message %s (it took %s seconds"
                      " for it to arrive for processing after being"
                      " sent)", message,
                      timeutils.delta_seconds(message['sent_on'],
                                              timeutils.utcnow()))
        try:
            kind = message['kind']
            sender = message['sender']
            body = message['body']
        except (__HOLE__, ValueError, TypeError):
            LOG.warn("Badly formatted message %s received", message,
                     exc_info=True)
            return
        target = self._targets.get(sender['id'])
        if target is None:
            # Must of been removed...
            return
        if kind == _KIND_COMPLETE_ME:
            target.dispatched[kind] += 1
            target.barrier.set()
        elif kind == _KIND_EVENT:
            task = target.task
            target.dispatched[kind] += 1
            task.notifier.notify(body['event_type'], body['details'])
        else:
            LOG.warn("Unknown message '%s' found in message from sender"
                     " %s to target '%s'", kind, sender, target)
dataset/ETHPy150Open openstack/taskflow/taskflow/engines/action_engine/executor.py/_Dispatcher._dispatch