Method ProcessURLs

runner/runner.go:85–142

ProcessURLs will fetch the HTTP response for all URLs in the provided reader and stream the extracted sources to the runner's Results channel.

(data io.Reader)

// ProcessURLs will fetch the HTTP response for all URLs in the provided reader
// and stream the extracted sources to the runner's Results channel.
func (r *runner) ProcessURLs(data io.Reader) {
	var (
		next = Read(data)
		wg   = sync.WaitGroup{}

		throttle = make(chan struct{}, r.Options.Threads)
	)

	for i := 0; i < r.Options.Threads; i++ {
		throttle <- struct{}{}
	}

	for {
		u, err := next()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			log.Println(fmt.Errorf("[error] parsing url %v: %w", u, err))
			continue
		}

		wg.Add(1)
		go func(u *url.URL) {
			defer func() {
				throttle <- struct{}{}
				wg.Done()
			}()

			resp, err := extractor.FetchResponse(u.String(), r.Options.Request.Method, r.Options.Request.Headers)
			if err != nil {
				log.Println(fmt.Errorf("[error] fetching response for url %s: %w", u.String(), err))
				return
			}
			defer resp.Body.Close()

			sources, err := extractor.ExtractSources(resp.Body)
			if err != nil {
				log.Println(fmt.Errorf("[error] extracting sources from response for url %s: %w", u.String(), err))
				return
			}

			filtered, err := extractor.Filter(sources, r.filters(u)...)
			if err != nil {
				log.Println(fmt.Errorf("[error] filtering sources for url %s: %w", u.String(), err))
				return
			}

			for source := range filtered {
				r.Results <- source
			}
		}(u)

		<-throttle
	}

	wg.Wait()
}