-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Bruno Calza <[email protected]>
- Loading branch information
1 parent
2fdd5e2
commit 48b51c5
Showing
8 changed files
with
238 additions
and
67 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,6 @@ | ||
| module github.com/textileio/go-tableland | ||
|
|
||
| go 1.18 | ||
| go 1.19 | ||
|
|
||
| require ( | ||
| cloud.google.com/go/bigquery v1.45.0 | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| package publisher | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync" | ||
| "time" | ||
|
|
||
| logger "github.com/rs/zerolog/log" | ||
| "github.com/textileio/go-tableland/pkg/merkletree" | ||
| "github.com/textileio/go-tableland/pkg/wallet" | ||
| ) | ||
|
|
||
| // MerkleRootPublisher is responsible for building Merkle Tree and publishing the root. | ||
| type MerkleRootPublisher struct { | ||
| store LeavesStore // where leaves are stored | ||
| registry MerkleRootRegistry // where root will be published | ||
|
|
||
| wallet *wallet.Wallet | ||
| interval time.Duration | ||
| fetchAmount int | ||
|
|
||
| quitOnce sync.Once | ||
| quit chan struct{} | ||
| } | ||
|
|
||
| // NewMerkleRootPublisher creates a new publisher. | ||
| func NewMerkleRootPublisher(s MerkleRootRegistry, wallet *wallet.Wallet, interval time.Duration) *MerkleRootPublisher { | ||
| return &MerkleRootPublisher{ | ||
| wallet: wallet, | ||
| interval: interval, | ||
| fetchAmount: 100, | ||
| quit: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| var log = logger.With(). | ||
| Str("component", "merkletreepublisher"). | ||
| Logger() | ||
|
|
||
| // Start starts the publisher. | ||
| func (p *MerkleRootPublisher) Start() { | ||
| ctx := context.Background() | ||
|
|
||
| ticker := time.NewTicker(p.interval) | ||
| go func() { | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| if err := p.publish(ctx); err != nil { | ||
| log.Err(err).Msg("failed to publish merkle root") | ||
| } | ||
| case <-p.quit: | ||
| log.Info().Msg("quiting merkle root publisher") | ||
| ticker.Stop() | ||
| return | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| // Close closes the published goroutine. | ||
| func (p *MerkleRootPublisher) Close() { | ||
| p.quitOnce.Do(func() { | ||
| p.quit <- struct{}{} | ||
| close(p.quit) | ||
| }) | ||
| } | ||
|
|
||
| func (p *MerkleRootPublisher) publish(ctx context.Context) error { | ||
| tables, err := p.store.FetchLatest(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("fetch unpublished metrics: %s", err) | ||
| } | ||
|
|
||
| if len(tables) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| for _, table := range tables { | ||
| chunks := chunker(table.Leaves, 32) | ||
|
|
||
| tree, err := merkletree.NewTree(chunks, nil) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if err := p.registry.Publish(ctx, table.TableID, table.HeightID, tree.MerkleRoot()); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| type TreeLeaves struct { | ||
| Leaves []byte | ||
| ChainID int64 | ||
| TableID int64 | ||
| HeightID int64 | ||
| TablePrefix string | ||
| } | ||
|
|
||
| // LeavesStore defines the API for fetching leaves from trees that need to be buil | ||
| type LeavesStore interface { | ||
| FetchLatest(context.Context) ([]TreeLeaves, error) | ||
| } | ||
|
|
||
| // MerkleRootRegistry defines the API for publishing root. | ||
| type MerkleRootRegistry interface { | ||
| Publish(context.Context, int64, int64, []byte) error | ||
| } | ||
|
|
||
| func chunker(data []byte, size int) [][]byte { | ||
| chunks := make([][]byte, len(data)/size) | ||
| for i := 0; i < len(data); i += size { | ||
| chunks[i/size] = data[i : i+size] | ||
| } | ||
|
|
||
| return chunks | ||
| } |
Oops, something went wrong.