def publish()

in ftw_compatible_tool/broker.py [0:0]

• 24 lines of code
• 10 McCabe index (conditional complexity)

    def publish(self, topic, *args, **kwargs):
        """ Publish messages to the topic's subscribers.

        Arguments:
            - topic: A TOPICS value.
            - args, kwargs: Arguments sent to subscribers.
                Type check will be done on these.
        """
        item = self._topic_items.get(topic)
        if item is None or len(item.subscribers) == 0:
            return False
        type_limit = item.type_limit
        subscribers = item.subscribers
        if type_limit:
            for subscriber in subscribers:
                if isinstance(subscriber, types.FunctionType):
                    target_args = inspect.getargspec(subscriber).args
                else:
                    target_args = inspect.getargspec(subscriber.__call__).args
                for i in range(min(len(args), len(target_args))):
                    if not isinstance(args[i], type_limit.get(target_args[i], type(args[i]))):
                        raise ValueError(
                            "type<%s> is not compatible with args<%s : (%s)>" %
                            (type_limit, i, args[i]))
                for k, v in kwargs.items():
                    if not isinstance(v, type_limit.get(k, type(v))):
                        raise ValueError(
                            "type<%s> is not compatible with args<%s : (%s)>" %
                            (type_limit, k, v))
        tuple(map(lambda subscriber: subscriber(*args, **kwargs), subscribers))
        return True