Implement formal fork support for leaves

internal/plumbing/blob_cache.go

@@ -154,6 +154,7 @@ func (blobCache *BlobCache) Consume(deps map[string]interface{}) (map[string]int
 	return map[string]interface{}{DependencyBlobCache: cache}, nil
 }
 
+// Fork clones this PipelineItem.
 func (blobCache *BlobCache) Fork(n int) []core.PipelineItem {
 	caches := make([]core.PipelineItem, n)
 	for i := 0; i < n; i++ {

internal/plumbing/day.go

@@ -114,6 +114,7 @@ func (days *DaysSinceStart) Consume(deps map[string]interface{}) (map[string]int
 	return map[string]interface{}{DependencyDay: day}, nil
 }
 
+// Fork clones this PipelineItem.
 func (days *DaysSinceStart) Fork(n int) []core.PipelineItem {
 	return core.ForkCopyPipelineItem(days, n)
 }

internal/plumbing/diff.go

@@ -129,6 +129,7 @@ func (diff *FileDiff) Consume(deps map[string]interface{}) (map[string]interface
 	return map[string]interface{}{DependencyFileDiff: result}, nil
 }
 
+// Fork clones this PipelineItem.
 func (diff *FileDiff) Fork(n int) []core.PipelineItem {
 	return core.ForkSamePipelineItem(diff, n)
 }

internal/plumbing/identity/identity.go

@@ -131,6 +131,7 @@ func (detector *Detector) Consume(deps map[string]interface{}) (map[string]inter
 	return map[string]interface{}{DependencyAuthor: authorID}, nil
 }
 
+// Fork clones this PipelineItem.
 func (detector *Detector) Fork(n int) []core.PipelineItem {
 	return core.ForkSamePipelineItem(detector, n)
 }

internal/plumbing/renames.go

@@ -204,6 +204,7 @@ func (ra *RenameAnalysis) Consume(deps map[string]interface{}) (map[string]inter
 	return map[string]interface{}{DependencyTreeChanges: reducedChanges}, nil
 }
 
+// Fork clones this PipelineItem.
 func (ra *RenameAnalysis) Fork(n int) []core.PipelineItem {
 	return core.ForkSamePipelineItem(ra, n)
 }

internal/plumbing/tree_diff.go

@@ -142,6 +142,7 @@ func (treediff *TreeDiff) Consume(deps map[string]interface{}) (map[string]inter
 	return map[string]interface{}{DependencyTreeChanges: diff}, nil
 }
 
+// Fork clones this PipelineItem.
 func (treediff *TreeDiff) Fork(n int) []core.PipelineItem {
 	return core.ForkCopyPipelineItem(treediff, n)
 }

internal/plumbing/uast/diff_refiner.go

@@ -15,6 +15,7 @@ import (
 // The idea behind this algorithm is simple: in case of multiple choices which are equally
 // optimal, choose the one which touches less AST nodes.
 type FileDiffRefiner struct {
+	core.NoopMerger
 }
 
 // Name of this PipelineItem. Uniquely identifies the type, used for mapping keys, etc.
@@ -161,16 +162,9 @@ func (ref *FileDiffRefiner) Consume(deps map[string]interface{}) (map[string]int
 	return map[string]interface{}{plumbing.DependencyFileDiff: result}, nil
 }
 
+// Fork clones this PipelineItem.
 func (ref *FileDiffRefiner) Fork(n int) []core.PipelineItem {
-	refs := make([]core.PipelineItem, n)
-	for i := 0; i < n; i++ {
-		refs[i] = ref
-	}
-	return refs
-}
-
-func (ref *FileDiffRefiner) Merge(branches []core.PipelineItem) {
-	// no-op
+	return core.ForkSamePipelineItem(ref, n)
 }
 
 // VisitEachNode is a handy routine to execute a callback on every node in the subtree,

internal/plumbing/uast/uast.go

@@ -288,6 +288,7 @@ func (exr *Extractor) Consume(deps map[string]interface{}) (map[string]interface
 	return map[string]interface{}{DependencyUasts: uasts}, nil
 }
 
+// Fork clones this PipelineItem.
 func (exr *Extractor) Fork(n int) []core.PipelineItem {
 	return core.ForkSamePipelineItem(exr, n)
 }
@@ -433,6 +434,7 @@ func (uc *Changes) Consume(deps map[string]interface{}) (map[string]interface{},
 	return map[string]interface{}{DependencyUastChanges: commit}, nil
 }
 
+// Fork clones this PipelineItem.
 func (uc *Changes) Fork(n int) []core.PipelineItem {
 	ucs := make([]core.PipelineItem, n)
 	for i := 0; i < n; i++ {
@@ -542,6 +544,7 @@ func (saver *ChangesSaver) Finalize() interface{} {
 	return saver.result
 }
 
+// Fork clones this PipelineItem.
 func (saver *ChangesSaver) Fork(n int) []core.PipelineItem {
 	return core.ForkSamePipelineItem(saver, n)
 }

leaves/burndown_test.go

@@ -178,6 +178,8 @@ func TestBurndownConsumeFinalize(t *testing.T) {
 	result, err := fd.Consume(deps)
 	assert.Nil(t, err)
 	deps[items.DependencyFileDiff] = result[items.DependencyFileDiff]
+	deps[core.DependencyCommit], _ = test.Repository.CommitObject(plumbing.NewHash(
+		"cce947b98a050c6d356bc6ba95030254914027b1"))
 	result, err = burndown.Consume(deps)
 	assert.Nil(t, result)
 	assert.Nil(t, err)

leaves/comment_sentiment.go

@@ -25,6 +25,8 @@ import (
 
 // CommentSentimentAnalysis measures comment sentiment through time.
 type CommentSentimentAnalysis struct {
+	core.NoopMerger
+	core.OneShotMergeProcessor
 	MinCommentLength int
 	Gap              float32
 
@@ -141,6 +143,7 @@ func (sent *CommentSentimentAnalysis) Initialize(repository *git.Repository) {
 	sent.commentsByDay = map[int][]string{}
 	sent.xpather = &uast_items.ChangesXPather{XPath: "//*[@roleComment]"}
 	sent.validate()
+	sent.OneShotMergeProcessor.Initialize()
 }
 
 // Consume runs this PipelineItem on the next commit data.
@@ -149,6 +152,9 @@ func (sent *CommentSentimentAnalysis) Initialize(repository *git.Repository) {
 // This function returns the mapping with analysis results. The keys must be the same as
 // in Provides(). If there was an error, nil is returned.
 func (sent *CommentSentimentAnalysis) Consume(deps map[string]interface{}) (map[string]interface{}, error) {
+	if !sent.ShouldConsumeCommit(deps) {
+		return nil, nil
+	}
 	changes := deps[uast_items.DependencyUastChanges].([]uast_items.Change)
 	day := deps[items.DependencyDay].(int)
 	commentNodes := sent.xpather.Extract(changes)
@@ -225,6 +231,11 @@ func (sent *CommentSentimentAnalysis) Finalize() interface{} {
 	return result
 }
 
+// Fork clones this PipelineItem.
+func (sent *CommentSentimentAnalysis) Fork(n int) []core.PipelineItem {
+	return core.ForkSamePipelineItem(sent, n)
+}
+
 // Serialize converts the analysis result as returned by Finalize() to text or bytes.
 // The text format is YAML and the bytes format is Protocol Buffers.
 func (sent *CommentSentimentAnalysis) Serialize(result interface{}, binary bool, writer io.Writer) error {

leaves/comment_sentiment_test.go

@@ -87,6 +87,15 @@ func TestCommentSentimentRegistration(t *testing.T) {
 	assert.True(t, matched)
 }
 
+func TestCommentSentimentFork(t *testing.T) {
+	sent1 := fixtureCommentSentiment()
+	clones := sent1.Fork(1)
+	assert.Len(t, clones, 1)
+	sent2 := clones[0].(*CommentSentimentAnalysis)
+	assert.True(t, sent1 == sent2)
+	sent1.Merge([]core.PipelineItem{sent2})
+}
+
 func TestCommentSentimentSerializeText(t *testing.T) {
 	sent := fixtureCommentSentiment()
 	result := CommentSentimentResult{

leaves/couples.go

@@ -20,6 +20,8 @@ import (
 // The results are matrices, where cell at row X and column Y is the number of commits which
 // changed X and Y together. In case with people, the numbers are summed for every common file.
 type CouplesAnalysis struct {
+	core.NoopMerger
+	core.OneShotMergeProcessor
 	// PeopleNumber is the number of developers for which to build the matrix. 0 disables this analysis.
 	PeopleNumber int
 
@@ -92,6 +94,7 @@ func (couples *CouplesAnalysis) Initialize(repository *git.Repository) {
 	}
 	couples.peopleCommits = make([]int, couples.PeopleNumber+1)
 	couples.files = map[string]map[string]int{}
+	couples.OneShotMergeProcessor.Initialize()
 }
 
 // Consume runs this PipelineItem on the next commit data.
@@ -100,6 +103,9 @@ func (couples *CouplesAnalysis) Initialize(repository *git.Repository) {
 // This function returns the mapping with analysis results. The keys must be the same as
 // in Provides(). If there was an error, nil is returned.
 func (couples *CouplesAnalysis) Consume(deps map[string]interface{}) (map[string]interface{}, error) {
+	if !couples.ShouldConsumeCommit(deps) {
+		return nil, nil
+	}
 	author := deps[identity.DependencyAuthor].(int)
 	if author == identity.AuthorMissing {
 		author = couples.PeopleNumber
@@ -217,6 +223,11 @@ func (couples *CouplesAnalysis) Finalize() interface{} {
 	}
 }
 
+// Fork clones this pipeline item.
+func (couples *CouplesAnalysis) Fork(n int) []core.PipelineItem {
+	return core.ForkSamePipelineItem(couples, n)
+}
+
 // Serialize converts the analysis result as returned by Finalize() to text or bytes.
 // The text format is YAML and the bytes format is Protocol Buffers.
 func (couples *CouplesAnalysis) Serialize(result interface{}, binary bool, writer io.Writer) error {

leaves/couples_test.go

@@ -168,6 +168,15 @@ func TestCouplesConsumeFinalize(t *testing.T) {
 	assert.Equal(t, cr.FilesMatrix[2][2], int64(2))
 }
 
+func TestCouplesFork(t *testing.T) {
+	couples1 := fixtureCouples()
+	clones := couples1.Fork(1)
+	assert.Len(t, clones, 1)
+	couples2 := clones[0].(*CouplesAnalysis)
+	assert.True(t, couples1 == couples2)
+	couples1.Merge([]core.PipelineItem{couples2})
+}
+
 func TestCouplesSerialize(t *testing.T) {
 	c := fixtureCouples()
 	c.PeopleNumber = 1

leaves/file_history.go

@@ -19,6 +19,8 @@ import (
 // FileHistory contains the intermediate state which is mutated by Consume(). It should implement
 // LeafPipelineItem.
 type FileHistory struct {
+	core.NoopMerger
+	core.OneShotMergeProcessor
 	files map[string][]plumbing.Hash
 }
 
@@ -65,6 +67,7 @@ func (history *FileHistory) Configure(facts map[string]interface{}) {
 // calls. The repository which is going to be analysed is supplied as an argument.
 func (history *FileHistory) Initialize(repository *git.Repository) {
 	history.files = map[string][]plumbing.Hash{}
+	history.OneShotMergeProcessor.Initialize()
 }
 
 // Consume runs this PipelineItem on the next commit data.
@@ -73,7 +76,10 @@ func (history *FileHistory) Initialize(repository *git.Repository) {
 // This function returns the mapping with analysis results. The keys must be the same as
 // in Provides(). If there was an error, nil is returned.
 func (history *FileHistory) Consume(deps map[string]interface{}) (map[string]interface{}, error) {
-	commit := deps[DependencyCommit].(*object.Commit).Hash
+	if !history.ShouldConsumeCommit(deps) {
+		return nil, nil
+	}
+	commit := deps[core.DependencyCommit].(*object.Commit).Hash
 	changes := deps[items.DependencyTreeChanges].(object.Changes)
 	for _, change := range changes {
 		action, _ := change.Action()
@@ -101,6 +107,11 @@ func (history *FileHistory) Finalize() interface{} {
 	return FileHistoryResult{Files: history.files}
 }
 
+// Fork clones this PipelineItem.
+func (history *FileHistory) Fork(n int) []core.PipelineItem {
+	return core.ForkSamePipelineItem(history, n)
+}
+
 // Serialize converts the analysis result as returned by Finalize() to text or bytes.
 // The text format is YAML and the bytes format is Protocol Buffers.
 func (history *FileHistory) Serialize(result interface{}, binary bool, writer io.Writer) error {

leaves/file_history_test.go

@@ -93,7 +93,7 @@ func TestFileHistoryConsume(t *testing.T) {
 	deps[items.DependencyTreeChanges] = changes
 	commit, _ := test.Repository.CommitObject(plumbing.NewHash(
 		"2b1ed978194a94edeabbca6de7ff3b5771d4d665"))
-	deps[DependencyCommit] = commit
+	deps[core.DependencyCommit] = commit
 	fh.files["cmd/hercules/main.go"] = []plumbing.Hash{plumbing.NewHash(
 		"0000000000000000000000000000000000000000")}
 	fh.files["analyser.go"] = []plumbing.Hash{plumbing.NewHash(
@@ -113,6 +113,15 @@ func TestFileHistoryConsume(t *testing.T) {
 	assert.Equal(t, fh.files, res.Files)
 }
 
+func TestFileHistoryFork(t *testing.T) {
+	fh1 := fixtureFileHistory()
+	clones := fh1.Fork(1)
+	assert.Len(t, clones, 1)
+	fh2 := clones[0].(*FileHistory)
+	assert.True(t, fh1 == fh2)
+	fh1.Merge([]core.PipelineItem{fh2})
+}
+
 func TestFileHistorySerializeText(t *testing.T) {
 	fh := fixtureFileHistory()
 	deps := map[string]interface{}{}
@@ -132,7 +141,7 @@ func TestFileHistorySerializeText(t *testing.T) {
 	deps[items.DependencyTreeChanges] = changes
 	commit, _ := test.Repository.CommitObject(plumbing.NewHash(
 		"2b1ed978194a94edeabbca6de7ff3b5771d4d665"))
-	deps[DependencyCommit] = commit
+	deps[core.DependencyCommit] = commit
 	fh.Consume(deps)
 	res := fh.Finalize().(FileHistoryResult)
 	buffer := &bytes.Buffer{}
@@ -159,7 +168,7 @@ func TestFileHistorySerializeBinary(t *testing.T) {
 	deps[items.DependencyTreeChanges] = changes
 	commit, _ := test.Repository.CommitObject(plumbing.NewHash(
 		"2b1ed978194a94edeabbca6de7ff3b5771d4d665"))
-	deps[DependencyCommit] = commit
+	deps[core.DependencyCommit] = commit
 	fh.Consume(deps)
 	res := fh.Finalize().(FileHistoryResult)
 	buffer := &bytes.Buffer{}

leaves/shotness.go

@@ -22,6 +22,8 @@ import (
 // ShotnessAnalysis contains the intermediate state which is mutated by Consume(). It should implement
 // LeafPipelineItem.
 type ShotnessAnalysis struct {
+	core.NoopMerger
+	core.OneShotMergeProcessor
 	XpathStruct string
 	XpathName   string
 
@@ -138,6 +140,7 @@ func (shotness *ShotnessAnalysis) Configure(facts map[string]interface{}) {
 func (shotness *ShotnessAnalysis) Initialize(repository *git.Repository) {
 	shotness.nodes = map[string]*nodeShotness{}
 	shotness.files = map[string]map[string]*nodeShotness{}
+	shotness.OneShotMergeProcessor.Initialize()
 }
 
 // Consume runs this PipelineItem on the next commit data.
@@ -146,7 +149,10 @@ func (shotness *ShotnessAnalysis) Initialize(repository *git.Repository) {
 // This function returns the mapping with analysis results. The keys must be the same as
 // in Provides(). If there was an error, nil is returned.
 func (shotness *ShotnessAnalysis) Consume(deps map[string]interface{}) (map[string]interface{}, error) {
-	commit := deps[DependencyCommit].(*object.Commit)
+	if !shotness.ShouldConsumeCommit(deps) {
+		return nil, nil
+	}
+	commit := deps[core.DependencyCommit].(*object.Commit)
 	changesList := deps[uast_items.DependencyUastChanges].([]uast_items.Change)
 	diffs := deps[items.DependencyFileDiff].(map[string]items.FileDiffData)
 	allNodes := map[string]bool{}
@@ -322,6 +328,11 @@ func (shotness *ShotnessAnalysis) Consume(deps map[string]interface{}) (map[stri
 	return nil, nil
 }
 
+// Fork clones this PipelineItem.
+func (shotness *ShotnessAnalysis) Fork(n int) []core.PipelineItem {
+	return core.ForkSamePipelineItem(shotness, n)
+}
+
 // Finalize returns the result of the analysis. Further Consume() calls are not expected.
 func (shotness *ShotnessAnalysis) Finalize() interface{} {
 	result := ShotnessResult{

leaves/shotness_test.go

@@ -77,7 +77,7 @@ func bakeShotness(t *testing.T, eraseEndPosition bool) (*ShotnessAnalysis, Shotn
 	dmp := diffmatchpatch.New()
 	src, dst, _ := dmp.DiffLinesToRunes(string(bytes1), string(bytes2))
 	state := map[string]interface{}{}
-	state[DependencyCommit] = &object.Commit{}
+	state[core.DependencyCommit] = &object.Commit{}
 	fileDiffs := map[string]items.FileDiffData{}
 	const fileName = "test.java"
 	fileDiffs[fileName] = items.FileDiffData{
@@ -130,7 +130,7 @@ func TestShotnessConsume(t *testing.T) {
 	dmp := diffmatchpatch.New()
 	src, dst, _ := dmp.DiffLinesToRunes(string(bytes1), string(bytes2))
 	state := map[string]interface{}{}
-	state[DependencyCommit] = &object.Commit{}
+	state[core.DependencyCommit] = &object.Commit{}
 	fileDiffs := map[string]items.FileDiffData{}
 	const fileName = "test.java"
 	const newfileName = "new.java"
@@ -209,6 +209,15 @@ func TestShotnessConsume(t *testing.T) {
 	assert.Len(t, sh.files, 0)
 }
 
+func TestShotnessFork(t *testing.T) {
+	sh1 := fixtureShotness()
+	clones := sh1.Fork(1)
+	assert.Len(t, clones, 1)
+	sh2 := clones[0].(*ShotnessAnalysis)
+	assert.True(t, sh1 == sh2)
+	sh1.Merge([]core.PipelineItem{sh2})
+}
+
 func TestShotnessConsumeNoEnd(t *testing.T) {
 	_, result1 := bakeShotness(t, false)
 	_, result2 := bakeShotness(t, true)