Python blinker 模块，signal() 实例源码

def subscribe(self, points, stream, startdatetime=None, enddatetime=None, callback=None):
        """Monitor whenever the PI point is read and an update has occurred. 
        Trigger the callback function when the value changes

        :param Points points: List of Point objects to start monitoring
        :param string stream: Name of the reading method used for monitoring the point. 
            Options are current, interpolatedattimes, recordedattime, end
        :param string startdatetime: Optional – Timestamp for when to start monitoring
        :param string enddatetime: Optional – Timestamp for when to stop monitoring
        :param func callback: Reference to the function to trigger when an update occurs
        """
        if not isinstance(points, Points):
            raise TypeError('The object "{}" is not of type "{}"'.format(
                points, Points))
        for p in points:
            formattedstartdate = self._parse_timestamp(startdatetime)
            formattedenddate = self._parse_timestamp(enddatetime)

            signalkey = '{}/{}/{}{}'.format(p.webid.__str__(), stream, formattedstartdate or '', formattedenddate or '')
            if signalkey not in self.signals:
                s = blinker.signal(signalkey)
                self.signals[signalkey] = s
                if callback:
                    self.signals[signalkey].connect(callback)
        return self.signals

def insert_many(cls, documents):
        """Insert a list of documents"""
        from mongoframes.queries import to_refs

        # Ensure all documents have been converted to frames
        frames = cls._ensure_frames(documents)

        # Send insert signal
        signal('insert').send(cls, frames=frames)

        # Prepare the documents to be inserted
        documents = [to_refs(f._document) for f in frames]

        # Bulk insert
        ids = cls.get_collection().insert_many(documents).inserted_ids

        # Apply the Ids to the frames
        for i, id in enumerate(ids):
            frames[i]._id = id

        # Send inserted signal
        signal('inserted').send(cls, frames=frames)

        return frames

def receive_data(sender, **kw):
    print ("Caught signal from : {}, data: {}".format(sender, kw))
    test = 1 / 0
    return test

def receive_data(sender, **kw):
    print ("Caught signal from : {}, data: {}".format(sender, kw))
    time.sleep(5)
    return "received"

def _process(self, func, args, kwargs, request, response):
        start_sig = blinker.signal('BeforeFunctionExecute')
        start_sig.send(request)

        if not self.auth_func(request, func):
            raise Unauthorized

        if self.thrift_wrapper:
            return self.thrift_wrapper(func)(*args, **kwargs)
        return self._http_wrapper(data=func(*args, **kwargs))

def _finish_exec(self, duration, func_logger, args, kwargs, request, func):
        end_sig = blinker.signal('AfterFunctionExecute')
        end_sig.send(request)
        func_logger.info(self._prepare_log(func.__name__, args, kwargs, duration, request))

def client_dispatcher_factory(client_context):
    def perform_push_client_msg(intent):
        client_context.queued_pushed_events.put_nowait(intent.payload)

    @do
    def perform_client_subscribe_event(intent):
        yield Effect(ERegisterEvent(EClientEvent, intent.event, intent.sender))
        # key = (intent.event, intent.sender)

        # def on_event(sender):
        #     payload = ejson_dumps({'event': intent.event, 'sender': sender})
        #     client_context.queued_pushed_events.put_nowait(payload)

        # # Attach the callbacks to the client context to make them have the same
        # # lifetime given event registration expires when callback is destroyed
        # # TODO: allow a subset of the possible events
        # client_context.subscribed_events[key] = on_event
        # signal(intent.event).connect(on_event, sender=intent.sender)

    @do
    def perform_client_unsubscribe_event(intent):
        yield Effect(EUnregisterEvent(EClientEvent, intent.event, intent.sender))
        # key = (intent.event, intent.sender)
        # try:
        #     del client_context.subscribed_events[key]
        # except KeyError:
        #     pass

    def perform_client_event(intent):
        payload = ejson_dumps({'event': intent.event, 'sender': intent.sender})
        client_context.queued_pushed_events.put_nowait(payload)

    return TypeDispatcher({
        EPushClientMsg: perform_push_client_msg,
        EClientSubscribeEvent: perform_client_subscribe_event,
        EClientUnsubscribeEvent: perform_client_unsubscribe_event,
        EClientEvent: perform_client_event
    })

def __init__(self, name):
        self.name = name
        self.event = signal(name)

def assert_signals(test, signal_name, fun=lambda x, n: True):
    if not isinstance(signal_name, (tuple, list)):
        signal_name = [signal_name]
    success = [False]*len(signal_name)
    loggers = [
        log_signal(i, n, success, fun)
        for i, n in
        enumerate(signal_name)
    ]
    for i, n in enumerate(signal_name):
        signal(n).connect(loggers[i])
    yield
    for n, s in zip(signal_name, success):
        test.assertTrue(s, msg=n + " signal was not received")

def send(self, sensor: Sensor):
        event = signal("sensor_update")
        self.sensor = sensor
        event.send(self)

