feat: add message hello and ack

.typos.toml

@@ -0,0 +1,3 @@
+[default.extend-words]
+# keyword fp
+"hel" = "hel"

examples/client/client.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"github.com/shoothzj/gox/netx"
+	"log/slog"
+	"opcua-go/opcua"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	logger := slog.Default()
+
+	config := &opcua.ClientConfig{
+		Address: netx.Address{
+			Host: "127.0.0.1",
+			Port: 4840,
+		},
+		Logger: logger,
+	}
+
+	client, err := opcua.NewClient(config)
+	if err != nil {
+		logger.Error("Failed to start client", slog.String("error", err.Error()))
+		os.Exit(1)
+	}
+
+	messageAcknowledge, err := client.Hello(&opcua.MessageHello{
+		Version:           0,
+		ReceiveBufferSize: 65535,
+		SendBufferSize:    65535,
+		MaxMessageSize:    2097152,
+		MaxChunkCount:     0,
+		EndpointUrl:       "opc.tcp://localhost:4840/opcua",
+	})
+
+	if err != nil {
+		logger.Error("Failed to hello", slog.String("error", err.Error()))
+		os.Exit(1)
+	}
+
+	logger.Info("Hello response", slog.Any("response", messageAcknowledge))
+
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
+
+	<-signalChan
+	logger.Info("Shutting down server...")
+
+	client.Close()
+	logger.Info("Client closed gracefully")
+}

go.mod

@@ -3,7 +3,7 @@ module opcua-go
 go 1.21
 
 require (
-	github.com/shoothzj/gox v0.0.3-0.20240915030409-0ff3ac4e657c
+	github.com/shoothzj/gox v0.0.4-0.20240916070027-2e4a2098d749
 	github.com/stretchr/testify v1.9.0
 )
 

go.sum

@@ -2,8 +2,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/shoothzj/gox v0.0.3-0.20240915030409-0ff3ac4e657c h1:bGa8KCyniEJwV1IF6BKSwjKlexk5If8cXp3M17JOfa4=
-github.com/shoothzj/gox v0.0.3-0.20240915030409-0ff3ac4e657c/go.mod h1:W8vthsaC2LWvu1zy/B9znKuOOSrFUqWCMHkdbqZJE04=
+github.com/shoothzj/gox v0.0.4-0.20240916070027-2e4a2098d749 h1:yytTUCM7e17ADriGwcEaWdQ63ocmsdDGH+Qj979gNF8=
+github.com/shoothzj/gox v0.0.4-0.20240916070027-2e4a2098d749/go.mod h1:JrD7u6Tem/9Ut4o0kQNK+aZkMWCba/d9VeabBZSEDT4=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=

opcua/client.go

