-
Notifications
You must be signed in to change notification settings - Fork 185
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge PR: push pendingtx to kafka (#942)
* push pendingtx to kafka * delete debug code * add some log * fix * update kafka msg type * change kafkaAddr Co-authored-by: Zhong Qiu <[email protected]> Co-authored-by: MengXiangJian <[email protected]>
- Loading branch information
3 people
authored
Aug 17, 2021
1 parent
2836a14
commit 2799e58
Showing
5 changed files
with
146 additions
and
6 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package pendingtx | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
|
|
||
| rpctypes "github.com/okex/exchain/app/rpc/types" | ||
| "github.com/segmentio/kafka-go" | ||
| ) | ||
|
|
||
| type KafkaClient struct { | ||
| Topic string | ||
| *kafka.Writer | ||
| } | ||
|
|
||
| func NewKafkaClient(addrs []string, topic string) *KafkaClient { | ||
| return &KafkaClient{ | ||
| Topic: topic, | ||
| Writer: kafka.NewWriter(kafka.WriterConfig{ | ||
| Brokers: addrs, | ||
| Topic: topic, | ||
| Balancer: &kafka.LeastBytes{}, | ||
| }), | ||
| } | ||
| } | ||
|
|
||
| type KafkaMsg struct { | ||
| Topic string `json:"topic"` | ||
| Source interface{} `json:"source"` | ||
| Data *rpctypes.Transaction `json:"data"` | ||
| } | ||
|
|
||
| func (kc *KafkaClient) Send(hash []byte, tx *rpctypes.Transaction) error { | ||
| msg, err := json.Marshal(KafkaMsg{ | ||
| Topic: kc.Topic, | ||
| Data: tx, | ||
| }) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // Automatic retries and reconnections on errors. | ||
| return kc.WriteMessages(context.Background(), | ||
| kafka.Message{ | ||
| Key: hash, | ||
| Value: msg, | ||
| }, | ||
| ) | ||
| } |
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,77 @@ | ||
| package pendingtx | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/cosmos/cosmos-sdk/client/context" | ||
| "github.com/ethereum/go-ethereum/common" | ||
| rpcfilters "github.com/okex/exchain/app/rpc/namespaces/eth/filters" | ||
| rpctypes "github.com/okex/exchain/app/rpc/types" | ||
| "github.com/tendermint/tendermint/libs/log" | ||
| coretypes "github.com/tendermint/tendermint/rpc/core/types" | ||
| tmtypes "github.com/tendermint/tendermint/types" | ||
| ) | ||
|
|
||
| type Watcher struct { | ||
| clientCtx context.CLIContext | ||
| events *rpcfilters.EventSystem | ||
| logger log.Logger | ||
|
|
||
| sender Sender | ||
| } | ||
|
|
||
| type Sender interface { | ||
| Send(hash []byte, tx *rpctypes.Transaction) error | ||
| } | ||
|
|
||
| func NewWatcher(clientCtx context.CLIContext, log log.Logger, sender Sender) *Watcher { | ||
| return &Watcher{ | ||
| clientCtx: clientCtx, | ||
| events: rpcfilters.NewEventSystem(clientCtx.Client), | ||
| logger: log.With("module", "pendingtx-watcher"), | ||
|
|
||
| sender: sender, | ||
| } | ||
| } | ||
|
|
||
| func (w *Watcher) Start() { | ||
| sub, _, err := w.events.SubscribePendingTxs() | ||
| if err != nil { | ||
| w.logger.Error("error creating block filter", "error", err.Error()) | ||
| } | ||
|
|
||
| go func(txsCh <-chan coretypes.ResultEvent, errCh <-chan error) { | ||
| for { | ||
| select { | ||
| case ev := <-txsCh: | ||
| data, ok := ev.Data.(tmtypes.EventDataTx) | ||
| if !ok { | ||
| w.logger.Error(fmt.Sprintf("invalid data type %T, expected EventDataTx", ev.Data), "ID", sub.ID()) | ||
| continue | ||
| } | ||
| txHash := common.BytesToHash(data.Tx.Hash()) | ||
| w.logger.Debug("receive tx from mempool", "txHash=", txHash.String()) | ||
|
|
||
| ethTx, err := rpctypes.RawTxToEthTx(w.clientCtx, data.Tx) | ||
| if err != nil { | ||
| w.logger.Error("failed to decode raw tx to eth tx", "hash", txHash.String(), "error", err) | ||
| continue | ||
| } | ||
|
|
||
| tx, err := rpctypes.NewTransaction(ethTx, txHash, common.Hash{}, uint64(data.Height), uint64(data.Index)) | ||
| if err != nil { | ||
| w.logger.Error("failed to new transaction", "hash", txHash.String(), "error", err) | ||
| continue | ||
| } | ||
|
|
||
| go func(hash []byte, tx *rpctypes.Transaction) { | ||
| w.logger.Debug("push pending tx to MQ", "txHash=", txHash.String()) | ||
| err = w.sender.Send(hash, tx) | ||
| if err != nil { | ||
| w.logger.Error("failed to send pending tx", "hash", txHash.String(), "error", err) | ||
| } | ||
| }(txHash.Bytes(), tx) | ||
| } | ||
| } | ||
| }(sub.Event(), sub.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