[Scanner] handle current execution for blobstore

common/reconciliation/common/blobstoreIterator.go

@@ -37,13 +37,19 @@ type (
 	}
 )
 
+var scanTypeExecFnMap = map[ScanType]func(data []byte) (*ScanOutputEntity, error){
+	ConcreteExecutionType: deserializeConcreteExecution,
+	CurrentExecutionType:  deserializeCurrentExecution,
+}
+
 // NewBlobstoreIterator constructs a new iterator backed by blobstore.
 func NewBlobstoreIterator(
 	client blobstore.Client,
 	keys Keys,
+	scanType ScanType,
 ) ScanOutputIterator {
 	return &blobstoreIterator{
-		itr: pagination.NewIterator(keys.MinPage, getBlobstoreFetchPageFn(client, keys)),
+		itr: pagination.NewIterator(keys.MinPage, getBlobstoreFetchPageFn(client, keys, scanTypeExecFnMap[scanType])),
 	}
 }
 
@@ -64,6 +70,7 @@ func (i *blobstoreIterator) HasNext() bool {
 func getBlobstoreFetchPageFn(
 	client blobstore.Client,
 	keys Keys,
+	execDeserializeFunc func(data []byte) (*ScanOutputEntity, error),
 ) pagination.FetchFn {
 	return func(token pagination.PageToken) (pagination.Page, error) {
 		index := token.(int)
@@ -83,17 +90,11 @@ func getBlobstoreFetchPageFn(
 			if len(p) == 0 {
 				continue
 			}
-			// TODO handle multiple execution types
-			soe := ScanOutputEntity{
-				Execution: &ConcreteExecution{},
-			}
-			if err := json.Unmarshal(p, &soe); err != nil {
-				return pagination.Page{}, err
-			}
-			if err := ValidateConcreteExecution(soe.Execution.(*ConcreteExecution)); err != nil {
+			soe, err := execDeserializeFunc(p)
+			if err != nil {
 				return pagination.Page{}, err
 			}
-			executions = append(executions, &soe)
+			executions = append(executions, soe)
 		}
 		var nextPageToken interface{} = index + 1
 		if nextPageToken.(int) > keys.MaxPage {
@@ -106,3 +107,29 @@ func getBlobstoreFetchPageFn(
 		}, nil
 	}
 }
+
+func deserializeConcreteExecution(data []byte) (*ScanOutputEntity, error) {
+	soe := &ScanOutputEntity{
+		Execution: &ConcreteExecution{},
+	}
+	if err := json.Unmarshal(data, &soe); err != nil {
+		return nil, err
+	}
+	if err := ValidateConcreteExecution(soe.Execution.(*ConcreteExecution)); err != nil {
+		return nil, err
+	}
+	return soe, nil
+}
+
+func deserializeCurrentExecution(data []byte) (*ScanOutputEntity, error) {
+	soe := &ScanOutputEntity{
+		Execution: &CurrentExecution{},
+	}
+	if err := json.Unmarshal(data, &soe); err != nil {
+		return nil, err
+	}
+	if err := ValidateCurrentExecution(soe.Execution.(*CurrentExecution)); err != nil {
+		return nil, err
+	}
+	return soe, nil
+}

common/reconciliation/common/writerIterator_test.go

@@ -92,8 +92,9 @@ func (s *WriterIteratorSuite) TestWriterIterator() {
 	s.Equal(0, flushedKeys.MinPage)
 	s.Equal(9, flushedKeys.MaxPage)
 	s.Equal(Extension("test"), flushedKeys.Extension)
-	blobstoreItr := NewBlobstoreIterator(blobstore, *flushedKeys)
+	blobstoreItr := NewBlobstoreIterator(blobstore, *flushedKeys, ConcreteExecutionType)
 	i := 0
+	s.True(blobstoreItr.HasNext())
 	for blobstoreItr.HasNext() {
 		exec, err := blobstoreItr.Next()
 		s.NoError(err)

service/worker/scanner/executions/shard/fixer.go

@@ -58,7 +58,7 @@ func NewFixer(
 	id := uuid.New()
 	return &fixer{
 		shardID:          shardID,
-		itr:              common.NewBlobstoreIterator(blobstoreClient, keys),
+		itr:              common.NewBlobstoreIterator(blobstoreClient, keys, scanType),
 		skippedWriter:    common.NewBlobstoreWriter(id, common.SkippedExtension, blobstoreClient, blobstoreFlushThreshold),
 		failedWriter:     common.NewBlobstoreWriter(id, common.FailedExtension, blobstoreClient, blobstoreFlushThreshold),
 		fixedWriter:      common.NewBlobstoreWriter(id, common.FixedExtension, blobstoreClient, blobstoreFlushThreshold),