-
-
Notifications
You must be signed in to change notification settings - Fork 50
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: create receiver command (#604)
* build of indexer * create receiver * indexer * fix build * mody tidy
- Loading branch information
Showing
36 changed files
with
1,002 additions
and
36 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
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,19 @@ | ||
| package commands | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/mailchain/mailchain/cmd/indexer/internal/processor" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| func doSequential(cmd *cobra.Command, p *processor.Sequential) { | ||
| for { | ||
| err := p.NextBlock(context.Background()) | ||
|
|
||
| if err != nil { | ||
| fmt.Fprintf(cmd.ErrOrStderr(), "%+v", err) | ||
| } | ||
| } | ||
| } |
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,117 @@ | ||
| package commands | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/jmoiron/sqlx" | ||
| "github.com/mailchain/mailchain/cmd/indexer/internal/processor" | ||
| sub "github.com/mailchain/mailchain/cmd/indexer/internal/substrate" | ||
| "github.com/mailchain/mailchain/cmd/internal/datastore/os" | ||
| "github.com/mailchain/mailchain/cmd/internal/datastore/pq" | ||
| "github.com/mailchain/mailchain/internal/protocols" | ||
| "github.com/mailchain/mailchain/internal/protocols/substrate" | ||
| "github.com/pkg/errors" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| func substrateCmd() *cobra.Command { | ||
| cmd := &cobra.Command{ | ||
| Use: "substrate", | ||
| Short: "run substrate sequential processor", | ||
| TraverseChildren: true, | ||
| RunE: func(cmd *cobra.Command, args []string) error { | ||
| network, _ := cmd.Flags().GetString("network") | ||
| protocol, _ := cmd.Flags().GetString("protocol") | ||
| blockNumber, _ := cmd.Flags().GetUint64("start-block") | ||
|
|
||
| addressRPC, _ := cmd.Flags().GetString("rpc-address") | ||
| if addressRPC == "" { | ||
| return errors.Errorf("rpc-address must not be empty") | ||
| } | ||
|
|
||
| rawStorePath, _ := cmd.Flags().GetString("raw-store-path") | ||
|
|
||
| connIndexer, err := newPostgresConnection(cmd, "indexer") | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| connPublicKey, err := newPostgresConnection(cmd, "pubkey") | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| connEnvelope, err := newPostgresConnection(cmd, "envelope") | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| defer connIndexer.Close() | ||
| defer connPublicKey.Close() | ||
| defer connEnvelope.Close() | ||
|
|
||
| seqProcessor, err := createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope, blockNumber, protocol, network, rawStorePath, addressRPC) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| doSequential(cmd, seqProcessor) | ||
|
|
||
| return nil | ||
| }, | ||
| } | ||
|
|
||
| cmd.Flags().Uint64("start-block", 0, "Block number from which the indexer will start") | ||
| cmd.Flags().String("protocol", protocols.Substrate, "Protocol to run against") | ||
| cmd.Flags().String("network", substrate.EdgewareMainnet, "Network to run against") | ||
| cmd.Flags().String("rpc-address", "", "Substrate RPC-JSON address") | ||
|
|
||
| return cmd | ||
| } | ||
|
|
||
| func createSubstrateProcessor(connIndexer, connPublicKey, connEnvelope *sqlx.DB, blockNumber uint64, protocol, network, rawStorePath, addressRPC string) (*processor.Sequential, error) { | ||
| ctx := context.Background() | ||
|
|
||
| subClient, err := sub.NewRPC(addressRPC) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| syncStore, err := pq.NewSyncStore(connIndexer) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| pubKeyStore, err := pq.NewPublicKeyStore(connPublicKey) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| transactionStore, err := pq.NewTransactionStore(connEnvelope) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| rawStore, err := os.NewRawTransactionStore(rawStorePath) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| processorTransaction := sub.NewExtrinsicProcessor( | ||
| transactionStore, | ||
| rawStore, | ||
| pubKeyStore, | ||
| ) | ||
|
|
||
| if err := syncStore.PutBlockNumber(ctx, protocol, network, blockNumber); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return processor.NewSequential( | ||
| protocols.Substrate, | ||
| network, | ||
| syncStore, | ||
| sub.NewBlockProcessor(processorTransaction), | ||
| subClient, | ||
| ), nil | ||
| } |
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,43 @@ | ||
| package substrate | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| gsrpc "github.com/centrifuge/go-substrate-rpc-client" | ||
| ) | ||
|
|
||
| func NewRPC(address string) (*BlockClient, error) { | ||
| api, err := gsrpc.NewSubstrateAPI(address) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &BlockClient{api: api}, nil | ||
| } | ||
|
|
||
| type BlockClient struct { | ||
| api *gsrpc.SubstrateAPI | ||
| } | ||
|
|
||
| func (c *BlockClient) BlockByNumber(ctx context.Context, blockNo uint64) (blk interface{}, err error) { | ||
| blkHash, err := c.api.RPC.Chain.GetBlockHash(blockNo) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| sb, err := c.api.RPC.Chain.GetBlock(blkHash) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &sb.Block, nil | ||
| } | ||
|
|
||
| func (c *BlockClient) LatestBlockNumber(ctx context.Context) (blockNo uint64, err error) { | ||
| signedBlock, err := c.api.RPC.Chain.GetBlockLatest() | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
|
|
||
| return uint64(signedBlock.Block.Header.Number), nil | ||
| } |
Oops, something went wrong.