gistfile1.txt

go(DbName, DDoc, IndexName, HttpResp, #index_query_args{bookmark=nil,limit=all}=QueryArgs) ->
    Shards = dreyfus_util:get_shards(DbName, QueryArgs),
    Workers0 = fabric_util:submit_jobs(Shards, dreyfus_rpc2, search_all,
                          [DDoc, IndexName, HttpResp, dreyfus_util:export(QueryArgs)]),
    Counters = fabric_dict:init(Workers0, nil),
    RexiMon = fabric_util:create_monitors(Workers0),
    try
        case fabric_util:stream_start(Workers0, #shard.ref) of
            {ok, Workers} ->
                try
                    go(DbName, DDoc, IndexName, QueryArgs, Counters, Counters,HttpResp)
                after
                    fabric_util:cleanup(Workers)
                end;
            {timeout, NewState} ->
                DefunctWorkers = fabric_util:remove_done_workers(
                    NewState#stream_acc.workers, waiting
                ),
                fabric_util:log_timeout(
                    DefunctWorkers,
                    "stream_all"
                ),
                {error, timeout};
            {error, Error} ->
                {error,Error}
        end
    after
        rexi_monitor:stop(RexiMon)
    end.