func handleGrpcSearchStream()

in internal/search/grpc.go [116:178]
57 lines of code
11 McCabe index (conditional complexity)

func handleGrpcSearchStream(ctx context.Context, stream SearchStream, endpoint string) ZoektResponse {
	var matches []*zproto.FileMatch
	var stats zproto.Stats

	for {
		resp, err := stream.Recv()
		if ctxErr := ctx.Err(); ctxErr != nil {
			return ZoektResponse{
				Error:    ctxErr,
				Endpoint: endpoint,
			}
		}

		if err != nil {
			if errors.Is(err, io.EOF) {
				return ZoektResponse{
					Endpoint: endpoint,
					Result: zoekt.SearchResult{
						Stats: zoekt.Stats{
							FileCount:    int(stats.FileCount),
							MatchCount:   int(stats.MatchCount),
							NgramMatches: int(stats.NgramMatches),
						},
						Files: convertGrpcFiles(matches),
					},
				}
			}
			return ZoektResponse{
				Error:    fmt.Errorf("error receiving from upstream stream: %w", err),
				Endpoint: endpoint,
			}
		}

		if resp == nil {
			return ZoektResponse{
				Error:    fmt.Errorf("received nil response from stream without EOF"),
				Endpoint: endpoint,
			}
		}

		chunk := resp.GetResponseChunk()
		if chunk == nil {
			continue
		}

		matches = append(matches, chunk.Files...)
		if chunk.Stats != nil {
			stats.FileCount += chunk.Stats.FileCount
			stats.MatchCount += chunk.Stats.MatchCount
			stats.NgramMatches += chunk.Stats.NgramMatches
		}

		if stats.NgramMatches == 0 {
			var ngramMatchCount int
			for _, file := range matches {
				for _, lm := range file.LineMatches {
					ngramMatchCount += len(lm.LineFragments)
				}
			}
			stats.NgramMatches = int64(ngramMatchCount)
		}
	}
}