@@ -2,6 +2,7 @@ package opcua
 
 import (
 	"crypto/tls"
+	"encoding/binary"
 	"fmt"
 	"github.com/shoothzj/gox/buffer"
 	"github.com/shoothzj/gox/netx"
@@ -21,8 +22,8 @@ type ClientConfig struct {
 }
 
 type sendRequest struct {
-	bytes    []byte
-	callback func([]byte, error)
+	buf      *buffer.Buffer
+	callback func(*buffer.Buffer, error)
 }
 
 type Client struct {
@@ -36,12 +37,24 @@ type Client struct {
 	closeCh      chan struct{}
 }
 
-func (c *Client) Send(bytes []byte) ([]byte, error) {
+func (c *Client) Hello(message *MessageHello) (*MessageAcknowledge, error) {
+	buf, err := message.Buffer()
+	if err != nil {
+		return nil, err
+	}
+	bufResp, err := c.Send(buf)
+	if err != nil {
+		return nil, err
+	}
+	return DecodeMessageAcknowledge(bufResp)
+}
+
+func (c *Client) Send(buf *buffer.Buffer) (*buffer.Buffer, error) {
 	wg := sync.WaitGroup{}
 	wg.Add(1)
-	var result []byte
+	var result *buffer.Buffer
 	var err error
-	c.sendAsync(bytes, func(resp []byte, e error) {
+	c.sendAsync(buf, func(resp *buffer.Buffer, e error) {
 		result = resp
 		err = e
 		wg.Done()
@@ -53,9 +66,9 @@ func (c *Client) Send(bytes []byte) ([]byte, error) {
 	return result, nil
 }
 
-func (c *Client) sendAsync(bytes []byte, callback func([]byte, error)) {
+func (c *Client) sendAsync(buf *buffer.Buffer, callback func(*buffer.Buffer, error)) {
 	sr := &sendRequest{
-		bytes:    bytes,
+		buf:      buf,
 		callback: callback,
 	}
 	c.eventsChan <- sr
@@ -77,19 +90,18 @@ func (c *Client) read() {
 				c.closeCh <- struct{}{}
 				break
 			}
-			if c.buffer.Size() < 4 {
+			if c.buffer.ReadableSize() < 8 {
 				continue
 			}
-			bytes := make([]byte, 4)
-			err = c.buffer.ReadExactly(bytes)
-			c.buffer.Compact()
+			bytes := make([]byte, 8)
+			err = c.buffer.PeekExactly(bytes)
 			if err != nil {
 				req.callback(nil, err)
 				c.closeCh <- struct{}{}
 				break
 			}
-			length := int(bytes[3]) | int(bytes[2])<<8 | int(bytes[1])<<16 | int(bytes[0])<<24
-			if c.buffer.Size() < length {
+			length := int(binary.LittleEndian.Uint32(bytes[4:8]))
+			if c.buffer.ReadableSize() < length {
 				continue
 			}
 			// in case ddos attack
@@ -106,7 +118,7 @@ func (c *Client) read() {
 				break
 			}
 			c.buffer.Compact()
-			req.callback(data, nil)
+			req.callback(buffer.NewBufferFromBytes(data), nil)
 		case <-c.closeCh:
 			return
 		}
@@ -117,14 +129,20 @@ func (c *Client) write() {
 	for {
 		select {
 		case req := <-c.eventsChan:
-			n, err := c.conn.Write(req.bytes)
+			bytes, err := req.buf.ReadNBytes(req.buf.ReadableSize())
+			if err != nil {
+				req.callback(nil, err)
+				c.closeCh <- struct{}{}
+				break
+			}
+			n, err := c.conn.Write(bytes)
 			if err != nil {
 				req.callback(nil, err)
 				c.closeCh <- struct{}{}
 				break
 			}
-			if n != len(req.bytes) {
-				req.callback(nil, fmt.Errorf("write %d bytes, but expect %d bytes", n, len(req.bytes)))
+			if n != len(bytes) {
+				req.callback(nil, fmt.Errorf("write %d bytes, but expect %d bytes", n, len(bytes)))
 				c.closeCh <- struct{}{}
 				break
 			}

opcua/const.go

@@ -0,0 +1,13 @@
+package opcua
+
+const (
+	LenChunkType         = 1
+	LenMaxChunkCount     = 4
+	LenMaxMessageSize    = 4
+	LenMessageSize       = 4
+	LenMessageType       = 3
+	LenReceiveBufferSize = 4
+	LenSendBufferSize    = 4
+	LenStringLen         = 4
+	LenVersion           = 4
+)

opcua/message_acknowledge.go

@@ -0,0 +1,78 @@
+package opcua
+
+import "github.com/shoothzj/gox/buffer"
+
+type MessageAcknowledge struct {
+	Version           uint32
+	ReceiveBufferSize uint32
+	SendBufferSize    uint32
+	MaxMessageSize    uint32
+	MaxChunkCount     uint32
+}
+
+func DecodeMessageAcknowledge(buf *buffer.Buffer) (resp *MessageAcknowledge, err error) {
+	resp = &MessageAcknowledge{}
+	resp.Version, err = buf.ReadUInt32Le()
+	if err != nil {
+		return nil, err
+	}
+	resp.ReceiveBufferSize, err = buf.ReadUInt32Le()
+	if err != nil {
+		return nil, err
+	}
+	resp.SendBufferSize, err = buf.ReadUInt32Le()
+	if err != nil {
+		return nil, err
+	}
+	resp.MaxMessageSize, err = buf.ReadUInt32Le()
+	if err != nil {
+		return nil, err
+	}
+	resp.MaxChunkCount, err = buf.ReadUInt32Le()
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
+func (m *MessageAcknowledge) Length() int {
+	length := 0
+	length += LenMessageType
+	length += LenChunkType
+	length += LenMessageSize
+	length += LenVersion
+	length += LenReceiveBufferSize
+	length += LenSendBufferSize
+	length += LenMaxMessageSize
+	length += LenMaxChunkCount
+	return length
+}
+
+func (m *MessageAcknowledge) Buffer() (*buffer.Buffer, error) {
+	buf := buffer.NewBuffer(m.Length())
+	if _, err := buf.Write([]byte{'A', 'C', 'K'}); err != nil {
+		return nil, err
+	}
+	if _, err := buf.Write([]byte{'F'}); err != nil {
+		return nil, err
+	}
+	if err := buf.PutUInt32Le(uint32(m.Length())); err != nil {
+		return nil, err
+	}
+	if err := buf.PutUInt32Le(m.Version); err != nil {
+		return nil, err
+	}
+	if err := buf.PutUInt32Le(m.ReceiveBufferSize); err != nil {
+		return nil, err
+	}
+	if err := buf.PutUInt32Le(m.SendBufferSize); err != nil {
+		return nil, err
+	}
+	if err := buf.PutUInt32Le(m.MaxMessageSize); err != nil {
+		return nil, err
+	}
+	if err := buf.PutUInt32Le(m.MaxChunkCount); err != nil {
+		return nil, err
+	}
+	return buf, nil
+}

opcua/message_acknowledge_test.go

@@ -0,0 +1,35 @@
+package opcua
+
+import (
+	"github.com/shoothzj/gox/testx"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestDecodeMessageAcknowledge(t *testing.T) {
+	buffer := testx.Hex2Buffer(t, "41434b461c00000000000000ffff0000ffff00000000200040000000")
+	err := buffer.Skip(8)
+	require.Nil(t, err)
+	msg, err := DecodeMessageAcknowledge(buffer)
+	require.Nil(t, err)
+	require.NotNil(t, msg)
+	assert.Equal(t, uint32(0), msg.Version)
+	assert.Equal(t, uint32(65535), msg.ReceiveBufferSize)
+	assert.Equal(t, uint32(65535), msg.SendBufferSize)
+	assert.Equal(t, uint32(2097152), msg.MaxMessageSize)
+	assert.Equal(t, uint32(64), msg.MaxChunkCount)
+}
+
+func TestEncodeMessageAcknowledge(t *testing.T) {
+	msg := &MessageAcknowledge{
+		Version:           0,
+		ReceiveBufferSize: 65535,
+		SendBufferSize:    65535,
+		MaxMessageSize:    2097152,
+		MaxChunkCount:     64,
+	}
+	buffer, err := msg.Buffer()
+	require.Nil(t, err)
+	assert.Equal(t, testx.Hex2Buffer(t, "41434b461c00000000000000ffff0000ffff00000000200040000000"), buffer)
+}