feat(sync): add actor to sync block each 8 seconds

app/actor/block/actor.go

@@ -0,0 +1,60 @@
+package block
+
+import (
+ "fmt"
+ "time"
+
+ "okp4/nemeton-leaderboard/app/messages"
+
+ "github.com/asynkron/protoactor-go/actor"
+ "github.com/cosmos/cosmos-sdk/client/grpc/tmservice"
+ "github.com/rs/zerolog/log"
+)
+
+type Actor struct {
+ grpcClientProps *actor.Props
+ grpcClient *actor.PID
+ currentBlock int64
+}
+
+func NewActor(grpcClientProps *actor.Props, blockHeight int64) *Actor {
+ return &Actor{
+ grpcClientProps: grpcClientProps,
+ grpcClient: nil,
+ currentBlock: blockHeight,
+ }
+}
+
+func (a *Actor) Receive(ctx actor.Context) {
+ switch ctx.Message().(type) {
+ case *actor.Started:
+ a.grpcClient = ctx.Spawn(a.grpcClientProps)
+ a.startSynchronization(ctx)
+ }
+}
+
+func (a *Actor) startSynchronization(ctx actor.Context) {
+ go func() {
+ for range time.Tick(8 * time.Second) {
+ result, err := ctx.RequestFuture(a.grpcClient, &messages.GetBlock{Height: a.currentBlock}, 5*time.Second).Result()
+ if err != nil {
+ log.Err(err).Msg("⚠️ Failed request current block.")
+ continue
+ }
+
+ var block *tmservice.Block
+ switch resp := result.(type) {
+ case *messages.GetBlockResponse:
+ block = resp.Block
+ default:
+ log.Panic().Err(fmt.Errorf("wrong response message")).Msg("❌ Could not get block.")
+ }
+
+ log.Info().Int64("blockHeight", block.Header.Height).Msg("Successful request block")
+
+ // TODO: Send to event handler the new block received
+
+ a.currentBlock += 1
+ }
+ }()
+}

app/system/bootstrap.go

@@ -1,6 +1,7 @@
 package system
 
 import (
+ "okp4/nemeton-leaderboard/app/actor/block"
  "okp4/nemeton-leaderboard/app/actor/cosmos"
  "okp4/nemeton-leaderboard/app/actor/event"
  "okp4/nemeton-leaderboard/app/actor/graphql"
@@ -39,7 +40,7 @@ func (app *App) Stop() error {
 }
 
 func boot(ctx actor.Context, listenAddr, mongoURI, dbName, grpcAddr string, tls credentials.TransportCredentials) {
- _ = actor.PropsFromProducer(func() actor.Actor {
+ grpcClientProps := actor.PropsFromProducer(func() actor.Actor {
  grpcClient, err := cosmos.NewGrpcClient(grpcAddr, tls)
  if err != nil {
  log.Panic().Err(err).Msg("❌ Could not create grpc client")
@@ -48,6 +49,13 @@ func boot(ctx actor.Context, listenAddr, mongoURI, dbName, grpcAddr string, tls
  return grpcClient
  })
 
+ blockSync := actor.PropsFromProducer(func() actor.Actor {
+ return block.NewActor(grpcClientProps, 10112)
+ })
+ if _, err := ctx.SpawnNamed(blockSync, "blockSync"); err != nil {
+ log.Panic().Err(err).Msg("❌Could not create block sync actor")
+ }
+
  graphqlProps := actor.PropsFromProducer(func() actor.Actor {
  return graphql.NewActor(listenAddr)
  })