jobs-monorepo/services/scraper-linkedin/pipeline/job_pipeline_workers.go
Elshimy Ziad Magdy Taha 8ac872bdc6 feat: restructure project into monorepo with CI/CD
- Reorganized project structure into services/ and apps/ directories for better separation of concerns
- Added comprehensive CI/CD pipeline with GitHub Actions for testing and Docker builds
- Created .dockerignore file to optimize container builds
- Updated Makefile with new targets for each service and application
- Added detailed README with architecture overview, setup instructions and development guidelines
- Moved cron-analyzer to dedicate
2025-11-01 16:20:44 +05:00

92 lines
2.2 KiB
Go

package pipeline
import (
"context"
"log"
"github.com/jobs-scraper/internal/pkg/domain"
)
func GetJobs(context context.Context, scraperService *Scraper, searchQuery domain.SearchQuery) <-chan domain.Job {
jobChan := make(chan domain.Job)
go func() {
defer close(jobChan)
if err := scraperService.ScrapeLinkedInJobsStreaming(context, searchQuery.NumPages, jobChan, searchQuery); err != nil {
log.Printf("Error scraping jobs: %v", err)
}
}()
return jobChan
}
func GetJobDescription(context context.Context, scraperService *Scraper, jobChan <-chan domain.Job, numWorkers int) <-chan domain.JobWithDescription {
jobDescriptionChan := make(chan domain.JobWithDescription)
go func() {
defer close(jobDescriptionChan)
for job := range jobChan {
select {
case <-context.Done():
return
default:
}
if jd, jc, err := scraperService.ScrapeJobDescriptionWithContext(context, job); err != nil {
log.Printf("Error scraping job description: %v", err)
} else {
jobDescriptionChan <- domain.JobWithDescription{
Job: job,
JobDescription: domain.JobDescription{
JobID: job.ID,
Description: jd,
Criteria: jc,
},
}
}
}
}()
return jobDescriptionChan
// unless you have proxies, you will get rate limited by linkedin,
// but if you do have you can use this code below,
// it will work faster (if u pass more than one worker)
// + dont forget to use buffered channels
// for range numWorkers {
// wg.Add(1)
// go func() {
// defer wg.Done()
// for job := range jobChan {
// select {
// case <-context.Done():
// return
// default:
// }
// // Scrape job description
// if jd, jc, err := scraperService.ScrapeJobDescriptionWithContext(context, job); err != nil {
// log.Printf("Error scraping job description: %v", err)
// } else {
// jobDescriptionChan <- domain.JobWithDescription{
// Job: job,
// JobDescription: domain.JobDescription{
// JobID: job.ID,
// Description: jd,
// Criteria: jc,
// },
// }
// }
// }
// }()
// }
// Close the output channel when all workers are done
// go func() {
// wg.Wait()
// close(jobDescriptionChan)
// }()
}