Resend all unprocessed items
355177
def resend_unprocessed(self):
        
        LOG.info("Re-sending %d unprocessed items.", len(self._unprocessed))

        while self._unprocessed:
            to_resend = self._unprocessed[:MAX_WRITE_BATCH]
            self._unprocessed = self._unprocessed[MAX_WRITE_BATCH:]
            LOG.info("Sending %d items", len(to_resend))
            self._write(to_resend)
            LOG.info("%d unprocessed items left", len(self._unprocessed))
156538
def collapse_all(self):
        """"""

        def aux(item):
            self.item(item, open=False)
            children = self.get_children(item)
            for c in children:
                aux(c)

        children = self.get_children("")
        for c in children:
            aux(c)
1107928
def dedup(seq):
    """"""
    seen = set()
    for item in seq:
        if item not in seen:
            seen.add(item)
            yield item
518575
def remove_each(text, items, count=None, strip=False):
    '''
    
    '''
    for item in items:
        text = remove(text, item, count=count, strip=strip)
    return text
434374
def cleanup(self):
        
        empty_keys = [k for k, v in self.items() if not v]

        for k in empty_keys:
            del self[k]
674040
async def reset_all(self, full=True):
        
        conn = await self.redis()

        if full:
            await conn.flushdb()
        else:
            keys = await conn.keys(self.generate_key('*'))
            await conn.delete(*keys)
720227
def _clean_item(self, item):
        '''
        
        '''
        item_copy = dict(item)
        del item_copy['body']
        del item_copy['links']
        del item_copy['response_headers']
        del item_copy['request_headers']
        del item_copy['status_code']
        del item_copy['status_msg']
        item_copy['action'] = 'ack'
        item_copy['logger'] = self.logger.name
        item_copy

        return item_copy
765448
function asyncMap(array, iterator, done) {
  if (!array || !array.length) {
    return done(null, [])
  }

  var error
    , out = []
    , todo = array.length

  array.forEach(exec)

  function exec(item) {
    process.nextTick(function tick() {
      iterator(item, add)
    })
  }

  function add(err, data) {
    if (error) {
      return
    }
    else if (err) {
      error = true
      return done(err)
    }

    out.push(data)

    if (--todo === 0) {
      done(null, out)
    }
  }
}
521002
function (items) {
		var args = slice(items);
		args.unshift(0);
		this.empty();
		this.insert.apply(this, args);
		return this.length;
	}
559936
def cleanup(self):
        
        remove = []
        for handle, assoc in self.assocs.items():
            if assoc.expiresIn == 0:
                remove.append(handle)
        for handle in remove:
            del self.assocs[handle]
        return len(remove), len(self.assocs)
1353097
function cleanRes (res) {
    return {
        body: res.body,
        rawHeaders: res.rawHeaders,
        statusCode: res.statusCode,
    }
}
764529
function (item) {
		var items = this.items;
		var removed = 0;
		for (var i = 0, l = items.length; i < l; i++) {
			if (items[i] === item) {
				items.splice(i, 1);
				i--;
				removed++;
			}
		}
		return removed;
	}
580114
private boolean appendList(List<Object> formattedEvents) {
        if (newLists.size() >= clientPoolSize)
            return false;
        boolean done = false;
        while (!done) {
            try {
                newLists.putLast(formattedEvents);
                done = true;
            } catch (InterruptedException e) {
                //Interrupted try again
            }
        }
        return true;
    }
685330
def append(self, val):
        """"""
        val.scraper = self.scraper
        val._collection_path = copy(self.collection._collection_path)
        val._collection_path.append(val)
        super(ItemList, self).append(val)
569561
def flush(self):
        
        items = []

        for data in self._to_put:
            items.append(encode_put(self.connection.dynamizer, data))

        for data in self._to_delete:
            items.append(encode_delete(self.connection.dynamizer, data))
        self._write(items)
        self._to_put = []
        self._to_delete = []
1170880
def deferred_cleanup(fn):
    """"""
    @functools.wraps(fn)
    def ret(*args, **kwargs):
        defers = []
        try:
            ret = fn(lambda *args: defers.extend(args), *args, **kwargs)
        finally:
            for defer in reversed(defers):
                defer()
        return ret
    return ret
648683
function (start/*, items... */) {
		var items = slice(arguments, 1);
		var item, index;
		for (var i = 0, l = items.length; i < l; i++) {
			item = items[i];
			index = start + i;
			$Array.splice.call(this, index, 0, item);
			this.trigger(EVENT_ADD, item, index);
		}
		return this.length;
	}