Merge pull request #7 from everFinance/feature/subscibe-blocks

add block subscribe

Author: zyjblockchain

example/all-tx-syncer.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+
 	"github.com/everFinance/arsyncer"
 )
 
@@ -11,7 +12,7 @@ func main() {
 	startHeight := int64(879220)
 	arNode := "https://arweave.net"
 	concurrencyNumber := 10 // runtime concurrency number, default 10
-	s := arsyncer.New(startHeight, nullFilterParams, arNode, concurrencyNumber, 15)
+	s := arsyncer.New(startHeight, nullFilterParams, arNode, concurrencyNumber, 15, true)
 
 	// run
 	s.Run()
@@ -21,7 +22,10 @@ func main() {
 		select {
 		case sTx := <-s.SubscribeTxCh():
 			// process synced txs
-			fmt.Println(sTx[0].BlockHeight)
+			fmt.Println("Tx", sTx[0].ID)
+		case sBlock := <-s.SubscribeBlockCh():
+			fmt.Println("Block", sBlock.Height, sBlock.IndepHash)
 		}
+
 	}
 }

example/from-tx-syncer.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+
 	"github.com/everFinance/arsyncer"
 )
 
@@ -14,7 +15,7 @@ func main() {
 	startHeight := int64(804524)
 	arNode := "https://arweave.net"
 	concurrencyNumber := 100 // runtime concurrency number, default 10
-	s := arsyncer.New(startHeight, ownerFilterParams, arNode, concurrencyNumber, 15)
+	s := arsyncer.New(startHeight, ownerFilterParams, arNode, concurrencyNumber, 15, false)
 
 	// run
 	s.Run()

example/swc-syncer.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+
 	"github.com/everFinance/arsyncer"
 	"github.com/everFinance/goar/types"
 )
@@ -16,7 +17,7 @@ func main() {
 	startHeight := int64(472810)
 	arNode := "https://arweave.net"
 	concurrencyNumber := 50 // runtime concurrency number, default 10
-	s := arsyncer.New(startHeight, swcFilterParams, arNode, concurrencyNumber, 15)
+	s := arsyncer.New(startHeight, swcFilterParams, arNode, concurrencyNumber, 15, false)
 
 	// run
 	s.Run()

example/target-tx-syncer.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+
 	"github.com/everFinance/arsyncer"
 )
 
@@ -14,7 +15,7 @@ func main() {
 	startHeight := int64(811484)
 	arNode := "https://arweave.net"
 	concurrencyNumber := 100 // runtime concurrency number, default 10
-	s := arsyncer.New(startHeight, ownerFilterParams, arNode, concurrencyNumber, 15)
+	s := arsyncer.New(startHeight, ownerFilterParams, arNode, concurrencyNumber, 15, false)
 
 	// run
 	s.Run()

go.mod

@@ -7,8 +7,10 @@ require (
 	github.com/getsentry/sentry-go v0.11.0
 	github.com/go-co-op/gocron v1.11.0
 	github.com/inconshreveable/log15 v0.0.0-20201112154412-8562bdadbbac
+	github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32 // indirect
 	github.com/panjf2000/ants/v2 v2.4.6
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/stretchr/testify v1.7.0
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
-	gopkg.in/h2non/gentleman.v2 v2.0.5 // indirect
+	gopkg.in/h2non/gentleman.v2 v2.0.5
 )

syncer.go

@@ -2,14 +2,15 @@ package arsyncer
 
 import (
 	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
 	"github.com/everFinance/goar"
 	"github.com/everFinance/goar/types"
 	"github.com/everFinance/goar/utils"
 	"github.com/go-co-op/gocron"
 	"github.com/panjf2000/ants/v2"
-	"sync"
-	"sync/atomic"
-	"time"
 )
 
 var log = NewLog("syncer")
@@ -27,9 +28,12 @@ type Syncer struct {
 	blockIdxs            *BlockIdxs
 	scheduler            *gocron.Scheduler
 	peers                []string
+
+	subscribeBlock     bool
+	SubscribeBlockChan chan *types.Block
 }
 
-func New(startHeight int64, filterParams FilterParams, arNode string, conNum int, stableDistance int64) *Syncer {
+func New(startHeight int64, filterParams FilterParams, arNode string, conNum int, stableDistance int64, subscribeBlock bool) *Syncer {
 	if conNum <= 0 {
 		conNum = 10 // default concurrency of number is 10
 	}
@@ -54,6 +58,7 @@ func New(startHeight int64, filterParams FilterParams, arNode string, conNum int
 		curHeight:            startHeight,
 		FilterParams:         filterParams,
 		blockChan:            make(chan *types.Block, 5*conNum),
+		SubscribeBlockChan:   make(chan *types.Block, 5*conNum),
 		blockTxsChan:         make(chan []SubscribeTx, conNum),
 		SubscribeChan:        make(chan []SubscribeTx, conNum),
 		arClient:             arCli,
@@ -63,6 +68,7 @@ func New(startHeight int64, filterParams FilterParams, arNode string, conNum int
 		blockIdxs:            idxs,
 		scheduler:            gocron.NewScheduler(time.UTC),
 		peers:                peers,
+		subscribeBlock:       subscribeBlock,
 	}
 }
 
@@ -84,6 +90,10 @@ func (s *Syncer) SubscribeTxCh() <-chan []SubscribeTx {
 	return s.SubscribeChan
 }
 
+func (s *Syncer) SubscribeBlockCh() <-chan *types.Block {
+	return s.SubscribeBlockChan
+}
+
 func (s *Syncer) GetSyncedHeight() int64 {
 	return atomic.LoadInt64(&s.nextSubscribeTxBlock)
 }
@@ -118,6 +128,9 @@ func (s *Syncer) pollingBlock() {
 			// add chan
 			for _, b := range blocks {
 				s.blockChan <- b
+				if s.subscribeBlock {
+					s.SubscribeBlockChan <- b
+				}
 			}
 		}
 	}