Mx 16271 indexing incoming tokens

client/disabled/elasticClient.go

@@ -0,0 +1,73 @@
+package disabled
+
+import (
+	"bytes"
+	"context"
+)
+
+type elasticClient struct{}
+
+// NewDisabledElasticClient -
+func NewDisabledElasticClient() *elasticClient {
+	return &elasticClient{}
+}
+
+// DoBulkRequest -
+func (ec *elasticClient) DoBulkRequest(_ context.Context, _ *bytes.Buffer, _ string) error {
+	return nil
+}
+
+// DoQueryRemove -
+func (ec *elasticClient) DoQueryRemove(_ context.Context, _ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// DoMultiGet -
+func (ec *elasticClient) DoMultiGet(_ context.Context, _ []string, _ string, _ bool, _ interface{}) error {
+	return nil
+}
+
+// DoScrollRequest -
+func (ec *elasticClient) DoScrollRequest(_ context.Context, _ string, _ []byte, _ bool, _ func(responseBytes []byte) error) error {
+	return nil
+}
+
+// DoCountRequest -
+func (ec *elasticClient) DoCountRequest(_ context.Context, _ string, _ []byte) (uint64, error) {
+	return 0, nil
+}
+
+// UpdateByQuery -
+func (ec *elasticClient) UpdateByQuery(_ context.Context, _ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// PutMappings -
+func (ec *elasticClient) PutMappings(_ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// CheckAndCreateIndex -
+func (ec *elasticClient) CheckAndCreateIndex(_ string) error {
+	return nil
+}
+
+// CheckAndCreateAlias -
+func (ec *elasticClient) CheckAndCreateAlias(_ string, _ string) error {
+	return nil
+}
+
+// CheckAndCreateTemplate -
+func (ec *elasticClient) CheckAndCreateTemplate(_ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// CheckAndCreatePolicy -
+func (ec *elasticClient) CheckAndCreatePolicy(_ string, _ *bytes.Buffer) error {
+	return nil
+}
+
+// IsInterfaceNil - returns true if there is no value under the interface
+func (ec *elasticClient) IsInterfaceNil() bool {
+	return ec == nil
+}

client/disabled/elasticClient_test.go

@@ -0,0 +1,31 @@
+package disabled
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/core/check"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDisabledElasticClient_MethodsShouldNotPanic(t *testing.T) {
+	t.Parallel()
+
+	ec := NewDisabledElasticClient()
+	require.False(t, check.IfNil(ec))
+
+	require.NotPanics(t, func() {
+		_ = ec.DoBulkRequest(context.Background(), new(bytes.Buffer), "")
+		_ = ec.DoQueryRemove(context.Background(), "", new(bytes.Buffer))
+		_ = ec.DoMultiGet(context.Background(), make([]string, 0), "", true, nil)
+		_ = ec.DoScrollRequest(context.Background(), "", []byte(""), true, nil)
+		_, _ = ec.DoCountRequest(context.Background(), "", []byte(""))
+		_ = ec.UpdateByQuery(context.Background(), "", new(bytes.Buffer))
+		_ = ec.PutMappings("", new(bytes.Buffer))
+		_ = ec.CheckAndCreateIndex("")
+		_ = ec.CheckAndCreateAlias("", "")
+		_ = ec.CheckAndCreateTemplate("", new(bytes.Buffer))
+		_ = ec.CheckAndCreatePolicy("", new(bytes.Buffer))
+	})
+}

data/tokens.go

@@ -45,7 +45,7 @@ type SourceToken struct {
 	CurrentOwner string `json:"currentOwner"`
 }
 
-// ResponseTokenInfo is the structure for the token info response
+// ResponseTokenInfo is the structure for the tokens info response
 type ResponseTokenInfo struct {
 	Docs []ResponseTokenInfoDB `json:"docs"`
 }

docker-compose.yml

@@ -15,7 +15,7 @@ services:
     ports:
       - "9200:9200"
       - "9300:9300"
-  elasticsearch2:
+  main-elasticsearch:
     container_name: es-container2
     image: docker.elastic.co/elasticsearch/elasticsearch:7.16.1
     environment:

factory/runType/runTypeComponentsFactory.go

@@ -17,7 +17,7 @@ func (rtcf *runTypeComponentsFactory) Create() (*runTypeComponents, error) {
 	return &runTypeComponents{
 		txHashExtractor:    transactions.NewTxHashExtractor(),
 		rewardTxData:       transactions.NewRewardTxData(),
-		indexTokensHandler: tokens.NewIndexTokensHandler(),
+		indexTokensHandler: tokens.NewDisabledIndexTokensHandler(),
 	}, nil
 }
 

factory/runType/runTypeComponentsHandler.go

@@ -109,7 +109,7 @@ func (mrtc *managedRunTypeComponents) RewardTxDataCreator() transactions.RewardT
 	return mrtc.runTypeComponents.rewardTxData
 }
 
-// IndexTokensHandlerCreator return index tokens handler
+// IndexTokensHandlerCreator returns the index tokens handler
 func (mrtc *managedRunTypeComponents) IndexTokensHandlerCreator() elasticproc.IndexTokensHandler {
 	mrtc.mutRunTypeCoreComponents.Lock()
 	defer mrtc.mutRunTypeCoreComponents.Unlock()

factory/runType/sovereignRunTypeComponentsFactory.go

@@ -1,6 +1,16 @@
 package runType
 
 import (
+	"math"
+	"net/http"
+	"time"
+
+	"github.com/elastic/go-elasticsearch/v7"
+
+	"github.com/multiversx/mx-chain-es-indexer-go/client"
+	"github.com/multiversx/mx-chain-es-indexer-go/client/disabled"
+	"github.com/multiversx/mx-chain-es-indexer-go/client/logging"
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/factory"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/tokens"
 	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/transactions"
@@ -21,7 +31,12 @@ func NewSovereignRunTypeComponentsFactory(mainChainElastic factory.ElasticConfig
 
 // Create will create the run type components
 func (srtcf *sovereignRunTypeComponentsFactory) Create() (*runTypeComponents, error) {
-	sovIndexTokensHandler, err := tokens.NewSovereignIndexTokensHandler(srtcf.mainChainElastic, srtcf.esdtPrefix)
+	mainChainElasticClient, err := createMainChainElasticClient(srtcf.mainChainElastic)
+	if err != nil {
+		return nil, err
+	}
+
+	sovIndexTokensHandler, err := tokens.NewSovereignIndexTokensHandler(srtcf.mainChainElastic.Enabled, mainChainElasticClient, srtcf.esdtPrefix)
 	if err != nil {
 		return nil, err
 	}
@@ -33,6 +48,26 @@ func (srtcf *sovereignRunTypeComponentsFactory) Create() (*runTypeComponents, er
 	}, nil
 }
 
+func createMainChainElasticClient(mainChainElastic factory.ElasticConfig) (elasticproc.DatabaseClientHandler, error) {
+	if mainChainElastic.Enabled {
+		argsEsClient := elasticsearch.Config{
+			Addresses:     []string{mainChainElastic.Url},
+			Username:      mainChainElastic.UserName,
+			Password:      mainChainElastic.Password,
+			Logger:        &logging.CustomLogger{},
+			RetryOnStatus: []int{http.StatusConflict},
+			RetryBackoff:  retryBackOff,
+		}
+		return client.NewElasticClient(argsEsClient)
+	} else {
+		return disabled.NewDisabledElasticClient(), nil
+	}
+}
+
+func retryBackOff(attempt int) time.Duration {
+	return time.Duration(math.Exp2(float64(attempt))) * time.Second
+}
+
 // IsInterfaceNil returns true if there is no value under the interface
 func (srtcf *sovereignRunTypeComponentsFactory) IsInterfaceNil() bool {
 	return srtcf == nil

integrationtests/consts.go

@@ -6,7 +6,7 @@ const (
 	//nolint
 	esURL = "http://localhost:9200"
 	//nolint
-	es2URL = "http://localhost:9201"
+	esMainURL = "http://localhost:9201"
 	//nolint
 	addressPrefix = "erd"
 )

integrationtests/incomingSCR_test.go

@@ -18,7 +18,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	indexerData "github.com/multiversx/mx-chain-es-indexer-go/process/dataindexer"
-	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc/factory"
+	"github.com/multiversx/mx-chain-es-indexer-go/process/elasticproc"
 )
 
 type esToken struct {
@@ -72,18 +72,16 @@ func createTokens() ([]esToken, []esNft) {
 func TestCrossChainTokensIndexingFromMainChain(t *testing.T) {
 	setLogLevelDebug()
 
-	mainChainEs := factory.ElasticConfig{
-		Enabled: true,
-		Url:     es2URL,
-	}
+	mainEsClient, err := createESClient(esMainURL)
+	require.Nil(t, err)
 
 	tokens, nfts := createTokens()
-	createTokensInSourceEs(t, mainChainEs, tokens, nfts)
+	createTokensInSourceEs(t, mainEsClient, tokens, nfts)
 
 	esClient, err := createESClient(esURL)
 	require.Nil(t, err)
 
-	esProc, err := CreateSovereignElasticProcessor(esClient, mainChainEs)
+	esProc, err := CreateSovereignElasticProcessor(esClient, mainEsClient)
 	require.Nil(t, err)
 
 	allTokens := getAllTokensIDs(tokens, nfts)
@@ -143,10 +141,7 @@ func TestCrossChainTokensIndexingFromMainChain(t *testing.T) {
 	}
 }
 
-func createTokensInSourceEs(t *testing.T, es factory.ElasticConfig, tokens []esToken, nfts []esNft) {
-	esClient, err := createESClient(es.Url)
-	require.Nil(t, err)
-
+func createTokensInSourceEs(t *testing.T, esClient elasticproc.DatabaseClientHandler, tokens []esToken, nfts []esNft) {
 	esProc, err := CreateElasticProcessor(esClient)
 	require.Nil(t, err)
 
@@ -254,7 +249,12 @@ func getAllTokensIDs(tokens []esToken, nfts []esNft) []string {
 func getAllNftIDs(nfts []esNft) []string {
 	allNfts := make([]string, 0)
 	for _, nft := range nfts {
-		allNfts = append(allNfts, nft.Collection+"-"+hex.EncodeToString(big.NewInt(0).SetUint64(nft.Nonce).Bytes()))
+		nonceBytes := big.NewInt(0).SetUint64(nft.Nonce).Bytes()
+		nonceHex := hex.EncodeToString(nonceBytes)
+		nftIdentifier := nft.Collection + "-" + nonceHex
+
+		allNfts = append(allNfts, nftIdentifier)
+
 	}
 	return allNfts
 }

integrationtests/utils.go

@@ -67,7 +67,7 @@ func CreateElasticProcessor(
 		Denomination:       18,
 		TxHashExtractor:    transactions.NewTxHashExtractor(),
 		RewardTxData:       transactions.NewRewardTxData(),
-		IndexTokensHandler: tokens.NewIndexTokensHandler(),
+		IndexTokensHandler: tokens.NewDisabledIndexTokensHandler(),
 	}
 
 	return factory.CreateElasticProcessor(args)
@@ -76,9 +76,9 @@ func CreateElasticProcessor(
 // CreateSovereignElasticProcessor -
 func CreateSovereignElasticProcessor(
 	esClient elasticproc.DatabaseClientHandler,
-	mainChainEs factory.ElasticConfig,
+	mainEsClient elasticproc.DatabaseClientHandler,
 ) (dataindexer.ElasticProcessor, error) {
-	sovIndexTokens, _ := tokens.NewSovereignIndexTokensHandler(mainChainEs, sovEsdtPrefix)
+	sovIndexTokens, _ := tokens.NewSovereignIndexTokensHandler(true, mainEsClient, sovEsdtPrefix)
 
 	args := factory.ArgElasticProcessorFactory{
 		Marshalizer:              &mock.MarshalizerMock{},

process/elasticproc/tokens/indexTokensHandler.go

@@ -7,8 +7,8 @@ import (
 
 type indexTokensHandler struct{}
 
-// NewIndexTokensHandler creates a new index tokens handler
-func NewIndexTokensHandler() *indexTokensHandler {
+// NewDisabledIndexTokensHandler creates a new disabled index tokens handler
+func NewDisabledIndexTokensHandler() *indexTokensHandler {
 	return &indexTokensHandler{}
 }
 

process/elasticproc/tokens/indexTokensHandler_test.go

@@ -9,14 +9,14 @@ import (
 func TestNewIndexTokensHandler(t *testing.T) {
 	t.Parallel()
 
-	ith := NewIndexTokensHandler()
+	ith := NewDisabledIndexTokensHandler()
 	require.False(t, ith.IsInterfaceNil())
 }
 
 func TestIndexTokensHandler_IndexCrossChainTokens(t *testing.T) {
 	t.Parallel()
 
-	ith := NewIndexTokensHandler()
+	ith := NewDisabledIndexTokensHandler()
 	err := ith.IndexCrossChainTokens(nil, nil, nil)
 	require.NoError(t, err)
 }