-
Notifications
You must be signed in to change notification settings - Fork 0
/
fetcher.go
80 lines (67 loc) · 2.67 KB
/
fetcher.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
package fetcher
import (
log "github.com/Sirupsen/logrus"
"github.com/andygrunwald/go-jira"
"github.com/spf13/viper"
)
// TicketFetcherJob represent a job to fetch a JIRA ticket
type TicketFetcherJob struct {
Key string
Responses chan *jira.Issue
}
// TicketFetcherWorker fetch a complete JIRA ticket using its key and a decicated JIRA client.
// It will send the result to the issues channel
func TicketFetcherWorker(id int, jiraEndpoint string, user string, password string, inputs <-chan *TicketFetcherJob) {
log.WithFields(log.Fields{
"workerID": id,
}).Info("ticketFetcherWorker starting ...")
// Init JIRA client connected to Amoobi's JIRA instance
jiraClient, err := InitJiraClient(jiraEndpoint, user, password)
if err != nil {
log.Error(err)
return
}
// For each issue key
for ticketJob := range inputs {
contextLogger := log.WithFields(log.Fields{
"workerID": id,
"issueKey": ticketJob.Key,
})
contextLogger.Info("ticketFetcherWorker processing ...")
// Fetch all info about issueKey and send them through the channel
issue, _, err := jiraClient.Issue.Get(ticketJob.Key)
if err != nil {
contextLogger.Errorf("Unable to get issue; cause: '%s'", err)
} else {
// Send fetched issue for further treatment
ticketJob.Responses <- issue
}
}
log.WithFields(log.Fields{
"workerID": id,
}).Info("ticketFetcherWorker work done !")
}
// ScheduleTicket schedules our issues for fetching
func ScheduleTicket(rawIssues []jira.Issue, inputs chan<- *TicketFetcherJob, issues chan *jira.Issue) {
// Schedule our issues for fetching
for _, rawIssue := range rawIssues {
// Send some work to our little workers that we love so much
inputs <- &TicketFetcherJob{Key: rawIssue.Key, Responses: issues}
}
}
// StartWorkers will start numberOfWorkerToStart TicketFetcherWorker
func StartWorkers(numberOfWorkerToStart int, endPoint string, user string, password string, inputs <-chan *TicketFetcherJob) {
// This starts up CYCLOP_NUMBER_OF_WORKERS workers, we do it now to
// take advantage of the time needed by the issue search to init our workers
for w := 1; w <= numberOfWorkerToStart; w++ {
go TicketFetcherWorker(w, endPoint, user, password, inputs)
}
}
// FetchTicketsDetail fetch all informations linked to a ticket like worklogs, timetracking, ...
func FetchTicketsDetail(rawIssues []jira.Issue, jobInputs chan<- *TicketFetcherJob) <-chan *jira.Issue {
// In order to use our pool of workers we need a way to get the result of the work
issues := make(chan *jira.Issue, viper.GetInt("nbWorkers"))
// Schedule ticket for fetching in a go function to avoid blocking due to queue size
go ScheduleTicket(rawIssues, jobInputs, issues)
return issues
}