Repository: gopcua/opcua
Pull request #146 (2019-03-14 11:27:09)

add support chunking for reading
This patch adds support for chunking when receiving messages.

wmnsk: The changes look good to me.
BTW, any reason naming the receiver(`Conn`) `a` ?
magiconair: I've started naming all receivers `a` recently since it makes renaming the struct easier and also reads somewhat nice. However, I don't have strong opinions about it.

Found this: https://github.com/golang/go/wiki/CodeReviewComments#receiver-names

I'll rename them.
magiconair: I'll squash and merge this but still need to add chunking for sending.

Patch:
diff --git a/examples/server/server.go b/examples/server/server.go
index 2b33597..96a70ea 100644
--- a/examples/server/server.go
+++ b/examples/server/server.go
@@ -15,12 +15,11 @@ import (
 func main() {
 	var (
 		endpoint = flag.String("endpoint", "opc.tcp://example.com/foo/bar", "OPC UA Endpoint URL")
-		bufsize  = flag.Int("bufsize", 0xffff, "Receive Buffer Size")
 	)
 	flag.Parse()
 
 	log.Printf("Listening on %s", *endpoint)
-	l, err := uacp.Listen(*endpoint, uint32(*bufsize))
+	l, err := uacp.Listen(*endpoint, nil)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/services/service.go b/services/service.go
index 82ef1e1..b18c2d1 100644
--- a/services/service.go
+++ b/services/service.go
@@ -66,14 +66,21 @@ func TypeID(v interface{}) uint16 {
 	return serviceTypeID[reflect.TypeOf(v)]
 }
 
-func Decode(typeID *datatypes.ExpandedNodeID, b []byte) (interface{}, error) {
+func Decode(b []byte) (*datatypes.ExpandedNodeID, interface{}, error) {
+	typeID := new(datatypes.ExpandedNodeID)
+	n, err := typeID.Decode(b)
+	if err != nil {
+		return nil, nil, err
+	}
+	b = b[n:]
+
 	id := uint16(typeID.NodeID.IntID())
 	typ := serviceType[id]
 	if typ == nil {
-		return nil, errors.NewErrUnsupported(id, "unsupported or not implemented yet.")
+		return nil, nil, errors.NewErrUnsupported(id, "unsupported or not implemented yet.")
 	}
 
 	v := reflect.New(typ.Elem()).Interface()
-	_, err := ua.Decode(b, v)
-	return v, err
+	_, err = ua.Decode(b, v)
+	return typeID, v, err
 }
diff --git a/uacp/conn.go b/uacp/conn.go
index d09c9a3..1279d61 100644
--- a/uacp/conn.go
+++ b/uacp/conn.go
@@ -13,6 +13,16 @@ import (
 	"github.com/gopcua/opcua/utils"
 )
 
+const (
+	KB = 1024
+	MB = 1024 * KB
+
+	DefaultReceiveBufSize = 0xffff
+	DefaultSendBufSize    = 0xffff
+	DefaultMaxChunkCount  = 512
+	DefaultMaxMessageSize = 2 * MB
+)
+
 // connid stores the current connection id. updated with atomic.AddUint32
 var connid uint32
 
@@ -33,10 +43,14 @@ func Dial(ctx context.Context, endpoint string) (*Conn, error) {
 	}
 
 	conn := &Conn{
-		id:         nextid(),
-		c:          c,
-		rcvBufSize: 0xffff,
-		sndBufSize: 0xffff,
+		id: nextid(),
+		c:  c,
+		ack: &Acknowledge{
+			ReceiveBufSize: DefaultReceiveBufSize,
+			SendBufSize:    DefaultSendBufSize,
+			MaxChunkCount:  0, // use what the server wants
+			MaxMessageSize: 0, // use what the server wants
+		},
 	}
 
 	log.Printf("conn %d: start HEL/ACK handshake", conn.id)
@@ -50,10 +64,9 @@ func Dial(ctx context.Context, endpoint string) (*Conn, error) {
 
 // Listener is a OPC UA Connection Protocol network listener.
 type Listener struct {
-	l          net.Listener
-	endpoint   string
-	rcvBufSize uint32
-	sndBufSize uint32
+	l        net.Listener
+	ack      *Acknowledge
+	endpoint string
 }
 
 // Listen acts like net.Listen for OPC UA Connection Protocol networks.
@@ -63,7 +76,16 @@ type Listener struct {
 // If the IP field of laddr is nil or an unspecified IP address, Listen listens
 // on all available unicast and anycast IP addresses of the local system.
 // If the Port field of laddr is 0, a port number is automatically chosen.
-func Listen(endpoint string, rcvBufSize uint32) (*Listener, error) {
+func Listen(endpoint string, ack *Acknowledge) (*Listener, error) {
+	if ack == nil {
+		ack = &Acknowledge{
+			ReceiveBufSize: DefaultReceiveBufSize,
+			SendBufSize:    DefaultSendBufSize,
+			MaxChunkCount:  DefaultMaxChunkCount,
+			MaxMessageSize: DefaultMaxMessageSize,
+		}
+	}
+
 	network, laddr, err := utils.ResolveEndpoint(endpoint)
 	if err != nil {
 		return nil, err
@@ -73,10 +95,9 @@ func Listen(endpoint string, rcvBufSize uint32) (*Listener, error) {
 		return nil, err
 	}
 	return &Listener{
-		l:          l,
-		endpoint:   endpoint,
-		rcvBufSize: rcvBufSize,
-		sndBufSize: 0xffff,
+		l:        l,
+		ack:      ack,
+		endpoint: endpoint,
 	}, nil
 }
 
@@ -89,12 +110,7 @@ func (l *Listener) Accept(ctx context.Context) (*Conn, error) {
 	if err != nil {
 		return nil, err
 	}
-	conn := &Conn{
-		id:         nextid(),
-		c:          c,
-		rcvBufSize: 0xffff,
-		sndBufSize: 0xffff,
-	}
+	conn := &Conn{nextid(), c, l.ack}
 	if err := conn.srvhandshake(l.endpoint); err != nil {
 		c.Close()
 		return nil, err
@@ -118,64 +134,79 @@ func (l *Listener) Endpoint() string {
 }
 
 type Conn struct {
-	id         uint32
-	c          net.Conn
-	rcvBufSize uint32
-	sndBufSize uint32
+	id  uint32
+	c   net.Conn
+	ack *Acknowledge
+}
+
+func (c *Conn) ID() uint32 {
+	return c.id
 }
 
-func (a *Conn) ID() uint32 {
-	return a.id
+func (c *Conn) ReceiveBufSize() uint32 {
+	return c.ack.ReceiveBufSize
 }
 
-func (a *Conn) Close() error {
-	log.Printf("conn %d: close", a.id)
-	return a.c.Close()
+func (c *Conn) SendBufSize() uint32 {
+	return c.ack.SendBufSize
 }
 
-func (a *Conn) Read(b []byte) (int, error) {
-	return a.c.Read(b)
+func (c *Conn) MaxMessageSize() uint32 {
+	return c.ack.MaxMessageSize
 }
 
-func (a *Conn) Write(b []byte) (int, error) {
-	return a.c.Write(b)
+func (c *Conn) MaxChunkCount() uint32 {
+	return c.ack.MaxChunkCount
 }
 
-func (a *Conn) SetDeadline(t time.Time) error {
-	return a.c.SetDeadline(t)
+func (c *Conn) Close() error {
+	log.Printf("conn %d: close", c.id)
+	return c.c.Close()
 }
 
-func (a *Conn) SetReadDeadline(t time.Time) error {
-	return a.c.SetReadDeadline(t)
+func (c *Conn) Read(b []byte) (int, error) {
+	return c.c.Read(b)
 }
 
-func (a *Conn) SetWriteDeadline(t time.Time) error {
-	return a.c.SetWriteDeadline(t)
+func (c *Conn) Write(b []byte) (int, error) {
+	return c.c.Write(b)
 }
 
-func (a *Conn) LocalAddr() net.Addr {
-	return a.c.LocalAddr()
+func (c *Conn) SetDeadline(t time.Time) error {
+	return c.c.SetDeadline(t)
 }
 
-func (a *Conn) RemoteAddr() net.Addr {
-	return a.c.RemoteAddr()
+func (c *Conn) SetReadDeadline(t time.Time) error {
+	return c.c.SetReadDeadline(t)
 }
 
-func (a *Conn) handshake(endpoint string) error {
-	m := &Hello{
-		Version:        0,
-		ReceiveBufSize: 0xffff,
-		SendBufSize:    0xffff,
-		MaxMessageSize: 0,
-		MaxChunkCount:  0,
+func (c *Conn) SetWriteDeadline(t time.Time) error {
+	return c.c.SetWriteDeadline(t)
+}
+
+func (c *Conn) LocalAddr() net.Addr {
+	return c.c.LocalAddr()
+}
+
+func (c *Conn) RemoteAddr() net.Addr {
+	return c.c.RemoteAddr()
+}
+
+func (c *Conn) handshake(endpoint string) error {
+	hel := &Hello{
+		Version:        c.ack.Version,
+		ReceiveBufSize: c.ack.ReceiveBufSize,
+		SendBufSize:    c.ack.SendBufSize,
+		MaxMessageSize: c.ack.MaxMessageSize,
+		MaxChunkCount:  c.ack.MaxChunkCount,
 		EndPointURL:    endpoint,
 	}
 
-	if err := a.send("HELF", m); err != nil {
+	if err := c.send("HELF", hel); err != nil {
 		return err
 	}
 
-	b, err := a.recv()
+	b, err := c.recv()
 	if err != nil {
 		return err
 	}
@@ -193,16 +224,23 @@ func (a *Conn) handshake(endpoint string) error {
 	if ack.Version != 0 {
 		return fmt.Errorf("invalid version %d", ack.Version)
 	}
-	a.rcvBufSize = ack.ReceiveBufSize
-	a.sndBufSize = ack.SendBufSize
-	log.Printf("conn %d: recv ACK:%v", a.id, ack)
+	if ack.MaxChunkCount == 0 {
+		ack.MaxChunkCount = DefaultMaxChunkCount
+		log.Printf("conn %d: server has no chunk limit. Using %d", c.id, ack.MaxChunkCount)
+	}
+	if ack.MaxMessageSize == 0 {
+		ack.MaxMessageSize = DefaultMaxMessageSize
+		log.Printf("conn %d: server has no message size limit. Using %d", c.id, ack.MaxMessageSize)
+	}
+	c.ack = ack
+	log.Printf("conn %d: recv ACK:%v", c.id, ack)
 	return nil
 }
 
-func (a *Conn) srvhandshake(endpoint string) error {
-	b, err := a.recv()
+func (c *Conn) srvhandshake(endpoint string) error {
+	b, err := c.recv()
 	if err != nil {
-		a.sendError(BadTCPInternalError)
+		c.sendError(BadTCPInternalError)
 		return err
 	}
 
@@ -213,22 +251,15 @@ func (a *Conn) srvhandshake(endpoint string) error {
 	case "HELF":
 		hel := new(Hello)
 		if _, err := ua.Decode(msg, hel); err != nil {
-			a.sendError(BadTCPInternalError)
+			c.sendError(BadTCPInternalError)
 			return err
 		}
 		if hel.EndPointURL != endpoint {
-			a.sendError(BadTCPEndpointURLInvalid)
+			c.sendError(BadTCPEndpointURLInvalid)
 			return fmt.Errorf("invalid endpoint url %s", hel.EndPointURL)
 		}
-		ack := &Acknowledge{
-			Version:        0,
-			ReceiveBufSize: a.rcvBufSize,
-			SendBufSize:    a.sndBufSize,
-			MaxMessageSize: 0, // what is a sensible default?
-			MaxChunkCount:  1000,
-		}
-		if err := a.send("ACKF", ack); err != nil {
-			a.sendError(BadTCPInternalError)
+		if err := c.send("ACKF", c.ack); err != nil {
+			c.sendError(BadTCPInternalError)
 			return err
 		}
 		return nil
@@ -236,42 +267,42 @@ func (a *Conn) srvhandshake(endpoint string) error {
 	case "RHEF":
 		rhe := new(ReverseHello)
 		if _, err := ua.Decode(msg, rhe); err != nil {
-			a.sendError(BadTCPInternalError)
+			c.sendError(BadTCPInternalError)
 			return err
 		}
 		if rhe.EndPointURL != endpoint {
-			a.sendError(BadTCPEndpointURLInvalid)
+			c.sendError(BadTCPEndpointURLInvalid)
 			return fmt.Errorf("invalid endpoint url %s", rhe.EndPointURL)
 		}
-		log.Printf("conn %d: connecting to %s", a.id, rhe.ServerURI)
-		a.c.Close()
+		log.Printf("conn %d: connecting to %s", c.id, rhe.ServerURI)
+		c.c.Close()
 		c, err := Dial(context.Background(), rhe.ServerURI)
 		if err != nil {
 			return err
 		}
-		a.c = c
+		c.c = c
 		return nil
 
 	default:
-		a.sendError(BadTCPInternalError)
+		c.sendError(BadTCPInternalError)
 		return fmt.Errorf("invalid handshake packet %q", msgtyp)
 	}
 }
 
-func (a *Conn) sendError(code uint32) {
+func (c *Conn) sendError(code uint32) {
 	// we swallow the error to silence complaints from the linter
 	// since sending an error will close the connection and we
 	// want to bubble a different error up.
-	_ = a.send("ERRF", &Error{Error: code})
+	_ = c.send("ERRF", &Error{Error: code})
 }
 
 // hdrlen is the size of the uacp header
 const hdrlen = 8
 
 // recv receives a message from the stream and returns it without the header.
-func (a *Conn) recv() ([]byte, error) {
+func (c *Conn) recv() ([]byte, error) {
 	hdr := make([]byte, hdrlen)
-	_, err := io.ReadFull(a.c, hdr)
+	_, err := io.ReadFull(c.c, hdr)
 	if err != nil {
 		return nil, fmt.Errorf("hdr read faled: %s", err)
 	}
@@ -281,20 +312,20 @@ func (a *Conn) recv() ([]byte, error) {
 		return nil, fmt.Errorf("hdr decode failed: %s", err)
 	}
 
-	if h.MessageSize > a.rcvBufSize {
-		return nil, fmt.Errorf("packet too large: %d > %d bytes", h.MessageSize, a.rcvBufSize)
+	if h.MessageSize > c.ack.ReceiveBufSize {
+		return nil, fmt.Errorf("packet too large: %d > %d bytes", h.MessageSize, c.ack.ReceiveBufSize)
 	}
 
 	b := make([]byte, h.MessageSize-hdrlen)
-	if _, err := io.ReadFull(a.c, b); err != nil {
+	if _, err := io.ReadFull(c.c, b); err != nil {
 		return nil, fmt.Errorf("read msg failed: %s", err)
 	}
 
-	log.Printf("conn %d: recv %s%c with %d bytes", a.id, h.MessageType, h.ChunkType, len(b))
+	log.Printf("conn %d: recv %s%c with %d bytes", c.id, h.MessageType, h.ChunkType, len(b))
 	return append(hdr, b...), nil
 }
 
-func (a *Conn) send(typ string, msg interface{}) error {
+func (c *Conn) send(typ string, msg interface{}) error {
 	if len(typ) != 4 {
 		return fmt.Errorf("invalid msg type: %s", typ)
 	}
@@ -310,8 +341,8 @@ func (a *Conn) send(typ string, msg interface{}) error {
 		MessageSize: uint32(len(body) + 8),
 	}
 
-	if h.MessageSize > a.sndBufSize {
-		return fmt.Errorf("send packet too large: %d > %d bytes", h.MessageSize, a.sndBufSize)
+	if h.MessageSize > c.ack.SendBufSize {
+		return fmt.Errorf("send packet too large: %d > %d bytes", h.MessageSize, c.ack.SendBufSize)
 	}
 
 	hdr, err := h.Encode()
@@ -320,10 +351,10 @@ func (a *Conn) send(typ string, msg interface{}) error {
 	}
 
 	b := append(hdr, body...)
-	if _, err := a.c.Write(b); err != nil {
+	if _, err := c.c.Write(b); err != nil {
 		return fmt.Errorf("write failed: %s", err)
 	}
-	log.Printf("conn %d: sent %s with %d bytes", a.id, typ, len(b))
+	log.Printf("conn %d: sent %s with %d bytes", c.id, typ, len(b))
 
 	return nil
 }
diff --git a/uasc/message.go b/uasc/message.go
index 75441d5..a407eb3 100644
--- a/uasc/message.go
+++ b/uasc/message.go
@@ -12,12 +12,56 @@ import (
 	"github.com/gopcua/opcua/ua"
 )
 
-// Message represents a OPC UA Secure Conversation message.
-type Message struct {
+type MessageHeader struct {
 	*Header
 	*AsymmetricSecurityHeader
 	*SymmetricSecurityHeader
 	*SequenceHeader
+}
+
+func (m *MessageHeader) Decode(b []byte) (int, error) {
+	buf := ua.NewBuffer(b)
+
+	m.Header = new(Header)
+	buf.ReadStruct(m.Header)
+
+	switch m.Header.MessageType {
+	case "OPN":
+		m.AsymmetricSecurityHeader = new(AsymmetricSecurityHeader)
+		buf.ReadStruct(m.AsymmetricSecurityHeader)
+
+	case "MSG", "CLO":
+		m.SymmetricSecurityHeader = new(SymmetricSecurityHeader)
+		buf.ReadStruct(m.SymmetricSecurityHeader)
+
+	default:
+		return buf.Pos(), errors.NewErrInvalidType(m, "decode", "should be one of OPN, MSG, CLO")
+	}
+
+	m.SequenceHeader = new(SequenceHeader)
+	buf.ReadStruct(m.SequenceHeader)
+
+	return buf.Pos(), buf.Error()
+}
+
+type MessageChunk struct {
+	*MessageHeader
+	Data []byte
+}
+
+func (m *MessageChunk) Decode(b []byte) (int, error) {
+	m.MessageHeader = new(MessageHeader)
+	n, err := m.MessageHeader.Decode(b)
+	if err != nil {
+		return n, err
+	}
+	m.Data = b[n:]
+	return len(b), nil
+}
+
+// Message represents a OPC UA Secure Conversation message.
+type Message struct {
+	*MessageHeader
 	TypeID  *datatypes.ExpandedNodeID
 	Service interface{}
 }
@@ -36,68 +80,47 @@ func NewMessage(srv interface{}, typeID uint16, cfg *Config) *Message {
 	switch typeID {
 	case id.OpenSecureChannelRequest_Encoding_DefaultBinary, id.OpenSecureChannelResponse_Encoding_DefaultBinary:
 		return &Message{
-			Header:                   NewHeader(MessageTypeOpenSecureChannel, ChunkTypeFinal, cfg.SecureChannelID),
-			AsymmetricSecurityHeader: NewAsymmetricSecurityHeader(cfg.SecurityPolicyURI, cfg.Certificate, cfg.Thumbprint),
-			SequenceHeader:           NewSequenceHeader(cfg.SequenceNumber, cfg.RequestID),
-			TypeID:                   datatypes.NewFourByteExpandedNodeID(0, typeID),
-			Service:                  srv,
+			MessageHeader: &MessageHeader{
+				Header:                   NewHeader(MessageTypeOpenSecureChannel, ChunkTypeFinal, cfg.SecureChannelID),
+				AsymmetricSecurityHeader: NewAsymmetricSecurityHeader(cfg.SecurityPolicyURI, cfg.Certificate, cfg.Thumbprint),
+				SequenceHeader:           NewSequenceHeader(cfg.SequenceNumber, cfg.RequestID),
+			},
+			TypeID:  datatypes.NewFourByteExpandedNodeID(0, typeID),
+			Service: srv,
 		}
 
 	case id.CloseSecureChannelRequest_Encoding_DefaultBinary, id.CloseSecureChannelResponse_Encoding_DefaultBinary:
 		return &Message{
-			Header:                  NewHeader(MessageTypeCloseSecureChannel, ChunkTypeFinal, cfg.SecureChannelID),
-			SymmetricSecurityHeader: NewSymmetricSecurityHeader(cfg.SecurityTokenID),
-			SequenceHeader:          NewSequenceHeader(cfg.SequenceNumber, cfg.RequestID),
-			TypeID:                  datatypes.NewFourByteExpandedNodeID(0, typeID),
-			Service:                 srv,
+			MessageHeader: &MessageHeader{
+				Header:                  NewHeader(MessageTypeCloseSecureChannel, ChunkTypeFinal, cfg.SecureChannelID),
+				SymmetricSecurityHeader: NewSymmetricSecurityHeader(cfg.SecurityTokenID),
+				SequenceHeader:          NewSequenceHeader(cfg.SequenceNumber, cfg.RequestID),
+			},
+			TypeID:  datatypes.NewFourByteExpandedNodeID(0, typeID),
+			Service: srv,
 		}
 
 	default:
 		return &Message{
-			Header:                  NewHeader(MessageTypeMessage, ChunkTypeFinal, cfg.SecureChannelID),
-			SymmetricSecurityHeader: NewSymmetricSecurityHeader(cfg.SecurityTokenID),
-			SequenceHeader:          NewSequenceHeader(cfg.SequenceNumber, cfg.RequestID),
-			TypeID:                  datatypes.NewFourByteExpandedNodeID(0, typeID),
-			Service:                 srv,
+			MessageHeader: &MessageHeader{
+				Header:                  NewHeader(MessageTypeMessage, ChunkTypeFinal, cfg.SecureChannelID),
+				SymmetricSecurityHeader: NewSymmetricSecurityHeader(cfg.SecurityTokenID),
+				SequenceHeader:          NewSequenceHeader(cfg.SequenceNumber, cfg.RequestID),
+			},
+			TypeID:  datatypes.NewFourByteExpandedNodeID(0, typeID),
+			Service: srv,
 		}
 	}
 }
 
 func (m *Message) Decode(b []byte) (int, error) {
-	buf := ua.NewBuffer(b)
-
-	m.Header = new(Header)
-	buf.ReadStruct(m.Header)
-
-	switch m.Header.MessageType {
-	case "OPN":
-		m.AsymmetricSecurityHeader = new(AsymmetricSecurityHeader)
-		buf.ReadStruct(m.AsymmetricSecurityHeader)
-
-	case "MSG", "CLO":
-		m.SymmetricSecurityHeader = new(SymmetricSecurityHeader)
-		buf.ReadStruct(m.SymmetricSecurityHeader)
-
-	default:
-		return buf.Pos(), errors.NewErrInvalidType(m, "decode", "should be one of OPN, MSG, CLO")
-	}
-
-	m.SequenceHeader = new(SequenceHeader)
-	buf.ReadStruct(m.SequenceHeader)
-
-	m.TypeID = new(datatypes.ExpandedNodeID)
-	buf.ReadStruct(m.TypeID)
-
-	if buf.Error() != nil {
-		return buf.Pos(), buf.Error()
-	}
-
-	svc, err := services.Decode(m.TypeID, buf.Bytes())
+	m.MessageHeader = new(MessageHeader)
+	n, err := m.MessageHeader.Decode(b)
 	if err != nil {
-		return 0, err
+		return n, err
 	}
-	m.Service = svc
-	return 0, nil
+	m.TypeID, m.Service, err = services.Decode(b[n:])
+	return len(b), err
 }
 
 func (m *Message) Encode() ([]byte, error) {
diff --git a/uasc/secure_channel.go b/uasc/secure_channel.go
index 12805cb..ff8826b 100644
--- a/uasc/secure_channel.go
+++ b/uasc/secure_channel.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	mrand "math/rand"
 	"reflect"
 	"sync"
 	"sync/atomic"
@@ -52,9 +53,13 @@ type SecureChannel struct {
 	handler map[uint32]chan Response
 }
 
+func init() {
+	mrand.Seed(time.Now().UnixNano())
+}
+
 func NewSecureChannel(c *uacp.Conn, cfg *Config) *SecureChannel {
 	if cfg == nil {
-		cfg = NewClientConfigSecurityNone(3333, 3600000)
+		cfg = NewClientConfigSecurityNone(uint32(mrand.Int31()), 3600000)
 	}
 
 	// always reset the secure channel id
@@ -152,8 +157,6 @@ func (s *SecureChannel) Send(svc interface{}, h func(interface{}) error) error {
 // SendAsync sends the service request and returns a channel which will receive the
 // response when it arrives.
 func (s *SecureChannel) SendAsync(svc interface{}) (resp chan Response, err error) {
-	log.Printf("conn %d: send %T", s.c.ID(), svc)
-
 	typeID := services.TypeID(svc)
 	if typeID == 0 {
 		return nil, fmt.Errorf("unknown service %T. Did you call register?", svc)
@@ -180,26 +183,27 @@ func (s *SecureChannel) SendAsync(svc interface{}) (resp chan Response, err erro
 	if err != nil {
 		return nil, err
 	}
+	reqid := m.SequenceHeader.RequestID
 
 	// send the message
 	if _, err := s.c.Write(b); err != nil {
 		return nil, err
 	}
-	log.Printf("conn %d: send %d bytes", s.c.ID(), len(b))
+	log.Printf("conn %d/%d: send %T with %d bytes", s.c.ID(), reqid, svc, len(b))
 
 	// register the handler
 	resp = make(chan Response)
 	s.mu.Lock()
-	s.handler[s.reqhdr.RequestHandle] = resp
+	s.handler[reqid] = resp
 	s.mu.Unlock()
 	return resp, nil
 }
 
-func (s *SecureChannel) recvsvc() (interface{}, error) {
+func (s *SecureChannel) readchunk() (*MessageChunk, error) {
+	// read and decode the header to get the message size
 	const hdrlen = 12
-
-	hdr := make([]byte, hdrlen)
-	_, err := io.ReadFull(s.c, hdr)
+	b := make([]byte, s.c.ReceiveBufSize())
+	_, err := io.ReadFull(s.c, b[:hdrlen])
 	if err == io.EOF {
 		return nil, err
 	}
@@ -211,49 +215,86 @@ func (s *SecureChannel) recvsvc() (interface{}, error) {
 	}
 
 	h := new(Header)
-	if _, err := h.Decode(hdr); err != nil {
+	if _, err := h.Decode(b[:hdrlen]); err != nil {
 		return nil, fmt.Errorf("sechan: decode header failed: %s", err)
 	}
+	b = b[:h.MessageSize]
 
+	// drop if the channel id does not match
 	if s.cfg.SecureChannelID > 0 && s.cfg.SecureChannelID != h.SecureChannelID {
 		return nil, fmt.Errorf("sechan: secure channel id mismatch: got 0x%04x, want 0x%04x", h.SecureChannelID, s.cfg.SecureChannelID)
 	}
 
-	b := make([]byte, h.MessageSize-hdrlen)
-	if _, err := io.ReadFull(s.c, b); err != nil {
+	// read the rest of the message
+	if _, err := io.ReadFull(s.c, b[hdrlen:]); err != nil {
 		return nil, fmt.Errorf("sechan: read message failed")
 	}
 
-	// todo(fs): handle ERR messages
-	m := new(Message)
-	if _, err := m.Decode(append(hdr, b...)); err != nil {
+	// decode the other headers
+	m := new(MessageChunk)
+	if _, err := m.Decode(b); err != nil {
 		return nil, fmt.Errorf("sechan: decode message failed: %s", err)
 	}
-	log.Printf("conn %d: recv %d bytes", s.c.ID(), len(hdr)+len(b))
+
+	// todo(fs): handle ERR messages
+	// todo(fs): handle crypto
 
 	if s.cfg.SecureChannelID == 0 {
 		s.cfg.SecureChannelID = h.SecureChannelID
-		log.Printf("conn %d: set secure channel id to %04x", s.c.ID(), s.cfg.SecureChannelID)
+		log.Printf("conn %d/%d: set secure channel id to %d", s.c.ID(), m.SequenceHeader.RequestID, s.cfg.SecureChannelID)
 	}
 
-	return m.Service, nil
+	return m, nil
 }
 
-// recv receives messages from the secure channel, decodes and forwards
+// recv receives message chunks from the secure channel, decodes and forwards
 // them to the registered callback channel, if there is one. Otherwise,
 // the message is dropped.
 func (s *SecureChannel) recv() {
+	// chunks maps request id to message chunks
+	chunks := map[uint32][]*MessageChunk{}
+
 	for {
 		select {
 		case <-s.quit:
 			return
 
 		default:
-			svc, err := s.recvsvc()
+			chunk, err := s.readchunk()
 			if err == io.EOF {
 				return
 			}
-			log.Printf("conn %d: recv %T", s.c.ID(), svc)
+
+			hdr := chunk.Header
+			reqid := chunk.SequenceHeader.RequestID
+			log.Printf("conn %d/%d: recv %s%c with %d bytes", s.c.ID(), reqid, hdr.MessageType, hdr.ChunkType, hdr.MessageSize)
+
+			if hdr.ChunkType != 'F' {
+				chunks[reqid] = append(chunks[reqid], chunk)
+				if n := len(chunks[reqid]); uint32(n) > s.c.MaxChunkCount() {
+					// todo(fs): send error
+					delete(chunks, reqid)
+					s.notifyCaller(reqid, nil, fmt.Errorf("too many chunks: %d > %d", n, s.c.MaxChunkCount()))
+				}
+				continue
+			}
+
+			// merge chunks
+			all := append(chunks[reqid], chunk)
+			delete(chunks, reqid)
+			b, err := mergeChunks(all)
+			if err != nil {
+				// todo(fs): send error
+				s.notifyCaller(reqid, nil, fmt.Errorf("chunk merge error: %v", err))
+				continue
+			}
+
+			if uint32(len(b)) > s.c.MaxMessageSize() {
+				// todo(fs): send error
+				s.notifyCaller(reqid, nil, fmt.Errorf("message too large: %d > %d", uint32(len(b)), s.c.MaxMessageSize()))
+				continue
+			}
+			// fmt.Println(utils.Wireshark(0, b))
 
 			// since we are not decoding the ResponseHeader separately
 			// we need to drop every message that has an error since we
@@ -262,32 +303,70 @@ func (s *SecureChannel) recv() {
 			// and subsequently remove it and the TypeID from all service
 			// structs and tests. We also need to add a deadline to all
 			// handlers and check them periodically to time them out.
+			_, svc, err := services.Decode(b)
 			if err != nil {
-				log.Printf("conn %d: recv %s", s.c.ID(), err)
+				s.notifyCaller(reqid, nil, err)
 				continue
 			}
+			s.notifyCaller(reqid, svc, err)
+		}
+	}
+}
 
-			// the response header is always the first field
-			val := reflect.ValueOf(svc)
-			resp := val.Elem().Field(0).Interface().(*services.ResponseHeader)
+func (s *SecureChannel) notifyCaller(reqid uint32, svc interface{}, err error) {
+	if err != nil {
+		log.Printf("conn %d/%d: %v", s.c.ID(), reqid, err)
+	} else {
+		log.Printf("conn %d/%d: recv %T", s.c.ID(), reqid, svc)
+	}
 
-			// check if we have a pending request handler for this response.
-			s.mu.Lock()
-			ch := s.handler[resp.RequestHandle]
-			delete(s.handler, resp.RequestHandle)
-			s.mu.Unlock()
+	// check if we have a pending request handler for this response.
+	s.mu.Lock()
+	ch := s.handler[reqid]
+	delete(s.handler, reqid)
+	s.mu.Unlock()
 
-			// no handler -> next response
-			if ch == nil {
-				log.Printf("%T: %s", svc, err)
-				continue
-			}
+	// no handler -> next response
+	if ch == nil {
+		log.Printf("conn %d/%d: no handler for %T", s.c.ID(), reqid, svc)
+		return
+	}
 
-			// send response to caller
-			go func() {
-				ch <- Response{svc, err}
-				close(ch)
-			}()
+	// send response to caller
+	go func() {
+		ch <- Response{svc, err}
+		close(ch)
+	}()
+}
+
+func mergeChunks(chunks []*MessageChunk) ([]byte, error) {
+	if len(chunks) == 0 {
+		return nil, nil
+	}
+	if len(chunks) == 1 {
+		return chunks[0].Data, nil
+	}
+
+	// todo(fs): check if this is correct and necessary
+	// sort.Sort(bySequence(chunks))
+
+	var b []byte
+	var seqnr uint32
+	for _, c := range chunks {
+		if c.SequenceHeader.SequenceNumber == seqnr {
+			continue // duplicate chunk
 		}
+		seqnr = c.SequenceHeader.SequenceNumber
+		b = append(b, c.Data...)
 	}
+	return b, nil
 }
+
+// todo(fs): we only need this if we need to sort chunks. Need to check the spec
+// type bySequence []*MessageChunk
+
+// func (a bySequence) Len() int      { return len(a) }
+// func (a bySequence) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
+// func (a bySequence) Less(i, j int) bool {
+// 	return a[i].SequenceHeader.SequenceNumber < a[j].SequenceHeader.SequenceNumber
+// }


Test patch:
diff --git a/uacp/conn_test.go b/uacp/conn_test.go
index e7fb65e..1693b19 100644
--- a/uacp/conn_test.go
+++ b/uacp/conn_test.go
@@ -14,7 +14,7 @@ import (
 
 func TestConn(t *testing.T) {
 	ep := "opc.tcp://127.0.0.1:4840/foo/bar"
-	ln, err := Listen(ep, 0xffff)
+	ln, err := Listen(ep, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -46,7 +46,7 @@ func TestConn(t *testing.T) {
 
 func TestClientWrite(t *testing.T) {
 	ep := "opc.tcp://127.0.0.1:4840/foo/bar"
-	ln, err := Listen(ep, 0xffff)
+	ln, err := Listen(ep, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -101,7 +101,7 @@ NEXT:
 
 func TestServerWrite(t *testing.T) {
 	ep := "opc.tcp://127.0.0.1:4840/foo/bar"
-	ln, err := Listen(ep, 0xffff)
+	ln, err := Listen(ep, nil)
 	if err != nil {
 		t.Fatal(err)
 	}