docs updates and refactoring

Author: giventocode

.travis.yml

@@ -7,10 +7,10 @@ before_install:
 # Download the binary to bin folder in $GOPATH
 - curl -L -s https://github.com/golang/dep/releases/download/v${DEP_VERSION}/dep-linux-amd64 -o $GOPATH/bin/dep
 # Make the binary executable
-- chmod +x $GOPATH/bin/dep
-- go get -d github.com/stretchr/testify/assert
+#- chmod +x $GOPATH/bin/dep
+#- go get -d github.com/stretchr/testify/assert
 install:
-# - dep ensure
+- export GO111MODULE=on
 script:
 - go test -v ./...
 - mkdir linux_amd64

blobporter.go

@@ -149,11 +149,10 @@ func main() {
 		tfer := transfer.NewTransfer(sourcePipeline.Source, targetPipeline, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, argsUtil.params.blockSize)
 		tfer.SetTransferTracker(argsUtil.params.tracker)
 		prog.newTransfer(float64(tfer.TotalSize), sourcesInfo, argsUtil.params.transferType)
-	
+
 		tfer.StartTransfer(argsUtil.params.dedupeLevel)
 		tfer.WaitForCompletion()
 
-	
 	}
 
 	if argsUtil.params.tracker != nil {
@@ -164,7 +163,6 @@ func main() {
 
 	prog.displayGlobalSummary()
 
-	
 }
 
 func getProgressBarDelegate(totalSize uint64, quietMode bool) func(r pipeline.WorkerResult, committedCount int, bufferLevel int) {

docs/gettingstarted.rst

@@ -9,7 +9,7 @@ Download, extract and set permissions
 
 ::
 
-    wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.6.15/bp_linux.tar.gz
+    wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.6.20/bp_linux.tar.gz
     tar -xvf bp_linux.tar.gz linux_amd64/blobporter
     chmod +x ~/linux_amd64/blobporter
     cd ~/linux_amd64
@@ -26,7 +26,7 @@ Set environment variables: ::
 Windows
 -------
 
-Download `BlobPorter.exe <https://github.com/Azure/blobporter/releases/download/v0.6.15/bp_windows.zip>`_
+Download `BlobPorter.exe <https://github.com/Azure/blobporter/releases/download/v0.6.20/bp_windows.zip>`_
 
 Set environment variables (if using the command prompt): ::
 

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/stretchr/testify v1.2.2
 	golang.org/x/crypto v0.0.0-20180621125126-a49355c7e3f8
 	golang.org/x/net v0.0.0-20180124060956-0ed95abb35c4
+	golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4
 	golang.org/x/sys v0.0.0-20180709060233-1b2967e3c290
 	golang.org/x/text v0.3.0
 )

go.sum

@@ -31,6 +31,8 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf
 golang.org/x/crypto v0.0.0-20180621125126-a49355c7e3f8/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/net v0.0.0-20180124060956-0ed95abb35c4 h1:BLERX6fu5dNMZcaGP2RzbrDZpHQbDkAoG9oiTRXbWr0=
 golang.org/x/net v0.0.0-20180124060956-0ed95abb35c4/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180709060233-1b2967e3c290/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

internal/azutil.go

@@ -32,7 +32,7 @@ func NewAzUtil(accountName string, accountKey string, container string, baseBlob
 	creds, err := azblob.NewSharedKeyCredential(accountName, accountKey)
 
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf(" error in the credential. Check the key value. Error:%v", err)
 	}
 
 	pipeline := newPipeline(creds, azblob.PipelineOptions{

internal/eventsink.go

@@ -88,13 +88,12 @@ type eventAggregateReq struct {
 
 type eventSink struct {
 	sync.Mutex
-	sums       map[string]EventItemAggregate
-	subs       map[EventSource][]EventSubscription
-	ondonesubs map[EventSource][]EventSubscription
-	eventsQ    chan EventItem
-	subsQ      chan EventSubscription
-	wg         *sync.WaitGroup
-	flushed    bool
+	sums         map[string]EventItemAggregate
+	subs         map[EventSource][]EventSubscription
+	ondonesubs   map[EventSource][]EventSubscription
+	eventsQ      chan EventItem
+	wg           *sync.WaitGroup
+	workerIsDone bool
 }
 
 func newEventSink() *eventSink {
@@ -104,72 +103,64 @@ func newEventSink() *eventSink {
 	return e
 
 }
-func (e *eventSink) Reset() error {
-	defer e.Unlock()
-	e.Lock()
-	if !e.flushed {
-		return fmt.Errorf("The sink is not flushed")
+func (e *eventSink) Reset() {
+
+	if !e.workerIsDone {
+		e.FlushAndWait()
 	}
+
 	e.init()
 	e.startWorker()
 
-	return nil
+	return
 }
 
 func (e *eventSink) startWorker() {
-	e.flushed = false
+	e.workerIsDone = false
 	go func() {
 		var sumEvent EventItemAggregate
-		defer func() { e.flushed = true }()
+		defer func() {
+			e.workerIsDone = true
+		}()
 		for {
-			select {
-			case event, ok := <-e.eventsQ:
-				if !ok {
-					defer func() {
-						for _, ondonesub := range e.ondonesubs {
-							for _, sub := range ondonesub {
-								for _, sumEvent := range e.sums {
-									if sub.Source == sumEvent.LastEventItem.Source {
-										sub.Delegate(sumEvent.LastEventItem, sumEvent)
-									}
+
+			event, ok := <-e.eventsQ
+
+			if !ok {
+
+				defer func() {
+					e.Lock()
+					defer e.Unlock()
+					for _, ondonesub := range e.ondonesubs {
+						for _, sub := range ondonesub {
+							for _, sumEvent := range e.sums {
+								if sub.Source == sumEvent.LastEventItem.Source {
+									sub.Delegate(sumEvent.LastEventItem, sumEvent)
 								}
 							}
 						}
-						e.wg.Done()
-					}()
-					return
-				}
-
-				if event.Action == Sum {
-					sumEvent = e.sums[event.key()]
-					sumEvent.NumItems++
-					sumEvent.LastEventItem = event
-					value := event.Data[0].Value.(float64)
-					sumEvent.Value += value
-					e.sums[event.key()] = sumEvent
-				}
-
-				esubs := e.subs[event.Source]
-				for _, sub := range esubs {
-					sub.Delegate(event, sumEvent)
-				}
-
-			case sub, ok := <-e.subsQ:
-				if !ok {
-					continue
-				}
-				switch sub.Type {
-				case RealTime:
-					subsForSource := e.subs[sub.Source]
-					subsForSource = append(subsForSource, sub)
-					e.subs[sub.Source] = subsForSource
-				case OnDone:
-					subsForSource := e.ondonesubs[sub.Source]
-					subsForSource = append(subsForSource, sub)
-					e.ondonesubs[sub.Source] = subsForSource
-				}
+					}
+					e.wg.Done()
+				}()
+
+				return
+			}
+
+			if event.Action == Sum {
+				sumEvent = e.sums[event.key()]
+				sumEvent.NumItems++
+				sumEvent.LastEventItem = event
+				value := event.Data[0].Value.(float64)
+				sumEvent.Value += value
+				e.sums[event.key()] = sumEvent
+			}
 
+			e.Lock()
+			esubs := e.subs[event.Source]
+			for _, sub := range esubs {
+				sub.Delegate(event, sumEvent)
 			}
+			e.Unlock()
 		}
 	}()
 }
@@ -178,28 +169,37 @@ func (e *eventSink) init() {
 	e.subs = make(map[EventSource][]EventSubscription)
 	e.ondonesubs = make(map[EventSource][]EventSubscription)
 	e.eventsQ = make(chan EventItem, 10000)
-	e.subsQ = make(chan EventSubscription, 100)
 	e.wg = &sync.WaitGroup{}
 	e.wg.Add(1)
 }
 
 //FlushAndWait closese the sink's channels as waits for processing of pending events
 func (e *eventSink) FlushAndWait() {
 	close(e.eventsQ)
-	close(e.subsQ)
 	e.wg.Wait()
+
 }
 
 //AddSubscription adds a subscription to the event sink
 func (e *eventSink) AddSubscription(source EventSource, subType EventSubscriptionType, delegate EventDelegate) {
-	select {
-	case e.subsQ <- EventSubscription{
+	e.Lock()
+	defer e.Unlock()
+
+	sub := EventSubscription{
 		Delegate: delegate,
 		Source:   source,
 		Type:     subType,
-	}:
-	default:
-		panic(fmt.Errorf("AddSubscription failed the channel is closed or full"))
+	}
+
+	switch subType {
+	case RealTime:
+		subsForSource := e.subs[source]
+		subsForSource = append(subsForSource, sub)
+		e.subs[source] = subsForSource
+	case OnDone:
+		subsForSource := e.ondonesubs[source]
+		subsForSource = append(subsForSource, sub)
+		e.ondonesubs[source] = subsForSource
 	}
 }
 

internal/eventsink_test.go

@@ -7,15 +7,19 @@ import (
 )
 
 type mockDel struct {
-	numOfEventItem          int
-	numOfEventItemAggregate int
-	numOfCalls              int
+	numOfEventItem  int
+	numOfCalls      int
+	aggregateResult int
 }
 
 func (m *mockDel) delegate(e EventItem, a EventItemAggregate) {
 	m.numOfEventItem++
-	m.numOfEventItemAggregate++
+
 	m.numOfCalls++
+
+	if e.Action == Sum {
+		m.aggregateResult = int(a.Value)
+	}
 }
 
 const testSource EventSource = 10
@@ -50,8 +54,8 @@ func TestOnDoneEvents(t *testing.T) {
 
 	es.FlushAndWait()
 
-	assert.Equal(t, md.numOfEventItemAggregate, 3, "Values must match")
-	assert.Equal(t, md.numOfCalls, 3, "Values must match")
+	assert.Equal(t, md.aggregateResult, 3, "Values must match")
+	assert.Equal(t, md.numOfCalls, 1, "Values must match")
 }
 
 func TestFlushCycle(t *testing.T) {
@@ -61,30 +65,25 @@ func TestFlushCycle(t *testing.T) {
 	es.AddSubscription(testSource, RealTime, md.delegate)
 	es.AddEvent(testSource, testEvent, "", EventData{Value: 1})
 
-	assert.Equal(t, md.numOfEventItem, 1, "Values must match")
-	assert.Equal(t, md.numOfCalls, 1, "Values must match")
+	es.Reset()
 
-	err := es.Reset()
-	assert.Error(t, err, "Reset must fail as it hasn't been flushed")
+	assert.Equal(t, 1, md.numOfEventItem, "Values must match")
+	assert.Equal(t, 1, md.numOfCalls, "Values must match")
 
-	//even if the reset fail the sink must be able to receive events
+	es.AddSubscription(testSource, RealTime, md.delegate)
 	es.AddEvent(testSource, testEvent, "", EventData{Value: 1})
-	assert.Equal(t, md.numOfEventItem, 2, "Values must match")
-	assert.Equal(t, md.numOfCalls, 2, "Values must match")
+
+	es.Reset()
+
+	assert.Equal(t, 2, md.numOfEventItem, "Values must match")
+	assert.Equal(t, 2, md.numOfCalls, "Values must match")
 
 	//add a sum event
 	es.AddSubscription(testSource, OnDone, md.delegate)
 	es.AddSumEvent(testSource, testEvent, "", oneFloat64)
 
-	//the sum event must be triggered at the end...
-	assert.Equal(t, md.numOfCalls, 2, "Values must match")
-
-	es.FlushAndWait()
-
-	assert.Equal(t, md.numOfEventItemAggregate, 1, "Values must match")
-	assert.Equal(t, md.numOfCalls, 3, "Values must match")
-
-	err = es.Reset()
-	assert.NoError(t, err, "Reset must succeed as the sink is flushed")
+	es.Reset()
 
+	assert.Equal(t, 1, md.aggregateResult, "Values must match")
+	assert.Equal(t, 3, md.numOfCalls,  "Values must match")
 }