update data-layer-sdk version

Synternet · Aug 20, 2024 · 6527d49 · 6527d49
1 parent 6e925ce
commit 6527d49
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 7 deletions.
diff --git a/go.mod b/go.mod
@@ -67,7 +67,7 @@ require (
 	github.com/nats-io/jwt v0.3.2
 	github.com/nats-io/nkeys v0.4.6
 	github.com/prometheus/client_golang v1.19.0
-	github.com/syntropynet/data-layer-sdk v0.1.0
+	github.com/synternet/data-layer-sdk v0.4.2
 	golang.org/x/sync v0.7.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20240227224415-6ceb2ff114de
 	gopkg.in/DataDog/dd-trace-go.v1 v1.62.0

diff --git a/go.sum b/go.sum
@@ -1161,8 +1161,8 @@ github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
 github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 h1:epCh84lMvA70Z7CTTCmYQn2CKbY8j86K7/FAIr141uY=
 github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7/go.mod h1:q4W45IWZaF22tdD+VEXcAWRA037jwmWEB5VWYORlTpc=
-github.com/syntropynet/data-layer-sdk v0.1.0 h1:fnLYLeIpA31IQHRgpIrb2fak2cP4P1hLBBw57RvNJ2s=
-github.com/syntropynet/data-layer-sdk v0.1.0/go.mod h1:2LxGQ1LF1m7CHdUJLMl2zbbz7A8T4lSVLKEClF5JSww=
+github.com/synternet/data-layer-sdk v0.4.2 h1:tfsMgkG6VkITxWOHV7jZuQpRt4KnjDyiVT76XzwKXAc=
+github.com/synternet/data-layer-sdk v0.4.2/go.mod h1:iHEVwnB8bpTRtMMiahVX5fGP/P4toHNIB6xsCstsCFM=
 github.com/tendermint/go-amino v0.16.0 h1:GyhmgQKvqF82e2oZeuMSp9JTN0N09emoSZlb2lyGa2E=
 github.com/tendermint/go-amino v0.16.0/go.mod h1:TQU0M1i/ImAo+tYpZi73AU3V/dKeCoMC9Sphe2ZwGME=
 github.com/tidwall/btree v1.7.0 h1:L1fkJH/AuEh5zBnnBbmTwQ5Lt+bRJ5A8EWecslvo9iI=

diff --git a/injective-chain/app/app.go b/injective-chain/app/app.go
@@ -10,6 +10,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"strconv"
+	"time"
 
 	icahost "github.com/cosmos/ibc-go/v8/modules/apps/27-interchain-accounts/host"
 	porttypes "github.com/cosmos/ibc-go/v8/modules/core/05-port/types"
@@ -176,8 +177,8 @@ import (
 	// unnamed import of statik for swagger UI support
 	_ "github.com/InjectiveLabs/injective-core/client/docs/statik"
 
-	publisherOptions "github.com/syntropynet/data-layer-sdk/pkg/options"
-	publisher "github.com/syntropynet/data-layer-sdk/pkg/service"
+	publisherOptions "github.com/synternet/data-layer-sdk/pkg/options"
+	publisher "github.com/synternet/data-layer-sdk/pkg/service"
 )
 
 func init() {
@@ -530,6 +531,7 @@ func initInjectiveApp(
 		publisher.WithPrefix(os.Getenv("PUB_PREFIX")),
 		publisher.WithNats(natsConnection),
 		publisher.WithNKeySeed(os.Getenv("NATS_NKEY")),
+		publisher.WithTelemetryPeriod(time.Second * 10),
 		publisher.WithVerbose(false),
 	}
 

diff --git a/injective-chain/app/publisher.go b/injective-chain/app/publisher.go
@@ -11,6 +11,7 @@ import (
 	"github.com/gorilla/websocket"
 	"github.com/nats-io/jwt"
 	"github.com/nats-io/nkeys"
+	"google.golang.org/protobuf/reflect/protoreflect"
 	"os"
 	"time"
 
@@ -104,6 +105,11 @@ type BlockProposal struct {
 	ProposerAddress    []byte    `json:"proposer_address"`
 }
 
+func (BlockProposal) ProtoReflect() protoreflect.Message { return nil }
+func (Block) ProtoReflect() protoreflect.Message         { return nil }
+func (Mempool) ProtoReflect() protoreflect.Message       { return nil }
+func (Transaction) ProtoReflect() protoreflect.Message   { return nil }
+
 func decodeTxs(txBytes [][]byte, encfg injcodectypes.EncodingConfig) ([]Transaction, error) {
 	var transactions []Transaction
 
@@ -118,6 +124,20 @@ func decodeTxs(txBytes [][]byte, encfg injcodectypes.EncodingConfig) ([]Transact
 	return transactions, nil
 }
 
+func (app *InjectiveApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (res *abci.ResponseFinalizeBlock, err error) {
+	res, err = app.BaseApp.FinalizeBlock(req)
+	if err == nil {
+		for _, rawTx := range req.Txs {
+			transaction, err := decodeTx(rawTx, app.encfg)
+			if err != nil {
+				fmt.Println(err)
+			}
+			app.publisher.Publish(transaction, "tx")
+		}
+	}
+	return res, err
+}
+
 func (app *InjectiveApp) ProcessProposal(req *abci.RequestProcessProposal) (resp *abci.ResponseProcessProposal, err error) {
 	decodedTxs, err := decodeTxs(req.Txs, app.encfg)
 	if err != nil {
@@ -379,7 +399,8 @@ func (app *InjectiveApp) StartWebSocketListener(conn *websocket.Conn) {
 						temp, _ := json.Marshal(txsstring)
 						err = json.Unmarshal(temp, &blockEvent)
 						blockEvent.Result.Data.Value.Block.Data.Txs = decodedTxs
-						app.publisher.Publish(blockEvent.Result.Data.Value.Block, "block")
+						blockBytes, _ := json.Marshal(blockEvent.Result.Data.Value.Block)
+						app.publisher.PublishBuf(blockBytes, "block")
 					}
 				}
 			}
@@ -395,7 +416,7 @@ func (app *InjectiveApp) PublishBlocksAndTxs() {
 	}
 	defer conn.Close()
 
-	err = app.SubscribeToEvents(conn, "NewBlock", "Tx")
+	err = app.SubscribeToEvents(conn, "NewBlock")
 	if err != nil {
 		fmt.Println("Failed to subscribe to events:", err)
 		return