def doc_feeder(self, step=None, verbose=True, query=None, scroll='10m', only_source=True, **kwargs):
        step = step or self.step
        q = query if query else {'query': {'match_all': {}}}
        _q_cnt = self.count(q=q, raw=True)
        n = _q_cnt['count']
        n_shards = _q_cnt['_shards']['total']
        assert n_shards == _q_cnt['_shards']['successful']
        # Not sure if scroll size is per shard anymore in the new ES...should check this
        _size = int(step / n_shards)
        assert _size * n_shards == step
        cnt = 0
        # t0 = time.time()
        # if verbose:
        #     t1 = time.time()

        res = self._es.search(index=self._index, doc_type=self._doc_type, body=q,
                              size=_size, search_type='scan', scroll=scroll, **kwargs)
        # double check initial scroll request returns no hits
        # assert len(res['hits']['hits']) == 0
        assert not res['hits']['hits']

        while True:
            # if verbose:
            #     t1 = time.time()
            res = self._es.scroll(scroll_id=res['_scroll_id'], scroll=scroll)
            # if len(res['hits']['hits']) == 0:
            if not res['hits']['hits']:
                break
            else:
                for rawdoc in res['hits']['hits']:
                    if rawdoc.get('_source', False) and only_source:
                        doc = rawdoc['_source']
                        doc["_id"] = rawdoc["_id"]
                        yield doc
                    else:
                        yield rawdoc
                    cnt += 1

        assert cnt == n, "Error: scroll query terminated early [{}, {}], please retry.\nLast response:\n{}".format(cnt, n, res)