mirror of
https://github.com/number571/go-peer.git
synced 2026-09-14 11:05:42 +05:00
500 lines
12 KiB
Go
500 lines
12 KiB
Go
package conn
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"math"
|
|
"net"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/number571/go-peer/pkg/crypto/random"
|
|
"github.com/number571/go-peer/pkg/crypto/scheme/layer1"
|
|
"github.com/number571/go-peer/pkg/encoding"
|
|
testutils "github.com/number571/go-peer/test/utils"
|
|
)
|
|
|
|
const (
|
|
tcMsgSize = (8 << 10)
|
|
tcWorkSize = 10
|
|
tcHead = 12345
|
|
tcBody = "hello, world!"
|
|
)
|
|
|
|
type tsConn struct {
|
|
readDlError bool
|
|
cancelBody bool
|
|
bodyPart bool
|
|
headSize uint32
|
|
bodySize uint64
|
|
}
|
|
|
|
func (p *tsConn) Read(b []byte) (n int, err error) {
|
|
if !p.bodyPart {
|
|
headBytes := encoding.Uint32ToBytes(p.headSize)
|
|
n = copy(b, headBytes[:])
|
|
p.bodyPart = true
|
|
return n, nil
|
|
}
|
|
if p.cancelBody {
|
|
return 0, errors.New("some error1") //nolint:err113
|
|
}
|
|
bodyBytes := random.NewRandom().GetBytes(p.bodySize)
|
|
n = copy(b, bodyBytes)
|
|
return n, nil
|
|
}
|
|
func (p *tsConn) Write(_ []byte) (n int, err error) {
|
|
return 0, errors.New("some error2") //nolint:err113
|
|
}
|
|
func (p *tsConn) Close() error { return nil }
|
|
func (p *tsConn) LocalAddr() net.Addr { return &net.TCPAddr{} }
|
|
func (p *tsConn) RemoteAddr() net.Addr { return &net.TCPAddr{} }
|
|
func (p *tsConn) SetDeadline(_ time.Time) error { return nil }
|
|
func (p *tsConn) SetReadDeadline(_ time.Time) error {
|
|
if p.bodyPart && p.readDlError {
|
|
return errors.New("some error3") //nolint:err113
|
|
}
|
|
return nil
|
|
}
|
|
func (p *tsConn) SetWriteDeadline(_ time.Time) error { return nil }
|
|
|
|
func TestError(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
str := "value"
|
|
err := &SConnError{str}
|
|
if err.Error() != errPrefix+str {
|
|
t.Fatal("incorrect err.Error()")
|
|
}
|
|
}
|
|
|
|
func TestSettings(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
for i := 0; i < 6; i++ {
|
|
testSettings(t, i)
|
|
}
|
|
}
|
|
|
|
func testSettings(t *testing.T, n int) {
|
|
defer func() {
|
|
if r := recover(); r == nil {
|
|
t.Fatal("nothing panics")
|
|
}
|
|
}()
|
|
switch n {
|
|
case 0:
|
|
_ = NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{}),
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
})
|
|
case 1:
|
|
_ = NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
})
|
|
case 2:
|
|
_ = NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
})
|
|
case 3:
|
|
_ = NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
})
|
|
case 4:
|
|
_ = NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
})
|
|
case 5:
|
|
_ = NewSettings(&SSettings{
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestClosedConn(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
listener := testNewService(t, testutils.TgAddrs[8], "")
|
|
defer testFreeService(listener)
|
|
|
|
conn, err := Connect(
|
|
context.Background(),
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
testutils.TgAddrs[8],
|
|
)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if err := conn.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
sett := layer1.NewConstructSettings(&layer1.SConstructSettings{
|
|
FSettings: conn.GetSettings().GetMessageSettings(),
|
|
})
|
|
|
|
pld := []byte("aaa")
|
|
msg := layer1.NewMessage(sett, pld)
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
if err := conn.WriteMessage(ctx, msg); err == nil {
|
|
t.Fatal("success write payload to closed connection")
|
|
}
|
|
|
|
readCh := make(chan struct{})
|
|
go func() { <-readCh }()
|
|
|
|
if _, err := conn.ReadMessage(ctx, readCh); err == nil {
|
|
t.Fatal("success read payload from closed connection")
|
|
}
|
|
|
|
sconn := conn.(*sConn)
|
|
if err := sconn.sendBytes(ctx, []byte("hello, world!")); err == nil {
|
|
t.Fatal("success send bytes to closed connection")
|
|
}
|
|
|
|
if _, err := sconn.recvDataBytes(ctx, 128, time.Second); err == nil {
|
|
t.Fatal("success recv data bytes from closed connection")
|
|
}
|
|
}
|
|
|
|
func TestInvalidConn(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
_, err := Connect(
|
|
context.Background(),
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
"INVALID_ADDRESS",
|
|
)
|
|
if err == nil {
|
|
t.Fatal("success connect to invalid address")
|
|
}
|
|
}
|
|
|
|
func TestReadMessage(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
rawConn := &tsConn{}
|
|
conn := LoadConn(
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
rawConn,
|
|
).(*sConn)
|
|
|
|
wg := sync.WaitGroup{}
|
|
wg.Add(1)
|
|
|
|
ch := make(chan struct{})
|
|
rawConn.bodyPart = false
|
|
rawConn.headSize = layer1.CMessageHeadSize + 10
|
|
rawConn.bodySize = layer1.CMessageHeadSize + 10
|
|
go func() {
|
|
defer wg.Done()
|
|
ctx := context.Background()
|
|
if _, err := conn.ReadMessage(ctx, ch); err == nil {
|
|
t.Error("success read message with invalid conn 1")
|
|
}
|
|
}()
|
|
<-ch
|
|
wg.Wait()
|
|
|
|
wg.Add(1)
|
|
rawConn.cancelBody = true
|
|
rawConn.bodyPart = false
|
|
rawConn.headSize = layer1.CMessageHeadSize + 10
|
|
rawConn.bodySize = layer1.CMessageHeadSize + 10
|
|
go func() {
|
|
defer wg.Done()
|
|
ctx := context.Background()
|
|
if _, err := conn.ReadMessage(ctx, ch); err == nil {
|
|
t.Error("success read message with invalid conn 2")
|
|
}
|
|
}()
|
|
<-ch
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestRecvDataBytes(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
rawConn := &tsConn{}
|
|
conn := LoadConn(
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
rawConn,
|
|
).(*sConn)
|
|
|
|
ch := make(chan struct{})
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
if _, err := conn.recvDataBytes(ctx, 1, 5*time.Second); err == nil {
|
|
t.Fatal("success recv data bytes with invalid conn 1")
|
|
}
|
|
|
|
rawConn.bodyPart = false
|
|
rawConn.headSize = layer1.CMessageHeadSize + 10
|
|
rawConn.bodySize = layer1.CMessageHeadSize + 10
|
|
rawConn.readDlError = true
|
|
go func() {
|
|
ctx := context.Background()
|
|
if _, err := conn.recvHeadBytes(ctx, ch, 5*time.Second); err == nil {
|
|
t.Error("success recv data bytes with invalid conn 2")
|
|
}
|
|
}()
|
|
<-ch
|
|
}
|
|
|
|
func TestSendBytes(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
rawConn := &tsConn{}
|
|
conn := LoadConn(
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
rawConn,
|
|
).(*sConn)
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
if err := conn.sendBytes(ctx, []byte{123}); err == nil {
|
|
t.Fatal("success send bytes with invalid conn 1")
|
|
}
|
|
|
|
if err := conn.sendBytes(context.Background(), []byte{123}); err == nil {
|
|
t.Fatal("success send bytes with invalid conn 2")
|
|
}
|
|
}
|
|
|
|
func TestRecvHeadBytes(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
rawConn := &tsConn{}
|
|
conn := LoadConn(
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
rawConn,
|
|
).(*sConn)
|
|
|
|
ch := make(chan struct{})
|
|
|
|
rawConn.bodyPart = false
|
|
rawConn.headSize = 1
|
|
go func() {
|
|
ctx := context.Background()
|
|
if _, err := conn.recvHeadBytes(ctx, ch, 5*time.Second); err == nil {
|
|
t.Error("success recv head bytes with invalid conn 1")
|
|
}
|
|
}()
|
|
<-ch
|
|
|
|
rawConn.bodyPart = false
|
|
rawConn.headSize = math.MaxUint32
|
|
go func() {
|
|
ctx := context.Background()
|
|
if _, err := conn.recvHeadBytes(ctx, ch, 5*time.Second); err == nil {
|
|
t.Error("success recv head bytes with invalid conn 2")
|
|
}
|
|
}()
|
|
<-ch
|
|
}
|
|
|
|
func TestConnWithNetworkKey(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
testConn(t, testutils.TgAddrs[9], "")
|
|
// testConn(t, testutils.TgAddrs[9], "hello, world!")
|
|
}
|
|
|
|
func testConn(t *testing.T, pAddr, pNetworkKey string) {
|
|
listener := testNewService(t, pAddr, pNetworkKey)
|
|
defer testFreeService(listener)
|
|
|
|
conn, err := Connect(
|
|
context.Background(),
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
pAddr,
|
|
)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
socket := conn.GetSocket()
|
|
remoteAddr := strings.ReplaceAll(pAddr, "localhost", "127.0.0.1")
|
|
if socket.RemoteAddr().String() != remoteAddr {
|
|
t.Fatal("got incorrect remote address")
|
|
}
|
|
|
|
msgSett := layer1.NewConstructSettings(&layer1.SConstructSettings{
|
|
FSettings: conn.GetSettings().GetMessageSettings(),
|
|
})
|
|
|
|
pld := []byte(tcBody)
|
|
msg := layer1.NewMessage(msgSett, pld)
|
|
ctx := context.Background()
|
|
if err := conn.WriteMessage(ctx, msg); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
readCh := make(chan struct{})
|
|
go func() { <-readCh }()
|
|
|
|
msgRecv, err := conn.ReadMessage(ctx, readCh)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if !bytes.Equal(msgRecv.GetBody(), []byte(tcBody)) {
|
|
t.Fatal("load payload not equal new payload")
|
|
}
|
|
}
|
|
|
|
func testNewService(t *testing.T, pAddr, pNetworkKey string) net.Listener {
|
|
listener, err := net.Listen("tcp", pAddr)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
go func() {
|
|
for {
|
|
aconn, err := listener.Accept()
|
|
if err != nil {
|
|
break
|
|
}
|
|
|
|
conn := LoadConn(
|
|
NewSettings(&SSettings{
|
|
FMessageSettings: layer1.NewSettings(&layer1.SSettings{
|
|
FWorkSizeBits: tcWorkSize,
|
|
FNetworkKey: pNetworkKey,
|
|
}),
|
|
FLimitMessageSizeBytes: tcMsgSize,
|
|
FWaitReadTimeout: time.Hour,
|
|
FDialTimeout: time.Minute,
|
|
FReadTimeout: time.Minute,
|
|
FWriteTimeout: time.Minute,
|
|
}),
|
|
aconn,
|
|
)
|
|
|
|
readCh := make(chan struct{})
|
|
go func() { <-readCh }()
|
|
|
|
ctx := context.Background()
|
|
|
|
msg, err := conn.ReadMessage(ctx, readCh)
|
|
if err != nil {
|
|
break
|
|
}
|
|
|
|
ok := func() bool {
|
|
defer func() { _ = conn.Close() }()
|
|
return conn.WriteMessage(ctx, msg) == nil
|
|
}()
|
|
|
|
if !ok {
|
|
break
|
|
}
|
|
}
|
|
}()
|
|
|
|
return listener
|
|
}
|
|
|
|
func testFreeService(listener net.Listener) {
|
|
if listener == nil {
|
|
return
|
|
}
|
|
_ = listener.Close()
|
|
}
|