-
Notifications
You must be signed in to change notification settings - Fork 35
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* TaskGroup goroutine manager added in pkg/execution package. Tests on TaskGroup added. * Networking package with a new connection handler Session added. * Logger interface removed from networking package. Standard slog package is used instead. * WIP. Simple connection replaced with NetClient. NetClient usage moved into Universal client. Handshake proto updated to compatibility with Handshake interface from networking package. * Fixed NetClient closing issue. Configuration option to set KeepAliveInterval added to networking.Config. * Redundant log removed. * Move save int conversion to safecast lib. * Fix data race error in 'networking_test' package Implement 'io.Stringer' for 'Session' struct. Data race happens because 'clientHandler' mock in 'TestSessionTimeoutOnHandshake' test reads 'Session' structure at the same time as 'clientSession.Close' call. * Replace atomic.Uint32 with atomic.Bool and use CompareAndSwap there it's possible. Replace random delay with constan to make test not blink. Simplify assertion in test to make it stable. * Assertions added. Style fixed. * Simplified closing and close logic in NetClient. Added logs on handshake rejection to clarify the reason of rejections. Added and used function to configure Session with list of Slog attributes. * Prepare for new timer in Go 1.23 Co-authored-by: Nikolay Eskov <[email protected]> * Move constant into function were it used. Proper error declaration. * Better way to prevent from running multiple receiveLoops. Shutdown lock replaced with sync.Once. * Better data emptyness checks. * Better read error handling. Co-authored-by: Nikolay Eskov <[email protected]> * Use constructor. * Wrap heavy logging into log level checks. Fix data lock and data access order. * Session configuration accepts slog handler to set up logging. Discarding slog handler implemented and used instead of setting default slog logger. Checks on interval values added to Session constructor. * Close error channel on sending data successfully. Better error channel passing. Reset receiving buffer by deffering. * Better error handling while reading. Co-authored-by: Nikolay Eskov <[email protected]> * Fine error assertions. * Fix blinking test. * Better configuration handling. Co-authored-by: Nikolay Eskov <[email protected]> * Fixed blinking test TestCloseParentContext. Wait group added to wait for client to finish sending handshake. Better wait groups naming. * Better test workflow. Better wait group naming. * Fix deadlock in test by introducing wait group instead of sleep. * Internal sendPacket reimplemented using io.Reader. Data restoration function removed. Handler's OnReceive use io.Reader to pass received data. Tests updated. Mocks regenerated. * Itest network client handler updated. * Changed the way OnReceive passes the receiveBuffer. Test updated. --------- Co-authored-by: Nikolay Eskov <[email protected]>
- Loading branch information
1 parent
4559244
commit 14ae6e5
Showing
29 changed files
with
2,508 additions
and
235 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 |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| quiet: False | ||
| with-expecter: True | ||
| dir: "{{.InterfaceDir}}/mocks" | ||
| mockname: "Mock{{.InterfaceName}}" | ||
| filename: "{{.InterfaceName | snakecase}}.go" | ||
|
|
||
| packages: | ||
| github.com/wavesplatform/gowaves/pkg/networking: | ||
| interfaces: | ||
| Header: | ||
| Protocol: | ||
| Handler: |
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,218 @@ | ||
| package clients | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "encoding/base64" | ||
| "io" | ||
| "log/slog" | ||
| "net" | ||
| "sync" | ||
| "sync/atomic" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/neilotoole/slogt" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/wavesplatform/gowaves/itests/config" | ||
| "github.com/wavesplatform/gowaves/pkg/networking" | ||
| "github.com/wavesplatform/gowaves/pkg/proto" | ||
| ) | ||
|
|
||
| const ( | ||
| appName = "wavesL" | ||
| nonce = uint64(0) | ||
| networkTimeout = 3 * time.Second | ||
| pingInterval = 5 * time.Second | ||
| ) | ||
|
|
||
| type NetClient struct { | ||
| ctx context.Context | ||
| t testing.TB | ||
| impl Implementation | ||
| n *networking.Network | ||
| c *networking.Config | ||
| s *networking.Session | ||
|
|
||
| closing atomic.Bool | ||
| closed sync.Once | ||
| } | ||
|
|
||
| func NewNetClient( | ||
| ctx context.Context, t testing.TB, impl Implementation, port string, peers []proto.PeerInfo, | ||
| ) *NetClient { | ||
| n := networking.NewNetwork() | ||
| p := newProtocol(t, nil) | ||
| h := newHandler(t, peers) | ||
| log := slogt.New(t) | ||
| conf := networking.NewConfig(p, h). | ||
| WithSlogHandler(log.Handler()). | ||
| WithWriteTimeout(networkTimeout). | ||
| WithKeepAliveInterval(pingInterval). | ||
| WithSlogAttributes(slog.String("suite", t.Name()), slog.String("impl", impl.String())) | ||
|
|
||
| conn, err := net.Dial("tcp", config.DefaultIP+":"+port) | ||
| require.NoError(t, err, "failed to dial TCP to %s node", impl.String()) | ||
|
|
||
| s, err := n.NewSession(ctx, conn, conf) | ||
| require.NoError(t, err, "failed to establish new session to %s node", impl.String()) | ||
|
|
||
| cli := &NetClient{ctx: ctx, t: t, impl: impl, n: n, c: conf, s: s} | ||
| h.client = cli // Set client reference in handler. | ||
| return cli | ||
| } | ||
|
|
||
| func (c *NetClient) SendHandshake() { | ||
| handshake := &proto.Handshake{ | ||
| AppName: appName, | ||
| Version: proto.ProtocolVersion(), | ||
| NodeName: "itest", | ||
| NodeNonce: nonce, | ||
| DeclaredAddr: proto.HandshakeTCPAddr{}, | ||
| Timestamp: proto.NewTimestampFromTime(time.Now()), | ||
| } | ||
| buf := bytes.NewBuffer(nil) | ||
| _, err := handshake.WriteTo(buf) | ||
| require.NoError(c.t, err, | ||
| "failed to marshal handshake to %s node at %q", c.impl.String(), c.s.RemoteAddr()) | ||
| _, err = c.s.Write(buf.Bytes()) | ||
| require.NoError(c.t, err, | ||
| "failed to send handshake to %s node at %q", c.impl.String(), c.s.RemoteAddr()) | ||
| } | ||
|
|
||
| func (c *NetClient) SendMessage(m proto.Message) { | ||
| _, err := m.WriteTo(c.s) | ||
| require.NoError(c.t, err, "failed to send message to %s node at %q", c.impl.String(), c.s.RemoteAddr()) | ||
| } | ||
|
|
||
| func (c *NetClient) Close() { | ||
| c.closed.Do(func() { | ||
| if c.closing.CompareAndSwap(false, true) { | ||
| c.t.Logf("Closing connection to %s node at %q", c.impl.String(), c.s.RemoteAddr().String()) | ||
| } | ||
| err := c.s.Close() | ||
| require.NoError(c.t, err, "failed to close session to %s node at %q", c.impl.String(), c.s.RemoteAddr()) | ||
| }) | ||
| } | ||
|
|
||
| func (c *NetClient) reconnect() { | ||
| c.t.Logf("Reconnecting to %q", c.s.RemoteAddr().String()) | ||
| conn, err := net.Dial("tcp", c.s.RemoteAddr().String()) | ||
| require.NoError(c.t, err, "failed to dial TCP to %s node", c.impl.String()) | ||
|
|
||
| s, err := c.n.NewSession(c.ctx, conn, c.c) | ||
| require.NoError(c.t, err, "failed to re-establish the session to %s node", c.impl.String()) | ||
| c.s = s | ||
|
|
||
| c.SendHandshake() | ||
| } | ||
|
|
||
| type protocol struct { | ||
| t testing.TB | ||
| dropLock sync.Mutex | ||
| drop map[proto.PeerMessageID]struct{} | ||
| } | ||
|
|
||
| func newProtocol(t testing.TB, drop []proto.PeerMessageID) *protocol { | ||
| m := make(map[proto.PeerMessageID]struct{}) | ||
| for _, id := range drop { | ||
| m[id] = struct{}{} | ||
| } | ||
| return &protocol{t: t, drop: m} | ||
| } | ||
|
|
||
| func (p *protocol) EmptyHandshake() networking.Handshake { | ||
| return &proto.Handshake{} | ||
| } | ||
|
|
||
| func (p *protocol) EmptyHeader() networking.Header { | ||
| return &proto.Header{} | ||
| } | ||
|
|
||
| func (p *protocol) Ping() ([]byte, error) { | ||
| msg := &proto.GetPeersMessage{} | ||
| return msg.MarshalBinary() | ||
| } | ||
|
|
||
| func (p *protocol) IsAcceptableHandshake(h networking.Handshake) bool { | ||
| hs, ok := h.(*proto.Handshake) | ||
| if !ok { | ||
| return false | ||
| } | ||
| // Reject nodes with incorrect network bytes, unsupported protocol versions, | ||
| // or a zero nonce (indicating a self-connection). | ||
| if hs.AppName != appName || hs.Version.Cmp(proto.ProtocolVersion()) < 0 || hs.NodeNonce == 0 { | ||
| p.t.Logf("Unacceptable handshake:") | ||
| if hs.AppName != appName { | ||
| p.t.Logf("\tinvalid application name %q, expected %q", hs.AppName, appName) | ||
| } | ||
| if hs.Version.Cmp(proto.ProtocolVersion()) < 0 { | ||
| p.t.Logf("\tinvalid application version %q should be equal or more than %q", | ||
| hs.Version, proto.ProtocolVersion()) | ||
| } | ||
| if hs.NodeNonce == 0 { | ||
| p.t.Logf("\tinvalid node nonce %d", hs.NodeNonce) | ||
| } | ||
| return false | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| func (p *protocol) IsAcceptableMessage(h networking.Header) bool { | ||
| hdr, ok := h.(*proto.Header) | ||
| if !ok { | ||
| return false | ||
| } | ||
| p.dropLock.Lock() | ||
| defer p.dropLock.Unlock() | ||
| _, ok = p.drop[hdr.ContentID] | ||
| return !ok | ||
| } | ||
|
|
||
| type handler struct { | ||
| peers []proto.PeerInfo | ||
| t testing.TB | ||
| client *NetClient | ||
| } | ||
|
|
||
| func newHandler(t testing.TB, peers []proto.PeerInfo) *handler { | ||
| return &handler{t: t, peers: peers} | ||
| } | ||
|
|
||
| func (h *handler) OnReceive(s *networking.Session, r io.Reader) { | ||
| data, err := io.ReadAll(r) | ||
| if err != nil { | ||
| h.t.Logf("Failed to read message from %q: %v", s.RemoteAddr(), err) | ||
| h.t.FailNow() | ||
| return | ||
| } | ||
| msg, err := proto.UnmarshalMessage(data) | ||
| if err != nil { // Fail test on unmarshal error. | ||
| h.t.Logf("Failed to unmarshal message from bytes: %q", base64.StdEncoding.EncodeToString(data)) | ||
| h.t.FailNow() | ||
| return | ||
| } | ||
| switch msg.(type) { // Only reply with peers on GetPeersMessage. | ||
| case *proto.GetPeersMessage: | ||
| h.t.Logf("Received GetPeersMessage from %q", s.RemoteAddr()) | ||
| rpl := &proto.PeersMessage{Peers: h.peers} | ||
| if _, sErr := rpl.WriteTo(s); sErr != nil { | ||
| h.t.Logf("Failed to send peers message: %v", sErr) | ||
| h.t.FailNow() | ||
| return | ||
| } | ||
| default: | ||
| } | ||
| } | ||
|
|
||
| func (h *handler) OnHandshake(_ *networking.Session, _ networking.Handshake) { | ||
| h.t.Logf("Connection to %s node at %q was established", h.client.impl.String(), h.client.s.RemoteAddr()) | ||
| } | ||
|
|
||
| func (h *handler) OnClose(s *networking.Session) { | ||
| h.t.Logf("Connection to %q was closed", s.RemoteAddr()) | ||
| if !h.client.closing.Load() && h.client != nil { | ||
| h.client.reconnect() | ||
| } | ||
| } |
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
Oops, something went wrong.