Function compareNDJSONStream

pkg/util.go:1178–1246

(expectedResp models.HTTPResp, stream io.Reader, jsonNoiseKeys map[string]struct{}, logger *zap.Logger)

}

func compareNDJSONStream(expectedResp models.HTTPResp, stream io.Reader, jsonNoiseKeys map[string]struct{}, logger *zap.Logger) (bool, string, *StreamMismatchInfo, error) {
	expectedQueue := extractExpectedRawQueue(expectedResp, canonicalizeNDJSONLine, true)
	actualQueue := make([]string, 0, len(expectedQueue))
	nextExpected := 0

	scanner := bufio.NewScanner(stream)
	scanner.Buffer(make([]byte, 0, 64*1024), maxStreamTokenSize)

	for scanner.Scan() {
		line := canonicalizeNDJSONLine(scanner.Text())
		if line == "" {
			continue
		}

		if nextExpected >= len(expectedQueue) {
			logger.Debug("received additional NDJSON data after expected stream was fully matched; closing stream capture",
				zap.Int("expected_frames", len(expectedQueue)))
			break
		}

		actualQueue = append(actualQueue, line)
		expected := expectedQueue[nextExpected]
		ok, cmpErr := compareJSONTextWithNoise(expected, line, jsonNoiseKeys)
		if cmpErr != nil || !ok {
			reason := "json mismatch"
			if cmpErr != nil {
				reason = cmpErr.Error()
			}
			logger.Debug("NDJSON stream mismatch",
				zap.Int("frame_index", nextExpected),
				zap.String("reason", reason),
				zap.String("expected_frame", expected),
				zap.String("actual_frame", line))
			mismatchInfo := &StreamMismatchInfo{
				FrameIndex:    nextExpected,
				ExpectedFrame: expected,
				ActualFrame:   line,
				Reason:        reason,
			}
			return false, strings.Join(actualQueue, "\n"), mismatchInfo, nil
		}

		nextExpected++
		if nextExpected == len(expectedQueue) {
			logger.Debug("all expected NDJSON frames matched; closing stream capture early to avoid waiting for extra stream events",
				zap.Int("matched_frames", nextExpected))
			break
		}
	}

	if err := scanner.Err(); err != nil {
		return false, strings.Join(actualQueue, "\n"), nil, err
	}

	if nextExpected < len(expectedQueue) {
		logger.Debug("NDJSON stream ended before all expected frames were received",
			zap.Int("expected_frames", len(expectedQueue)),
			zap.Int("matched_frames", nextExpected))