Skip to content

Commit

Permalink
Process issues and worklogs concurrently
Browse files Browse the repository at this point in the history
  • Loading branch information
@berlam
berlam committed Oct 4, 2019
1 parent bf8b3f9 commit ea41562
Show file tree
Hide file tree
Showing 5 changed files with 124 additions and 108 deletions.
8 changes: 4 additions & 4 deletions pkg/jira/cloud/api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,10 @@ func (api Api) User(user *pkg.User) (model.Account, *time.Location, error) {
return api.previousVersion().User(user)
}

func (api Api) Issues(jql model.Jql, startAt int) ([]model.Issue, error) {
return api.previousVersion().Issues(jql, startAt)
func (api Api) Issues(jql model.Jql, issueFunc model.IssueFunc) error {
return api.previousVersion().Issues(jql, issueFunc)
}

func (api Api) Worklog(key model.IssueKey, startAt int) ([]model.Worklog, error) {
return api.previousVersion().Worklog(key, startAt)
func (api Api) Worklog(key model.IssueKey, worklogFunc model.WorklogFunc) error {
return api.previousVersion().Worklog(key, worklogFunc)
}
30 changes: 12 additions & 18 deletions pkg/jira/cloud/api_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ package cloud

import (
"eager/pkg"
"eager/pkg/jira/model"
"net/url"
"os"
"testing"
Expand All @@ -22,29 +21,24 @@ func Test(t *testing.T) {
Server: path,
Userinfo: url.UserPassword(cloudUser, cloudToken),
}
projects, e := api.Projects(0)
if e != nil || len(projects) == 0 || projects[0] != "TEST" {
t.Error("Project not found")
return
}
user := &pkg.User{
DisplayName: "Berla Atlassian Test User 2",
TimeZone: time.UTC,
}
account, _, e := api.User(user, projects)
account, _, e := api.User(user)
if e != nil || account == "" {
t.Error("User not found")
return
}
jql := model.Jql{}.Users(account).Projects(projects...)
issues, e := api.Issues(jql, 0)
if e != nil || len(issues) == 0 {
t.Error("Issues not found")
return
}
worklog, e := api.Worklog(issues[0].Key(), 0)
if e != nil || len(worklog) == 0 {
t.Error("Worklog not found")
return
}
//jql := model.Jql{}.Users(account)
//e := api.Issues(jql, 0)
//if e != nil || len(issues) == 0 {
// t.Error("Issues not found")
// return
//}
//e := api.Worklog(issues[0].Key(), 0)
//if e != nil || len(worklog) == 0 {
// t.Error("Worklog not found")
// return
//}
}
93 changes: 62 additions & 31 deletions pkg/jira/jira.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -115,44 +115,75 @@ func do(api model.Api, year int, month time.Month, projects []pkg.Project, accou
}

jql := new(model.Jql).Between(fromDate, toDate).Users(accountIds...).Projects(projects...)
issues, err := api.Issues(jql, 0)

if err != nil {
log.Println("Could not get issues.", err)
return pkg.Timesheet{}
}
// The chan for errors
errors := make(chan error)

burstLimit := 5
// If we do not throttle here, the server will sometimes respond with 401.
// That looks like a rate limit but I have not found any numbers in the docs for the cloud setup.
throttle := make(chan struct{}, burstLimit)
c := make(chan pkg.Timesheet)
var wg sync.WaitGroup
wg.Add(len(issues))
for _, item := range issues {
go func(item model.Issue) {
defer wg.Done()
throttle <- struct{}{}
items, err := api.Worklog(item.Key(), 0)
<-throttle
if err != nil {
log.Println("Could not get effort for "+item.Key(), err)
return
}
worklog := item.Worklog(accounts, items, fromDate, toDate)
for account := range accounts {
c <- worklog[account]
}
}(item)
}
// The chan for issues
issues := make(chan model.Issue)
go func() {
defer close(c)
defer close(issues)
err = api.Issues(jql, func(issue model.Issue) {
issues <- issue
})
if err != nil {
log.Println("Could not get issues.", err)
errors <- err
}
}()

// The chan for effort
effort := make(chan pkg.Effort)
go func() {
var wg sync.WaitGroup
throttle := make(chan struct{}, 5)
defer close(effort)
defer close(throttle)
for issue := range issues {
wg.Add(1)
throttle <- struct{}{}
go func(issue model.Issue) {
defer func() {
<-throttle
wg.Done()
}()
err = api.Worklog(issue.Key(), func(worklog model.Worklog) {
account := worklog.Author().Id()
user := accounts[account]
if user == nil {
return
}
date := worklog.Date().In(user.TimeZone)
date = time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, time.UTC)
if !date.Before(fromDate) && date.Before(toDate) {
effort <- pkg.Effort{
User: user,
Description: worklog.Comment(),
Project: pkg.Project(issue.Project()),
Task: pkg.Task(issue.Key()),
Date: date,
Duration: worklog.Duration(),
}
}
})
if err != nil {
log.Println("Could not get effort for "+issue.Key(), err)
errors <- err
}
}(issue)
}
wg.Wait()
}()

var timesheet pkg.Timesheet
for effort := range c {
timesheet = append(timesheet, effort...)
go func() {
defer close(errors)
for e := range effort {
timesheet = append(timesheet, e)
}
}()
if <-errors != nil {
return pkg.Timesheet{}
}

return timesheet
Expand Down
18 changes: 15 additions & 3 deletions pkg/jira/model/api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,16 +21,28 @@ type UserAccessor interface {
}

type IssueAccessor interface {
Issues(jql Jql, startAt int) ([]Issue, error)
Issues(jql Jql, issueFunc IssueFunc) error
}

type IssueFunc func(Issue)

func (f IssueFunc) Process(issue Issue) {
f(issue)
}

type WorklogAccessor interface {
Worklog(key IssueKey, startAt int) ([]Worklog, error)
Worklog(key IssueKey, worklogFunc WorklogFunc) error
}

type WorklogFunc func(Worklog)

func (f WorklogFunc) Process(worklog Worklog) {
f(worklog)
}

type Issue interface {
Project() pkg.Project
Key() IssueKey
Worklog(accounts map[Account]*pkg.User, worklog []Worklog, fromDate, toDate time.Time) map[Account]pkg.Timesheet
}

type Worklog interface {
Expand Down
83 changes: 31 additions & 52 deletions pkg/jira/v2/api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,11 @@ func (api Api) User(user *pkg.User) (model.Account, *time.Location, error) {
return result[0].AccountId, result[0].Location(), nil
}

func (api Api) Issues(jql model.Jql, startAt int) ([]model.Issue, error) {
func (api Api) Issues(jql model.Jql, issueFunc model.IssueFunc) error {
return api.issues(jql, 0, issueFunc)
}

func (api Api) issues(jql model.Jql, startAt int, issueFunc model.IssueFunc) error {
body, _ := json.Marshal(issueQuery{
Fields: []string{"project"},
Jql: jql.Build(),
Expand All @@ -123,7 +127,7 @@ func (api Api) Issues(jql model.Jql, startAt int) ([]model.Issue, error) {
searchUrl, _ := api.Server.Parse(searchIssueUrl)
response, err := pkg.CreateJsonRequest(api.Client, http.MethodPost, searchUrl, api.Userinfo, bytes.NewBuffer(body))
if err != nil {
return nil, err
return err
}
defer func() {
err := response.Body.Close()
Expand All @@ -135,33 +139,36 @@ func (api Api) Issues(jql model.Jql, startAt int) ([]model.Issue, error) {
reader, _ := charset.NewReader(response.Body, response.Header.Get("Content-Type"))
data, err := ioutil.ReadAll(reader)
if err != nil {
return nil, err
return err
}
if response.StatusCode != 200 {
return nil, fmt.Errorf(response.Status)
return fmt.Errorf(response.Status)
}

var result = issueQueryResult{}
err = json.Unmarshal(data, &result)
if err != nil {
return nil, err
return err
}
issues := result.issues()
for _, e := range issues {
issueFunc(e)
}
if (result.IsLast == nil && result.Total >= startAt+result.MaxResults) || (result.IsLast != nil && !*result.IsLast) {
nextIssues, err := api.Issues(jql, startAt+result.MaxResults)
if err != nil {
return nil, err
}
issues = append(issues, nextIssues...)
err = api.issues(jql, startAt+result.MaxResults, issueFunc)
}
return issues, err
return err
}

func (api Api) Worklog(key model.IssueKey, startAt int) ([]model.Worklog, error) {
func (api Api) Worklog(key model.IssueKey, worklogFunc model.WorklogFunc) error {
return api.worklog(key, 0, worklogFunc)
}

func (api Api) worklog(key model.IssueKey, startAt int, worklogFunc model.WorklogFunc) error {
worklogUrl, err := api.Server.Parse(fmt.Sprintf(worklogUrl, string(key), strconv.Itoa(startAt)))
response, err := pkg.CreateJsonRequest(api.Client, http.MethodGet, worklogUrl, api.Userinfo, nil)
if err != nil {
return nil, err
return err
}
defer func() {
err := response.Body.Close()
Expand All @@ -173,23 +180,22 @@ func (api Api) Worklog(key model.IssueKey, startAt int) ([]model.Worklog, error)
reader, _ := charset.NewReader(response.Body, response.Header.Get("Content-Type"))
data, err := ioutil.ReadAll(reader)
if err != nil {
return nil, err
return err
}
if response.StatusCode != 200 {
return nil, fmt.Errorf(response.Status)
return fmt.Errorf(response.Status)
}

var result = worklogQueryResult{}
err = json.Unmarshal(data, &result)
items := result.worklogs()
for _, e := range items {
worklogFunc(e)
}
if (result.IsLast == nil && result.Total >= startAt+result.MaxResults) || (result.IsLast != nil && !*result.IsLast) {
nextItems, err := api.Worklog(key, startAt+result.MaxResults)
if err != nil {
return nil, err
}
items = append(items, nextItems...)
err = api.worklog(key, startAt+result.MaxResults, worklogFunc)
}
return items, nil
return err
}

func (result issueQueryResult) issues() []model.Issue {
Expand All @@ -208,39 +214,12 @@ func (result worklogQueryResult) worklogs() []model.Worklog {
return worklogs
}

func (issue issue) Key() model.IssueKey {
return issue.ApiKey
func (issue issue) Project() pkg.Project {
return pkg.Project(issue.Fields.Project.Key)
}

func (issue issue) Worklog(accounts map[model.Account]*pkg.User, worklog []model.Worklog, fromDate, toDate time.Time) map[model.Account]pkg.Timesheet {
pKey := issue.Fields.Project.Key
iKey := issue.ApiKey
result := make(map[model.Account]pkg.Timesheet)
total := len(worklog)
for _, effort := range worklog {
account := effort.Author().Id()
user := accounts[account]
if user == nil {
continue
}
date := effort.Date().In(user.TimeZone)
date = time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, time.UTC)
if !date.Before(fromDate) && date.Before(toDate) {
current := result[account]
if current == nil {
current = make(pkg.Timesheet, 0, total)
}
result[account] = append(current, pkg.Effort{
User: user,
Description: effort.Comment(),
Project: pkg.Project(pKey),
Task: pkg.Task(iKey),
Date: date,
Duration: effort.Duration(),
})
}
}
return result
func (issue issue) Key() model.IssueKey {
return issue.ApiKey
}

func (author author) Id() model.Account {
Expand Down

0 comments on commit ea41562

Please sign in to comment.