def send(self, name: str, new_state) -> None:
        event = signal("change_actuator_request")
        self.__name = name
        self.__new_state = new_state
        event.send(self)

def send(self, device_name: str, latitude: float, longitude: float):
        event = signal("location")
        self.__device_name = device_name
        self.__latitude = latitude
        self.__longitude = longitude
        event.send(self)

def register(self, sid, callback, keys=(), tfilter=None):
        """Add new request to request pool

        sid: Subscriber ID (any hashable)
        callback: Callable that receives a tuple of Torrents on updates
        keys: Wanted Torrent keys
        tfilter: None for all torrents or TorrentFilter instance
        """
        if isinstance(tfilter, abc.Sequence):
            tfilter = TorrentFilter('|'.join('id=%s' % tid for tid in tfilter))

        log.debug('Registering subscriber: %s', sid)
        event = blinker.signal(sid)
        event.connect(callback)
        self._keys[event] = tuple(keys)
        self._tfilters[event] = tfilter

        # It's possible that a currently ongoing request doesn't collect the
        # keys this new callback needs.  In that case, the request is finished
        # AFTER we added the callback, and the callback would be called with
        # lacking keys, resuling in a KeyError.
        # Therefore we ask the poller to dump the result of a currently
        # ongoing request to prevent this.
        if self.running:
            self.skip_ongoing_request()

        self._combine_requests()

def remove(self, sid):
        """Unsubscribe previously registered subscriber"""
        log.debug('Removing subscriber: %s', sid)
        event = blinker.signal(sid)
        del self._keys[event]
        del self._tfilters[event]
        self._combine_requests()

def on_dc_switch(self):
        # Reload user object in request
        self.request.user = self.request.user.__class__.objects.get(pk=self.request.user.pk)
        self.setup_user()
        self.set_active_user()

        # Inform other sessions for this user about the DC change
        task_id = task_id_from_request(self.user_id, tt=TT_INTERNAL, tg=TG_DC_UNBOUND, dc_id=self.dc_id)
        self.last_tasks.append(task_id)
        new_task = signal('task-for-' + self.user_id)
        new_task.send('_dc_switch', task_id=task_id, event_status='internal')

    # noinspection PyUnusedLocal

def que_monitor(self):
        new_task = signal('task-for-' + self.user_id)

        # noinspection PyUnusedLocal
        @new_task.connect
        def process_task(sender, task_id=None, event_status=None, **kwargs):
            self.log('Got signal for %s task %s', event_status, task_id, level=DEBUG)
            task_prefix = task_prefix_from_task_id(task_id)

            if task_prefix[4] != self.dc_id and task_prefix[3] != TG_DC_UNBOUND:
                self.log('Ignoring dc-bound task %s, because user works in DC %s', task_id, self.dc_id)
                return

            if event_status == 'sent':
                self._task_sent(task_id, event_status, sender, task_prefix[1])
            elif event_status == 'event':
                self._task_event(task_id, sender)
            elif event_status == 'internal':
                self._task_internal(task_id, sender, **kwargs)
            else:
                self._task_status(task_id, event_status, task_prefix[1])

        self.log('Ready')
        self.set_active_user()

        try:
            while True:
                sleep(1.0)
        finally:
            self.log('Game over')
            self.del_active_user()

def register_signal(self):
        signal.signal(signal.SIGINT, self._stop_handler)
        signal.signal(signal.SIGHUP, self._stop_handler)
        signal.signal(signal.SIGTERM, self._stop_handler)
        signal.signal(signal.SIGQUIT, self._stop_handler)

def subscriber(sender):
    print('Got a signal sent by {sender}'.format(sender=sender))


# e.g.3

def go(self):
        ready = signal('ready')
        ready.send(self)
        print('Processing.')
        complete = signal('complete')
        complete.send(self)

def b_subscriber(sender):
    print('Caught signal form processor_b.')
    assert sender.name == 'b'


# e.g.5

def receive_data(sender, **kw):
    print ("Caught signal form {sender}, data {data}".format(sender=sender, data=kw))
    return 'received!'


# e.g.6

def fire_signal(sig, **kwargs):
    #sys.stderr.write('FIRING signal %r\n' % sig)
    from mqe import c
    sender = c
    sig.send(sender, **kwargs)

def register_for_private_messages(self, event_listener=None):
        """
        Register the event_listener for private message events. Be careful, you should ignore your own messages by
        comparing the invoker_id to your client id ...
        :param event_listener: Blinker signal handler function to be informed: on_event(sender, **kw), kw will contain
        the event
        :type event_listener: (str, dict[str, any]) -> None
        """
        self._send("servernotifyregister", ["event=textprivate"])
        if event_listener is not None:
            blinker.signal("event").connect(event_listener)

