Function queryBackend

cmd/dcs-web/querymanager.go:183–273

(ctx context.Context, queryid, src string, backend sourcebackendpb.SourceBackendClient, backendidx int, searchRequest *sourcebackendpb.SearchRequest)

)

func queryBackend(ctx context.Context, queryid, src string, backend sourcebackendpb.SourceBackendClient, backendidx int, searchRequest *sourcebackendpb.SearchRequest) {
	// When exiting this function, check that all results were processed. If
	// not, the backend query must have failed for some reason. Send a progress
	// update to prevent the query from running forever.
	defer func() {
		stateMu.RLock()
		s, ok := state[queryid]
		stateMu.RUnlock()
		if !ok {
			return // query no longer exists
		}
		s.filesMu.Lock()
		filesTotal := s.filesTotal[backendidx]
		filesProcessed := s.filesProcessed[backendidx]
		s.filesMu.Unlock()

		if filesProcessed == filesTotal {
			return
		}

		if filesTotal == -1 {
			filesTotal = 0
		}

		storeProgress(queryid, backendidx, &sourcebackendpb.ProgressUpdate{
			FilesProcessed: uint64(filesTotal),
			FilesTotal:     uint64(filesTotal),
		})

		addEventMarshal(queryid, &Error{
			Type:      "error",
			ErrorType: "backendunavailable",
		})
	}()

	ctx, cancelfunc := context.WithCancel(ctx)
	stream, err := backend.Search(ctx, searchRequest)
	if err != nil {
		log.Printf("[%s] [src:%s] Search RPC failed: %v\n", queryid, src, err)
		return
	}

	stateMu.RLock()
	s, ok := state[queryid]
	stateMu.RUnlock()
	if !ok {
		log.Printf("[%s] [src:%s] query no longer exists\n", queryid, src)
		return
	}
	bstate := s.perBackend[backendidx]
	tempFileWriter := bstate.tempFileWriter
	orderlyFinished := false
	done := false

	for !done {
		msg, err := stream.Recv()
		if err == io.EOF {
			log.Printf("[%s] [src:%s] EOF\n", queryid, src)