forked from pactus-project/pactus
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(other): add zeromq server with configration (pactus-project#1660)
- Loading branch information
Showing
19 changed files
with
654 additions
and
0 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 |
|---|---|---|
|
|
@@ -60,6 +60,7 @@ jobs: | |
| http | ||
| jsonrpc | ||
| nanomsg | ||
| zeromq | ||
| windows | ||
| linux | ||
| macos | ||
|
|
||
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
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,26 @@ | ||
| package zmq | ||
|
|
||
| import ( | ||
| "github.com/go-zeromq/zmq4" | ||
| "github.com/pactus-project/pactus/types/block" | ||
| "github.com/pactus-project/pactus/util/logger" | ||
| ) | ||
|
|
||
| type blockInfoPub struct { | ||
| basePub | ||
| } | ||
|
|
||
| func newBlockInfoPub(socket zmq4.Socket, logger *logger.SubLogger) Publisher { | ||
| return &blockInfoPub{ | ||
| basePub: basePub{ | ||
| topic: BlockInfo, | ||
| zmqSocket: socket, | ||
| logger: logger, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func (*blockInfoPub) onNewBlock(_ *block.Block) { | ||
| // TODO implement me | ||
| panic("implement me") | ||
| } |
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,92 @@ | ||
| package zmq | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
| "net/url" | ||
| "strings" | ||
| "time" | ||
| ) | ||
|
|
||
| type Config struct { | ||
| ZmqPubBlockInfo string `toml:"zmqpubblockinfo"` | ||
| ZmqPubTxInfo string `toml:"zmqpubtxinfo"` | ||
| ZmqPubRawBlock string `toml:"zmqpubrawblock"` | ||
| ZmqPubRawTx string `toml:"zmqpubrawtx"` | ||
| ZmqPubHWM int `toml:"zmqpubhwm"` | ||
|
|
||
| // Private config | ||
| ZmqAutomaticReconnect bool `toml:"-"` | ||
| ZmqDialerRetryTime time.Duration `toml:"-"` | ||
| ZmqDialerMaxRetries int `toml:"-"` | ||
| } | ||
|
|
||
| func DefaultConfig() *Config { | ||
| return &Config{ | ||
| ZmqAutomaticReconnect: true, | ||
| ZmqDialerMaxRetries: 10, | ||
| ZmqDialerRetryTime: 250 * time.Millisecond, | ||
| ZmqPubHWM: 1000, | ||
| } | ||
| } | ||
|
|
||
| func (c *Config) BasicCheck() error { | ||
| if c.ZmqPubBlockInfo != "" { | ||
| if err := validateTopicSocket(c.ZmqPubBlockInfo); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| if c.ZmqPubTxInfo != "" { | ||
| if err := validateTopicSocket(c.ZmqPubTxInfo); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| if c.ZmqPubRawBlock != "" { | ||
| if err := validateTopicSocket(c.ZmqPubRawBlock); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| if c.ZmqPubRawTx != "" { | ||
| if err := validateTopicSocket(c.ZmqPubRawTx); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| if c.ZmqPubHWM < 0 { | ||
| return fmt.Errorf("invalid publisher hwm %d", c.ZmqPubHWM) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func validateTopicSocket(socket string) error { | ||
| addr, err := url.Parse(socket) | ||
| if err != nil { | ||
| return errors.New("failed to parse ZmqPub value: " + err.Error()) | ||
| } | ||
|
|
||
| if addr.Scheme != "tcp" { | ||
| return errors.New("invalid scheme: zeromq socket schema") | ||
| } | ||
|
|
||
| if addr.Host == "" { | ||
| return errors.New("invalid host: host is empty") | ||
| } | ||
|
|
||
| parts := strings.Split(addr.Host, ":") | ||
| if len(parts) != 2 || parts[0] == "" || parts[1] == "" { | ||
| return errors.New("invalid host: missing or malformed host/port") | ||
| } | ||
|
|
||
| port := parts[1] | ||
| for _, r := range port { | ||
| if r < '0' || r > '9' { | ||
| return errors.New("invalid port: non-numeric characters detected") | ||
| } | ||
| } | ||
|
|
||
| return 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package zmq | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestDefaultConfig(t *testing.T) { | ||
| cfg := DefaultConfig() | ||
|
|
||
| assert.NotNil(t, cfg, "DefaultConfig should not return nil") | ||
| assert.Equal(t, "", cfg.ZmqPubBlockInfo, "ZmqPubBlockInfo should be empty") | ||
| assert.Equal(t, "", cfg.ZmqPubTxInfo, "ZmqPubTxInfo should be empty") | ||
| assert.Equal(t, "", cfg.ZmqPubRawBlock, "ZmqPubRawBlock should be empty") | ||
| assert.Equal(t, "", cfg.ZmqPubRawTx, "ZmqPubRawTx should be empty") | ||
| assert.Equal(t, 1000, cfg.ZmqPubHWM, "ZmqPubHWM should default to 1000") | ||
| } | ||
|
|
||
| func TestBasicCheck(t *testing.T) { | ||
| testCases := []struct { | ||
| name string | ||
| config *Config | ||
| expectErr bool | ||
| }{ | ||
| { | ||
| name: "Valid configuration", | ||
| config: &Config{ | ||
| ZmqPubBlockInfo: "tcp://127.0.0.1:28332", | ||
| ZmqPubTxInfo: "tcp://127.0.0.1:28333", | ||
| ZmqPubRawBlock: "tcp://127.0.0.1:28334", | ||
| ZmqPubRawTx: "tcp://127.0.0.1:28335", | ||
| ZmqPubHWM: 1000, | ||
| }, | ||
| expectErr: false, | ||
| }, | ||
| { | ||
| name: "Invalid scheme", | ||
| config: &Config{ | ||
| ZmqPubBlockInfo: "udp://127.0.0.1:28332", | ||
| }, | ||
| expectErr: true, | ||
| }, | ||
| { | ||
| name: "Missing port", | ||
| config: &Config{ | ||
| ZmqPubBlockInfo: "tcp://127.0.0.1", | ||
| }, | ||
| expectErr: true, | ||
| }, | ||
| { | ||
| name: "Empty host", | ||
| config: &Config{ | ||
| ZmqPubBlockInfo: "tcp://:28332", | ||
| }, | ||
| expectErr: true, | ||
| }, | ||
| { | ||
| name: "Negative ZmqPubHWM", | ||
| config: &Config{ | ||
| ZmqPubHWM: -1, | ||
| }, | ||
| expectErr: true, | ||
| }, | ||
| { | ||
| name: "Empty configuration", | ||
| config: DefaultConfig(), | ||
| expectErr: false, | ||
| }, | ||
| } | ||
|
|
||
| for _, tc := range testCases { | ||
| t.Run(tc.name, func(t *testing.T) { | ||
| err := tc.config.BasicCheck() | ||
| if tc.expectErr { | ||
| assert.Error(t, err, "BasicCheck should return an error") | ||
| } else { | ||
| assert.NoError(t, err, "BasicCheck should not return an error") | ||
| } | ||
| }) | ||
| } | ||
| } |
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,28 @@ | ||
| package zmq | ||
|
|
||
| import ( | ||
| "github.com/go-zeromq/zmq4" | ||
| "github.com/pactus-project/pactus/types/block" | ||
| "github.com/pactus-project/pactus/util/logger" | ||
| ) | ||
|
|
||
| type Publisher interface { | ||
| Address() string | ||
| TopicName() string | ||
|
|
||
| onNewBlock(blk *block.Block) | ||
| } | ||
|
|
||
| type basePub struct { | ||
| topic Topic | ||
| zmqSocket zmq4.Socket | ||
| logger *logger.SubLogger | ||
| } | ||
|
|
||
| func (b *basePub) Address() string { | ||
| return b.zmqSocket.Addr().String() | ||
| } | ||
|
|
||
| func (b *basePub) TopicName() string { | ||
| return b.topic.String() | ||
| } |
Oops, something went wrong.