def register_for_server_events(self, event_listener=None):
        """
        Register event_listener for receiving server_events.
        :param event_listener: Blinker signal handler function to be informed: on_event(sender, **kw), kw will contain
        the event
        :type event_listener: (str, dict[str, any]) -> None
        """
        self._send("servernotifyregister", ["event=server"])
        if event_listener is not None:
            blinker.signal("event").connect(event_listener)

def _parse_resp(self, resp):
        """
        Parses a response. Messages starting with notify... are handled as events and the listeners connected are
        informed. Messages starting with error are split by " " and returned, all other messages will just be returned
         and can be handled by the caller.
        :param resp: Message to parse.
        :type resp: byte
        :return: None if message notifies of an event, dictionary containing id and message on acknowledgements and
        bytes on any other message.
        :rtype: None | dict[str, str] | bytes
        """
        # Acknowledgements
        if resp.startswith(b'error'):
            resp = resp.split(b' ')
            return resp
        # Events
        elif resp.startswith(b'notify'):
            resp = resp.decode(encoding='UTF-8').split(" ")
            event_type = resp[0]
            event = dict()
            for info in resp[1:]:
                split = info.split('=', 1)
                if len(split) == 2:
                    key, value = split
                    event[key] = utilities.unescape(value)
            event = Events.EventParser.parse_event(event, event_type)
            signal = blinker.signal("event")
            self._logger.debug("Sending signal")
            threading.Thread(target=signal.send, kwargs={'event': event}).start()
            return None
        # Query-Responses and other things(What could these be?)
        else:
            return resp

def quit(self):
        """
        Stops the connection from receiving and sends the quit signal.
        """
        # Avoid unclean exit by interfering with response to pending query
        if self._tel_lock.acquire():
            self.stop_recv.set()
        self._tel_lock.release()
        self._send("quit")

def insert(self):
        """Insert this document"""
        from mongoframes.queries import to_refs

        # Send insert signal
        signal('insert').send(self.__class__, frames=[self])

        # Prepare the document to be inserted
        document = to_refs(self._document)

        # Insert the document and update the Id
        self._id = self.get_collection().insert_one(document).inserted_id

        # Send inserted signal
        signal('inserted').send(self.__class__, frames=[self])

def update(self, *fields):
        """
        Update this document. Optionally a specific list of fields to update can
        be specified.
        """
        from mongoframes.queries import to_refs

        assert '_id' in self._document, "Can't update documents without `_id`"

        # Send update signal
        signal('update').send(self.__class__, frames=[self])

        # Check for selective updates
        if len(fields) > 0:
            document = {}
            for field in fields:
                document[field] = self._path_to_value(field, self._document)
        else:
            document = self._document

        # Prepare the document to be updated
        document = to_refs(document)
        document.pop('_id', None)

        # Update the document
        self.get_collection().update_one({'_id': self._id}, {'$set': document})

        # Send updated signal
        signal('updated').send(self.__class__, frames=[self])

def delete(self):
        """Delete this document"""

        assert '_id' in self._document, "Can't delete documents without `_id`"

        # Send delete signal
        signal('delete').send(self.__class__, frames=[self])

        # Delete the document
        self.get_collection().delete_one({'_id': self._id})

        # Send deleted signal
        signal('deleted').send(self.__class__, frames=[self])

def update_many(cls, documents, *fields):
        """
        Update multiple documents. Optionally a specific list of fields to
        update can be specified.
        """
        from mongoframes.queries import to_refs

        # Ensure all documents have been converted to frames
        frames = cls._ensure_frames(documents)

        all_count = len(documents)
        assert len([f for f in frames if '_id' in f._document]) == all_count, \
                "Can't update documents without `_id`s"

        # Send update signal
        signal('update').send(cls, frames=frames)

        # Prepare the documents to be updated

        # Check for selective updates
        if len(fields) > 0:
            documents = []
            for frame in frames:
                document = {'_id': frame._id}
                for field in fields:
                    document[field] = cls._path_to_value(
                        field,
                        frame._document
                        )
                documents.append(to_refs(document))
        else:
            documents = [to_refs(f._document) for f in frames]

        # Update the documents
        for document in documents:
            _id = document.pop('_id')
            cls.get_collection().update(
                {'_id': _id}, {'$set': document})

        # Send updated signal
        signal('updated').send(cls.__class__, frames=frames)

def listen(cls, event, func):
        """Add a callback for a signal against the class"""
        signal(event).connect(func, sender=cls)

def stop_listening(cls, event, func):
        """Remove a callback for a signal against the class"""
        signal(event).disconnect(func, sender=cls)

    # Misc.

def test_blinker(self):
        # a few tests so we know how blinker works
        self.assertEqual( signals.creation, blinker.signal('creation') )

        # subscribe
        signals.creation.connect( lambda instance: None )

def send(self):
        msg_signal = signal('msg_signal')
        msg_signal.send(self)  # I can get msg use self.msg from socketio register function:)

def send(self):
        msg_signal = signal('msg_signal')
        msg_signal.send(self)  # I can get msg use self.msg from socketio register function:)