dddp/websocket.py
def send(self, data, tx_id=None):
        """Send `data` (raw string or EJSON payload) to WebSocket client."""
        # buffer data until we get pre-requisite data
        if tx_id is None:
            tx_id = self.get_tx_id()
        self._tx_buffer[tx_id] = data

        # de-queue messages from buffer
        while self._tx_next_id in self._tx_buffer:
            # pull next message from buffer
            data = self._tx_buffer.pop(self._tx_next_id)
            if self._tx_buffer:
                safe_call(self.logger.debug, 'TX found %d', self._tx_next_id)
            # advance next message ID
            self._tx_next_id = next(self._tx_next_id_gen)
            if not isinstance(data, basestring):
                # ejson payload
                msg = data.get('msg', None)
                if msg in (ADDED, CHANGED, REMOVED):
                    ids = self.remote_ids[data['collection']]
                    meteor_id = data['id']
                    if msg == ADDED:
                        if meteor_id in ids:
                            msg = data['msg'] = CHANGED
                        else:
                            ids.add(meteor_id)
                    elif msg == CHANGED:
                        if meteor_id not in ids:
                            # object has become visible, treat as `added`.
                            msg = data['msg'] = ADDED
                            ids.add(meteor_id)
                    elif msg == REMOVED:
                        try:
                            ids.remove(meteor_id)
                        except KeyError:
                            continue  # client doesn't have this, don't send.
                data = 'a%s' % ejson.dumps([ejson.dumps(data)])
            # send message
            safe_call(self.logger.debug, '> %s %r', self, data)
            try:
                self.ws.send(data)
            except geventwebsocket.WebSocketError:
                self.ws.close()
                self._tx_buffer.clear()
                break
        num_waiting = len(self._tx_buffer)
        if num_waiting > 10:
            safe_call(
                self.logger.warn,
                'TX received %d, waiting for %d, have %d waiting: %r.',
                tx_id, self._tx_next_id, num_waiting, self._tx_buffer,
            )
def send(self, data, tx_id=None):
        """Send `data` (raw string or EJSON payload) to WebSocket client."""
        # buffer data until we get pre-requisite data
        if tx_id is None:
            tx_id = self.get_tx_id()
        self._tx_buffer[tx_id] = data

        # de-queue messages from buffer
        while self._tx_next_id in self._tx_buffer:
            # pull next message from buffer
            data = self._tx_buffer.pop(self._tx_next_id)
            if self._tx_buffer:
                safe_call(self.logger.debug, 'TX found %d', self._tx_next_id)
            # advance next message ID
            self._tx_next_id = next(self._tx_next_id_gen)
            if not isinstance(data, basestring):
                # ejson payload
                msg = data.get('msg', None)
                if msg in (ADDED, CHANGED, REMOVED):
                    ids = self.remote_ids[data['collection']]
                    meteor_id = data['id']
                    if msg == ADDED:
                        if meteor_id in ids:
                            msg = data['msg'] = CHANGED
                        else:
                            ids.add(meteor_id)
                    elif msg == CHANGED:
                        if meteor_id not in ids:
                            # object has become visible, treat as `added`.
                            msg = data['msg'] = ADDED
                            ids.add(meteor_id)
                    elif msg == REMOVED:
                        try:
                            ids.remove(meteor_id)
                        except KeyError:
                            continue  # client doesn't have this, don't send.
                data = 'a%s' % ejson.dumps([ejson.dumps(data)])
            # send message
            safe_call(self.logger.debug, '> %s %r', self, data)
            try:
                self.ws.send(data)
            except geventwebsocket.WebSocketError:
                self.ws.close()
                self._tx_buffer.clear()
                break
        num_waiting = len(self._tx_buffer)
        if num_waiting > 10:
            safe_call(
                self.logger.warn,
                'TX received %d, waiting for %d, have %d waiting: %r.',
                tx_id, self._tx_next_id, num_waiting, self._tx_buffer,
            )
Send `data` (raw string or EJSON payload) to WebSocket client.