Merge branch 'hyperblock-processor' into implement-blocks-pool

factory/wsConnectorFactory.go

@@ -43,6 +43,7 @@ func CreateWSConnector(cfg config.WebSocketConfig) (process.WSConnector, error)
 		return nil, err
 	}
 
+	// TODO: move cache to config
 	cacheConfig := storageUnit.CacheConfig{
 		Type:        storageUnit.SizeLRUCache,
 		SizeInBytes: 209715200, // 200MB

process/blocksPool.go

@@ -93,6 +93,7 @@ func (bp *blocksPool) prunePersister(round uint64) error {
 	return nil
 }
 
+// PutBlock will put the provided outport block data to the pool
 func (bp *blocksPool) PutBlock(hash []byte, outportBlock *outport.OutportBlock) error {
 	bp.mutMap.Lock()
 	defer bp.mutMap.Unlock()
@@ -159,6 +160,7 @@ func (bp *blocksPool) putOutportBlock(hash []byte, outportBlock *outport.Outport
 	return nil
 }
 
+// GetBlock will return outport block data from the pool
 func (bp *blocksPool) GetBlock(hash []byte) (*outport.OutportBlock, error) {
 	data, err := bp.storer.Get(hash)
 	if err != nil {
@@ -174,6 +176,7 @@ func (bp *blocksPool) GetBlock(hash []byte) (*outport.OutportBlock, error) {
 	return outportBlock, nil
 }
 
+// IsInterfaceNil returns nil if there is no value under the interface
 func (bp *blocksPool) IsInterfaceNil() bool {
 	return bp == nil
 }

process/blocksPool_test.go

@@ -0,0 +1,3 @@
+package process_test
+
+// TODO: add tests

process/dataAggregator.go

@@ -12,18 +12,21 @@ type dataAggregator struct {
 	blocksPool BlocksPool
 }
 
+// NewDataAggregator will create a new data aggregator instance
 func NewDataAggregator(
 	blocksPool BlocksPool,
 ) (*dataAggregator, error) {
 	if check.IfNil(blocksPool) {
-		return nil, errNilBlocksPool
+		return nil, ErrNilBlocksPool
 	}
 
 	return &dataAggregator{
 		blocksPool: blocksPool,
 	}, nil
 }
 
+// ProcessHyperBlock will process meta outport block. It will try to fetch and aggregate
+// notarized shards data
 func (da *dataAggregator) ProcessHyperBlock(outportBlock *outport.OutportBlock) (*data.HyperOutportBlock, error) {
 	hyperOutportBlock := &data.HyperOutportBlock{}
 	hyperOutportBlock.MetaOutportBlock = outportBlock

process/dataAggregator_test.go

@@ -0,0 +1,57 @@
+package process_test
+
+import (
+	"encoding/hex"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/data/outport"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/process"
+	"github.com/multiversx/mx-chain-ws-connector-template-go/testscommon"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewDataAggregator(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil blocks pool", func(t *testing.T) {
+		t.Parallel()
+
+		da, err := process.NewDataAggregator(nil)
+		require.Nil(t, da)
+		require.Equal(t, process.ErrNilBlocksPool, err)
+	})
+
+	t.Run("should work", func(t *testing.T) {
+		t.Parallel()
+
+		da, err := process.NewDataAggregator(&testscommon.BlocksPoolStub{})
+		require.Nil(t, err)
+		require.False(t, da.IsInterfaceNil())
+	})
+}
+
+func TestDataAggregator_ProcessHyperBlock(t *testing.T) {
+	t.Parallel()
+
+	headerHash := []byte("headerHash1")
+
+	shardOutportBlock := createOutportBlock()
+	shardOutportBlock.BlockData.HeaderHash = headerHash
+
+	blocksPoolStub := &testscommon.BlocksPoolStub{
+		GetBlockCalled: func(hash []byte) (*outport.OutportBlock, error) {
+
+			return shardOutportBlock, nil
+		},
+	}
+
+	da, err := process.NewDataAggregator(blocksPoolStub)
+	require.Nil(t, err)
+
+	outportBlock := createMetaOutportBlock()
+	outportBlock.NotarizedHeadersHashes = []string{hex.EncodeToString(headerHash)}
+
+	hyperOutportBlock, err := da.ProcessHyperBlock(outportBlock)
+	require.Nil(t, err)
+	require.Equal(t, shardOutportBlock, hyperOutportBlock.NotarizedHeadersOutportData[0].OutportBlock)
+}

process/dataProcessor.go

@@ -26,16 +26,16 @@ func NewDataProcessor(
 	blockCreator BlockContainerHandler,
 ) (DataProcessor, error) {
 	if publisher == nil {
-		return nil, errNilWriter
+		return nil, ErrNilPublisher
 	}
 	if check.IfNil(blocksPool) {
-		return nil, errNilBlocksPool
+		return nil, ErrNilBlocksPool
 	}
 	if check.IfNil(marshaller) {
-		return nil, errNilMarshaller
+		return nil, ErrNilMarshaller
 	}
 	if check.IfNil(dataAggregator) {
-		return nil, errNilDataAggregator
+		return nil, ErrNilDataAggregator
 	}
 
 	dp := &dataProcessor{
@@ -71,7 +71,7 @@ func (dp *dataProcessor) saveBlock(marshalledData []byte) error {
 	}
 
 	if outportBlock == nil || outportBlock.BlockData == nil {
-		return errNilOutportBlockData
+		return ErrNilOutportBlockData
 	}
 
 	log.Info("saving block", "hash", outportBlock.BlockData.GetHeaderHash(), "shardID", outportBlock.ShardID)