Merge pull request #190 from MichaelMure/bridge-ctx

Bridge: pull/push enhancements

4 files changed, 173 insertions, 63 deletions

diff --git a/bridge/gitlab/import.go b/bridge/gitlab/import.go
index e135b8bc..1391bf88 100644
--- a/bridge/gitlab/import.go
+++ b/bridge/gitlab/import.go
@@ -1,6 +1,7 @@
 package gitlab
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"time"
@@ -21,11 +22,8 @@ type gitlabImporter struct {
 	// iterator
 	iterator *iterator
 
-	// number of imported issues
-	importedIssues int
-
-	// number of imported identities
-	importedIdentities int
+	// send only channel
+	out chan<- core.ImportResult
 }
 
 func (gi *gitlabImporter) Init(conf core.Configuration) error {
@@ -35,49 +33,60 @@ func (gi *gitlabImporter) Init(conf core.Configuration) error {
 
 // ImportAll iterate over all the configured repository issues (notes) and ensure the creation
 // of the missing issues / comments / label events / title changes ...
-func (gi *gitlabImporter) ImportAll(repo *cache.RepoCache, since time.Time) error {
-	gi.iterator = NewIterator(gi.conf[keyProjectID], gi.conf[keyToken], since)
-
-	// Loop over all matching issues
-	for gi.iterator.NextIssue() {
-		issue := gi.iterator.IssueValue()
-		fmt.Printf("importing issue: %v\n", issue.Title)
+func (gi *gitlabImporter) ImportAll(ctx context.Context, repo *cache.RepoCache, since time.Time) (<-chan core.ImportResult, error) {
+	gi.iterator = NewIterator(ctx, 10, gi.conf[keyProjectID], gi.conf[keyToken], since)
+	out := make(chan core.ImportResult)
+	gi.out = out
+
+	go func() {
+		defer close(gi.out)
+
+		// Loop over all matching issues
+		for gi.iterator.NextIssue() {
+			issue := gi.iterator.IssueValue()
+
+			// create issue
+			b, err := gi.ensureIssue(repo, issue)
+			if err != nil {
+				err := fmt.Errorf("issue creation: %v", err)
+				out <- core.NewImportError(err, "")
+				return
+			}
 
-		// create issue
-		b, err := gi.ensureIssue(repo, issue)
-		if err != nil {
-			return fmt.Errorf("issue creation: %v", err)
-		}
+			// Loop over all notes
+			for gi.iterator.NextNote() {
+				note := gi.iterator.NoteValue()
+				if err := gi.ensureNote(repo, b, note); err != nil {
+					err := fmt.Errorf("note creation: %v", err)
+					out <- core.NewImportError(err, entity.Id(strconv.Itoa(note.ID)))
+					return
+				}
+			}
 
-		// Loop over all notes
-		for gi.iterator.NextNote() {
-			note := gi.iterator.NoteValue()
-			if err := gi.ensureNote(repo, b, note); err != nil {
-				return fmt.Errorf("note creation: %v", err)
+			// Loop over all label events
+			for gi.iterator.NextLabelEvent() {
+				labelEvent := gi.iterator.LabelEventValue()
+				if err := gi.ensureLabelEvent(repo, b, labelEvent); err != nil {
+					err := fmt.Errorf("label event creation: %v", err)
+					out <- core.NewImportError(err, entity.Id(strconv.Itoa(labelEvent.ID)))
+					return
+				}
 			}
-		}
 
-		// Loop over all label events
-		for gi.iterator.NextLabelEvent() {
-			labelEvent := gi.iterator.LabelEventValue()
-			if err := gi.ensureLabelEvent(repo, b, labelEvent); err != nil {
-				return fmt.Errorf("label event creation: %v", err)
+			// commit bug state
+			if err := b.CommitAsNeeded(); err != nil {
+				err := fmt.Errorf("bug commit: %v", err)
+				out <- core.NewImportError(err, "")
+				return
 			}
 		}
 
 		if err := gi.iterator.Error(); err != nil {
-			fmt.Printf("import error: %v\n", err)
-			return err
+			out <- core.NewImportError(err, "")
 		}
+	}()
 
-		// commit bug state
-		if err := b.CommitAsNeeded(); err != nil {
-			return fmt.Errorf("bug commit: %v", err)
-		}
-	}
-
-	fmt.Printf("Successfully imported %d issues and %d identities from Gitlab\n", gi.importedIssues, gi.importedIdentities)
-	return nil
+	return out, nil
 }
 
 func (gi *gitlabImporter) ensureIssue(repo *cache.RepoCache, issue *gitlab.Issue) (*cache.BugCache, error) {
@@ -89,13 +98,13 @@ func (gi *gitlabImporter) ensureIssue(repo *cache.RepoCache, issue *gitlab.Issue
 
 	// resolve bug
 	b, err := repo.ResolveBugCreateMetadata(keyGitlabUrl, issue.WebURL)
-	if err != nil && err != bug.ErrBugNotExist {
-		return nil, err
-	}
-
 	if err == nil {
+		gi.out <- core.NewImportNothing("", "bug already imported")
 		return b, nil
 	}
+	if err != bug.ErrBugNotExist {
+		return nil, err
+	}
 
 	// if bug was never imported
 	cleanText, err := text.Cleanup(issue.Description)
@@ -123,7 +132,7 @@ func (gi *gitlabImporter) ensureIssue(repo *cache.RepoCache, issue *gitlab.Issue
 	}
 
 	// importing a new bug
-	gi.importedIssues++
+	gi.out <- core.NewImportBug(b.Id())
 
 	return b, nil
 }
@@ -149,28 +158,36 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 			return nil
 		}
 
-		_, err = b.CloseRaw(
+		op, err := b.CloseRaw(
 			author,
 			note.CreatedAt.Unix(),
 			map[string]string{
 				keyGitlabId: gitlabID,
 			},
 		)
-		return err
+		if err != nil {
+			return err
+		}
+
+		gi.out <- core.NewImportStatusChange(op.Id())
 
 	case NOTE_REOPENED:
 		if errResolve == nil {
 			return nil
 		}
 
-		_, err = b.OpenRaw(
+		op, err := b.OpenRaw(
 			author,
 			note.CreatedAt.Unix(),
 			map[string]string{
 				keyGitlabId: gitlabID,
 			},
 		)
-		return err
+		if err != nil {
+			return err
+		}
+
+		gi.out <- core.NewImportStatusChange(op.Id())
 
 	case NOTE_DESCRIPTION_CHANGED:
 		issue := gi.iterator.IssueValue()
@@ -181,7 +198,7 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 		// TODO: Check only one time and ignore next 'description change' within one issue
 		if errResolve == cache.ErrNoMatchingOp && issue.Description != firstComment.Message {
 			// comment edition
-			_, err = b.EditCommentRaw(
+			op, err := b.EditCommentRaw(
 				author,
 				note.UpdatedAt.Unix(),
 				firstComment.Id(),
@@ -190,8 +207,11 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 					keyGitlabId: gitlabID,
 				},
 			)
+			if err != nil {
+				return err
+			}
 
-			return err
+			gi.out <- core.NewImportTitleEdition(op.Id())
 		}
 
 	case NOTE_COMMENT:
@@ -204,7 +224,7 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 		if errResolve == cache.ErrNoMatchingOp {
 
 			// add comment operation
-			_, err = b.AddCommentRaw(
+			op, err := b.AddCommentRaw(
 				author,
 				note.CreatedAt.Unix(),
 				cleanText,
@@ -213,8 +233,11 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 					keyGitlabId: gitlabID,
 				},
 			)
-
-			return err
+			if err != nil {
+				return err
+			}
+			gi.out <- core.NewImportComment(op.Id())
+			return nil
 		}
 
 		// if comment was already exported
@@ -228,7 +251,7 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 		// compare local bug comment with the new note body
 		if comment.Message != cleanText {
 			// comment edition
-			_, err = b.EditCommentRaw(
+			op, err := b.EditCommentRaw(
 				author,
 				note.UpdatedAt.Unix(),
 				comment.Id(),
@@ -236,7 +259,10 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 				nil,
 			)
 
-			return err
+			if err != nil {
+				return err
+			}
+			gi.out <- core.NewImportCommentEdition(op.Id())
 		}
 
 		return nil
@@ -247,7 +273,7 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 			return nil
 		}
 
-		_, err = b.SetTitleRaw(
+		op, err := b.SetTitleRaw(
 			author,
 			note.CreatedAt.Unix(),
 			body,
@@ -255,8 +281,11 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 				keyGitlabId: gitlabID,
 			},
 		)
+		if err != nil {
+			return err
+		}
 
-		return err
+		gi.out <- core.NewImportTitleEdition(op.Id())
 
 	case NOTE_UNKNOWN,
 		NOTE_ASSIGNED,
@@ -269,6 +298,9 @@ func (gi *gitlabImporter) ensureNote(repo *cache.RepoCache, b *cache.BugCache, n
 		NOTE_UNLOCKED,
 		NOTE_MENTIONED_IN_ISSUE,
 		NOTE_MENTIONED_IN_MERGE_REQUEST:
+
+		reason := fmt.Sprintf("unsupported note type: %s", noteType.String())
+		gi.out <- core.NewImportNothing("", reason)
 		return nil
 
 	default:
@@ -337,10 +369,7 @@ func (gi *gitlabImporter) ensurePerson(repo *cache.RepoCache, id int) (*cache.Id
 		return nil, err
 	}
 
-	// importing a new identity
-	gi.importedIdentities++
-
-	return repo.NewIdentityRaw(
+	i, err = repo.NewIdentityRaw(
 		user.Name,
 		user.PublicEmail,
 		user.Username,
@@ -351,6 +380,12 @@ func (gi *gitlabImporter) ensurePerson(repo *cache.RepoCache, id int) (*cache.Id
 			keyGitlabLogin: user.Username,
 		},
 	)
+	if err != nil {
+		return nil, err
+	}
+
+	gi.out <- core.NewImportIdentity(i.Id())
+	return i, nil
 }
 
 func parseID(id int) string {
diff --git a/bridge/gitlab/import_notes.go b/bridge/gitlab/import_notes.go
index c0796037..b38cb371 100644
--- a/bridge/gitlab/import_notes.go
+++ b/bridge/gitlab/import_notes.go
@@ -28,6 +28,45 @@ const (
 	NOTE_UNKNOWN
 )
 
+func (nt NoteType) String() string {
+	switch nt {
+	case NOTE_COMMENT:
+		return "note comment"
+	case NOTE_TITLE_CHANGED:
+		return "note title changed"
+	case NOTE_DESCRIPTION_CHANGED:
+		return "note description changed"
+	case NOTE_CLOSED:
+		return "note closed"
+	case NOTE_REOPENED:
+		return "note reopened"
+	case NOTE_LOCKED:
+		return "note locked"
+	case NOTE_UNLOCKED:
+		return "note unlocked"
+	case NOTE_CHANGED_DUEDATE:
+		return "note changed duedate"
+	case NOTE_REMOVED_DUEDATE:
+		return "note remove duedate"
+	case NOTE_ASSIGNED:
+		return "note assigned"
+	case NOTE_UNASSIGNED:
+		return "note unassigned"
+	case NOTE_CHANGED_MILESTONE:
+		return "note changed milestone"
+	case NOTE_REMOVED_MILESTONE:
+		return "note removed in milestone"
+	case NOTE_MENTIONED_IN_ISSUE:
+		return "note mentioned in issue"
+	case NOTE_MENTIONED_IN_MERGE_REQUEST:
+		return "note mentioned in merge request"
+	case NOTE_UNKNOWN:
+		return "note unknown"
+	default:
+		panic("unknown note type")
+	}
+}
+
 // GetNoteType parse a note system and body and return the note type and it content
 func GetNoteType(n *gitlab.Note) (NoteType, string) {
 	// when a note is a comment system is set to false
diff --git a/bridge/gitlab/import_test.go b/bridge/gitlab/import_test.go
index c38d3ce3..20fc67c7 100644
--- a/bridge/gitlab/import_test.go
+++ b/bridge/gitlab/import_test.go
@@ -1,6 +1,7 @@
 package gitlab
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"testing"
@@ -99,10 +100,16 @@ func TestImport(t *testing.T) {
 	})
 	require.NoError(t, err)
 
+	ctx := context.Background()
 	start := time.Now()
-	err = importer.ImportAll(backend, time.Time{})
+
+	events, err := importer.ImportAll(ctx, backend, time.Time{})
 	require.NoError(t, err)
 
+	for result := range events {
+		require.NoError(t, result.Err)
+	}
+
 	fmt.Printf("test repository imported in %f seconds\n", time.Since(start).Seconds())
 
 	require.Len(t, backend.AllBugsIds(), len(tests))
diff --git a/bridge/gitlab/iterator.go b/bridge/gitlab/iterator.go
index 883fea9c..198af79b 100644
--- a/bridge/gitlab/iterator.go
+++ b/bridge/gitlab/iterator.go
@@ -1,6 +1,7 @@
 package gitlab
 
 import (
+	"context"
 	"time"
 
 	"github.com/xanzy/go-gitlab"
@@ -38,6 +39,9 @@ type iterator struct {
 	// number of issues and notes to query at once
 	capacity int
 
+	// shared context
+	ctx context.Context
+
 	// sticky error
 	err error
 
@@ -52,12 +56,13 @@ type iterator struct {
 }
 
 // NewIterator create a new iterator
-func NewIterator(projectID, token string, since time.Time) *iterator {
+func NewIterator(ctx context.Context, capacity int, projectID, token string, since time.Time) *iterator {
 	return &iterator{
 		gc:       buildClient(token),
 		project:  projectID,
 		since:    since,
-		capacity: 20,
+		capacity: capacity,
+		ctx:      ctx,
 		issue: &issueIterator{
 			index: -1,
 			page:  1,
@@ -79,6 +84,9 @@ func (i *iterator) Error() error {
 }
 
 func (i *iterator) getNextIssues() bool {
+	ctx, cancel := context.WithTimeout(i.ctx, defaultTimeout)
+	defer cancel()
+
 	issues, _, err := i.gc.Issues.ListProjectIssues(
 		i.project,
 		&gitlab.ListProjectIssuesOptions{
@@ -90,6 +98,7 @@ func (i *iterator) getNextIssues() bool {
 			UpdatedAfter: &i.since,
 			Sort:         gitlab.String("asc"),
 		},
+		gitlab.WithContext(ctx),
 	)
 
 	if err != nil {
@@ -116,6 +125,10 @@ func (i *iterator) NextIssue() bool {
 		return false
 	}
 
+	if i.ctx.Err() != nil {
+		return false
+	}
+
 	// first query
 	if i.issue.cache == nil {
 		return i.getNextIssues()
@@ -135,6 +148,9 @@ func (i *iterator) IssueValue() *gitlab.Issue {
 }
 
 func (i *iterator) getNextNotes() bool {
+	ctx, cancel := context.WithTimeout(i.ctx, defaultTimeout)
+	defer cancel()
+
 	notes, _, err := i.gc.Notes.ListIssueNotes(
 		i.project,
 		i.IssueValue().IID,
@@ -146,6 +162,7 @@ func (i *iterator) getNextNotes() bool {
 			Sort:    gitlab.String("asc"),
 			OrderBy: gitlab.String("created_at"),
 		},
+		gitlab.WithContext(ctx),
 	)
 
 	if err != nil {
@@ -171,6 +188,10 @@ func (i *iterator) NextNote() bool {
 		return false
 	}
 
+	if i.ctx.Err() != nil {
+		return false
+	}
+
 	if len(i.note.cache) == 0 {
 		return i.getNextNotes()
 	}
@@ -189,6 +210,9 @@ func (i *iterator) NoteValue() *gitlab.Note {
 }
 
 func (i *iterator) getNextLabelEvents() bool {
+	ctx, cancel := context.WithTimeout(i.ctx, defaultTimeout)
+	defer cancel()
+
 	labelEvents, _, err := i.gc.ResourceLabelEvents.ListIssueLabelEvents(
 		i.project,
 		i.IssueValue().IID,
@@ -198,6 +222,7 @@ func (i *iterator) getNextLabelEvents() bool {
 				PerPage: i.capacity,
 			},
 		},
+		gitlab.WithContext(ctx),
 	)
 
 	if err != nil {
@@ -224,6 +249,10 @@ func (i *iterator) NextLabelEvent() bool {
 		return false
 	}
 
+	if i.ctx.Err() != nil {
+		return false
+	}
+
 	if len(i.labelEvent.cache) == 0 {
 		return i.getNextLabelEvents()
 	}