Repository: gopcua/opcua
Pull request #482 (2021-10-11 12:38:08)

Release 0.2

magiconair: We will release this on Monday so that we can move forward. 

Patch:
diff --git a/client.go b/client.go
index 77286b1..068501c 100644
--- a/client.go
+++ b/client.go
@@ -691,12 +691,15 @@ func (c *Client) ActivateSession(s *Session) error {
 		// save the nonce for the next request
 		s.serverNonce = res.ServerNonce
 
-		if err := c.CloseSession(); err != nil {
-			// try to close the newly created session but report
-			// only the initial error.
-			_ = c.closeSession(s)
-			return err
-		}
+		// close the previous session
+		//
+		// https://github.com/gopcua/opcua/issues/474
+		//
+		// We decided not to check the error of CloseSession() since we
+		// can't do much about it anyway and it creates a race in the
+		// re-connection logic.
+		c.CloseSession()
+
 		c.session.Store(s)
 		return nil
 	})
@@ -799,7 +802,26 @@ func (c *Client) Read(req *ua.ReadRequest) (*ua.ReadResponse, error) {
 
 	var res *ua.ReadResponse
 	err := c.Send(req, func(v interface{}) error {
-		return safeAssign(v, &res)
+		err := safeAssign(v, &res)
+
+		// If the client cannot decode an extension object then its
+		// value will be nil. However, since the EO was known to the
+		// server the StatusCode for that data value will be OK. We
+		// therefore check for extension objects with nil values and set
+		// the status code to StatusBadDataTypeIDUnknown.
+		if err == nil {
+			for _, dv := range res.Results {
+				if dv.Value == nil {
+					continue
+				}
+				val := dv.Value.Value()
+				if eo, ok := val.(*ua.ExtensionObject); ok && eo.Value == nil {
+					dv.Status = ua.StatusBadDataTypeIDUnknown
+				}
+			}
+		}
+
+		return err
 	})
 	return res, err
 }
diff --git a/client_sub.go b/client_sub.go
index 74e5a18..1818d89 100644
--- a/client_sub.go
+++ b/client_sub.go
@@ -50,6 +50,7 @@ func (c *Client) Subscribe(params *SubscriptionParameters, notifyCh chan *Publis
 		RevisedLifetimeCount:      res.RevisedLifetimeCount,
 		RevisedMaxKeepAliveCount:  res.RevisedMaxKeepAliveCount,
 		Notifs:                    notifyCh,
+		items:                     make(map[uint32]*monitoredItem),
 		params:                    params,
 		nextSeq:                   1,
 		c:                         c,
diff --git a/subscription.go b/subscription.go
index 39c6716..12e30fa 100644
--- a/subscription.go
+++ b/subscription.go
@@ -29,7 +29,7 @@ type Subscription struct {
 	RevisedMaxKeepAliveCount  uint32
 	Notifs                    chan *PublishNotificationData
 	params                    *SubscriptionParameters
-	items                     []*monitoredItem
+	items                     map[uint32]*monitoredItem
 	lastSeq                   uint32
 	nextSeq                   uint32
 	c                         *Client
@@ -44,11 +44,9 @@ type SubscriptionParameters struct {
 }
 
 type monitoredItem struct {
-	ItemToMonitor        *ua.ReadValueID
-	MonitoringParameters *ua.MonitoringParameters
-	MonitoringMode       ua.MonitoringMode
-	TimestampsToReturn   ua.TimestampsToReturn
-	createResult         *ua.MonitoredItemCreateResult
+	item *ua.MonitoredItemCreateRequest
+	res  *ua.MonitoredItemCreateResult
+	ts   ua.TimestampsToReturn
 }
 
 func NewMonitoredItemCreateRequestWithDefaults(nodeID *ua.NodeID, attributeID ua.AttributeID, clientHandle uint32) *ua.MonitoredItemCreateRequest {
@@ -96,6 +94,7 @@ func (s *Subscription) delete() error {
 	})
 	switch {
 	case err == ua.StatusOK:
+		s.items = make(map[uint32]*monitoredItem)
 		return nil
 	case err != nil:
 		return err
@@ -128,17 +127,15 @@ func (s *Subscription) Monitor(ts ua.TimestampsToReturn, items ...*ua.MonitoredI
 	}
 
 	// store monitored items
+	// todo(fs): should we guard this with a lock?
 	for i, item := range items {
 		result := res.Results[i]
 
-		mi := &monitoredItem{
-			ItemToMonitor:        item.ItemToMonitor,
-			MonitoringParameters: item.RequestedParameters,
-			MonitoringMode:       item.MonitoringMode,
-			TimestampsToReturn:   ts,
-			createResult:         result,
+		s.items[result.MonitoredItemID] = &monitoredItem{
+			item: item,
+			res:  result,
+			ts:   ts,
 		}
-		s.items = append(s.items, mi)
 	}
 
 	return res, err
@@ -149,10 +146,20 @@ func (s *Subscription) Unmonitor(monitoredItemIDs ...uint32) (*ua.DeleteMonitore
 		MonitoredItemIDs: monitoredItemIDs,
 		SubscriptionID:   s.SubscriptionID,
 	}
+
 	var res *ua.DeleteMonitoredItemsResponse
 	err := s.c.Send(req, func(v interface{}) error {
 		return safeAssign(v, &res)
 	})
+
+	if err == nil {
+		// remove monitored items
+		// todo(fs): should we guard this with a lock?
+		for _, id := range monitoredItemIDs {
+			delete(s.items, id)
+		}
+	}
+
 	return res, err
 }
 
@@ -298,17 +305,13 @@ func (s *Subscription) recreate() error {
 	dlog.Printf("subscription registered")
 
 	// Sort by timestamp to return
-	itemsByTs := make(map[ua.TimestampsToReturn][]*ua.MonitoredItemCreateRequest)
-	for _, m := range s.items {
-		cr := &ua.MonitoredItemCreateRequest{
-			ItemToMonitor:       m.ItemToMonitor,
-			MonitoringMode:      m.MonitoringMode,
-			RequestedParameters: m.MonitoringParameters,
-		}
-		itemsByTs[m.TimestampsToReturn] = append(itemsByTs[m.TimestampsToReturn], cr)
+	itemsByTimestamps := make(map[ua.TimestampsToReturn][]*ua.MonitoredItemCreateRequest)
+	for _, mi := range s.items {
+		itemsByTimestamps[mi.ts] = append(itemsByTimestamps[mi.ts], mi.item)
 	}
 
-	for ts, items := range itemsByTs {
+	s.items = make(map[uint32]*monitoredItem, len(s.items))
+	for ts, items := range itemsByTimestamps {
 		req := &ua.CreateMonitoredItemsRequest{
 			SubscriptionID:     s.SubscriptionID,
 			TimestampsToReturn: ts,
@@ -329,8 +332,12 @@ func (s *Subscription) recreate() error {
 			}
 		}
 
-		for i, m := range s.items {
-			m.createResult = res.Results[i]
+		for i, item := range items {
+			s.items[res.Results[i].MonitoredItemID] = &monitoredItem{
+				item: item,
+				res:  res.Results[i],
+				ts:   ts,
+			}
 		}
 	}
 	dlog.Printf("subscription successfully recreated")
diff --git a/ua/extension_object.go b/ua/extension_object.go
index 6027d98..86d3c97 100644
--- a/ua/extension_object.go
+++ b/ua/extension_object.go
@@ -5,7 +5,6 @@
 package ua
 
 import (
-	"github.com/gopcua/opcua/errors"
 	"github.com/gopcua/opcua/id"
 )
 
@@ -76,7 +75,7 @@ func (e *ExtensionObject) Decode(b []byte) (int, error) {
 	typeID := e.TypeID.NodeID
 	e.Value = eotypes.New(typeID)
 	if e.Value == nil {
-		return buf.Pos(), errors.Errorf("invalid extension object with id %s", typeID)
+		return buf.Pos(), buf.Error()
 	}
 
 	body.ReadStruct(e.Value)
diff --git a/uasc/message.go b/uasc/message.go
index e466e05..8657d11 100644
--- a/uasc/message.go
+++ b/uasc/message.go
@@ -5,6 +5,8 @@
 package uasc
 
 import (
+	"math"
+
 	"github.com/gopcua/opcua/errors"
 	"github.com/gopcua/opcua/ua"
 )
@@ -111,25 +113,74 @@ func (m *Message) Decode(b []byte) (int, error) {
 }
 
 func (m *Message) Encode() ([]byte, error) {
-	body := ua.NewBuffer(nil)
+	chunks, err := m.EncodeChunks(math.MaxUint32)
+	if err != nil {
+		return nil, err
+	}
+	return chunks[0], nil
+}
+
+func (m *Message) EncodeChunks(maxBodySize uint32) ([][]byte, error) {
+	dataBody := ua.NewBuffer(nil)
+	dataBody.WriteStruct(m.TypeID)
+	dataBody.WriteStruct(m.Service)
+
+	if dataBody.Error() != nil {
+		return nil, dataBody.Error()
+	}
+
+	nrChunks := uint32(dataBody.Len())/(maxBodySize) + 1
+	chunks := make([][]byte, nrChunks)
+
 	switch m.Header.MessageType {
 	case "OPN":
-		body.WriteStruct(m.AsymmetricSecurityHeader)
+		partialHeader := ua.NewBuffer(nil)
+		partialHeader.WriteStruct(m.AsymmetricSecurityHeader)
+		partialHeader.WriteStruct(m.SequenceHeader)
+
+		if partialHeader.Error() != nil {
+			return nil, partialHeader.Error()
+		}
+
+		m.Header.MessageSize = uint32(12 + partialHeader.Len() + dataBody.Len())
+		buf := ua.NewBuffer(nil)
+		buf.WriteStruct(m.Header)
+		buf.Write(partialHeader.Bytes())
+		buf.Write(dataBody.Bytes())
+
+		return [][]byte{buf.Bytes()}, buf.Error()
+
 	case "CLO", "MSG":
-		body.WriteStruct(m.SymmetricSecurityHeader)
+
+		for i := uint32(0); i < nrChunks-1; i++ {
+			m.Header.MessageSize = maxBodySize + 24
+			m.Header.ChunkType = ChunkTypeIntermediate
+			chunk := ua.NewBuffer(nil)
+			chunk.WriteStruct(m.Header)
+			chunk.WriteStruct(m.SymmetricSecurityHeader)
+			chunk.WriteStruct(m.SequenceHeader)
+			chunk.Write(dataBody.ReadN(int(maxBodySize)))
+			if chunk.Error() != nil {
+				return nil, chunk.Error()
+			}
+
+			chunks[i] = chunk.Bytes()
+		}
+
+		m.Header.ChunkType = ChunkTypeFinal
+		m.Header.MessageSize = uint32(24 + dataBody.Len())
+		chunk := ua.NewBuffer(nil)
+		chunk.WriteStruct(m.Header)
+		chunk.WriteStruct(m.SymmetricSecurityHeader)
+		chunk.WriteStruct(m.SequenceHeader)
+		chunk.Write(dataBody.Bytes())
+		if chunk.Error() != nil {
+			return nil, chunk.Error()
+		}
+
+		chunks[nrChunks-1] = chunk.Bytes()
+		return chunks, nil
 	default:
 		return nil, errors.Errorf("invalid message type %q", m.Header.MessageType)
 	}
-	body.WriteStruct(m.SequenceHeader)
-	body.WriteStruct(m.TypeID)
-	body.WriteStruct(m.Service)
-	if body.Error() != nil {
-		return nil, body.Error()
-	}
-
-	m.Header.MessageSize = uint32(12 + body.Len())
-	buf := ua.NewBuffer(nil)
-	buf.WriteStruct(m.Header)
-	buf.Write(body.Bytes())
-	return buf.Bytes(), buf.Error()
 }
diff --git a/uasc/secure_channel.go b/uasc/secure_channel.go
index ac4796b..0d86054 100644
--- a/uasc/secure_channel.go
+++ b/uasc/secure_channel.go
@@ -8,6 +8,7 @@ import (
 	"context"
 	"crypto/rsa"
 	"crypto/x509"
+	"encoding/binary"
 	"io"
 	"math"
 	"sync"
@@ -526,6 +527,7 @@ func (s *SecureChannel) open(ctx context.Context, instance *channelInstance, req
 	reqID := s.nextRequestID()
 
 	s.openingInstance.algo = algo
+	s.openingInstance.SetMaximumBodySize(int(s.c.SendBufSize()))
 
 	localNonce, err := algo.MakeNonce()
 	if err != nil {
@@ -565,6 +567,8 @@ func (s *SecureChannel) handleOpenSecureChannelResponse(resp *ua.OpenSecureChann
 		return err
 	}
 
+	instance.SetMaximumBodySize(int(s.c.SendBufSize()))
+
 	s.instancesMu.Lock()
 	defer s.instancesMu.Unlock()
 
@@ -754,16 +758,6 @@ func (s *SecureChannel) sendAsyncWithTimeout(
 		return nil, err
 	}
 
-	b, err := m.Encode()
-	if err != nil {
-		return nil, err
-	}
-
-	b, err = instance.signAndEncrypt(m, b)
-	if err != nil {
-		return nil, err
-	}
-
 	var resp chan *response
 
 	if respRequired {
@@ -781,16 +775,33 @@ func (s *SecureChannel) sendAsyncWithTimeout(
 		s.handlersMu.Unlock()
 	}
 
-	// send the message
-	var n int
-	if n, err = s.c.Write(b); err != nil {
+	chunks, err := m.EncodeChunks(instance.maxBodySize)
+	if err != nil {
 		return nil, err
 	}
 
-	atomic.AddUint64(&instance.bytesSent, uint64(n))
-	atomic.AddUint32(&instance.messagesSent, 1)
+	for i, chunk := range chunks {
+		if i > 0 { // fix sequence number on subsequent chunks
+			number := instance.nextSequenceNumber()
+			binary.LittleEndian.PutUint32(chunk[16:], uint32(number))
+		}
+
+		chunk, err = instance.signAndEncrypt(m, chunk)
+		if err != nil {
+			return nil, err
+		}
 
-	debug.Printf("uasc %d/%d: send %T with %d bytes", s.c.ID(), reqID, req, len(b))
+		// send the message
+		var n int
+		if n, err = s.c.Write(chunk); err != nil {
+			return nil, err
+		}
+
+		atomic.AddUint64(&instance.bytesSent, uint64(n))
+		atomic.AddUint32(&instance.messagesSent, 1)
+
+		debug.Printf("uasc %d/%d: send %T with %d bytes", s.c.ID(), reqID, req, len(chunk))
+	}
 
 	return resp, nil
 }
diff --git a/uasc/secure_channel_instance.go b/uasc/secure_channel_instance.go
index e6e6265..8950c6f 100644
--- a/uasc/secure_channel_instance.go
+++ b/uasc/secure_channel_instance.go
@@ -33,11 +33,12 @@ type channelInstance struct {
 	securityTokenID uint32
 	sequenceNumber  uint32
 	algo            *uapolicy.EncryptionAlgorithm
+	maxBodySize     uint32
 
-	messagesSent uint32
-	// messagesReceived uint32
-	bytesSent uint64
+	bytesSent uint64 // atomic.Load/Store - needs to be aligned for 32bit systems
 	// bytesReceived    uint64
+	messagesSent uint32 // atomic.Load/Store
+	// messagesReceived uint32
 }
 
 func newChannelInstance(sc *SecureChannel) *channelInstance {
@@ -130,6 +131,23 @@ func (c *channelInstance) newMessage(srv interface{}, typeID uint16, requestID u
 	}
 }
 
+func (c *channelInstance) SetMaximumBodySize(chunkSize int) {
+	sequenceHeaderSize := 8
+	headerSize := 12
+	symmetricAlgorithmHeader := 4
+
+	// this is the formula proposed by OPCUA - source node-opcua
+	maxBodySize :=
+		c.algo.PlaintextBlockSize()*
+			((chunkSize-headerSize-symmetricAlgorithmHeader-c.algo.SignatureLength()-1)/c.algo.BlockSize()) -
+			sequenceHeaderSize
+	c.maxBodySize = uint32(maxBodySize)
+
+	// this is the formula proposed by ERN - source node-opcua
+	// maxBlock := (chunkSize - headerSize) / c.algo.BlockSize()
+	// c.maxBodySize = c.algo.PlaintextBlockSize()*maxBlock - sequenceHeaderSize - c.algo.SignatureLength() - 1
+}
+
 // signAndEncrypt encrypts the message bytes stored in b and returns the
 // data signed and encrypted per the security policy information from the
 // secure channel.


Test patch:
diff --git a/subscription_test.go b/subscription_test.go
new file mode 100644
index 0000000..f3a1e8c
--- /dev/null
+++ b/subscription_test.go
@@ -0,0 +1,106 @@
+package opcua
+
+import (
+	"testing"
+
+	"github.com/gopcua/opcua/ua"
+)
+
+// Running tool: /Users/frank/sdk/go1.17.1/bin/go test -benchmem -run=^$ -bench ^BenchmarkUnmonitorItems$ github.com/gopcua/opcua
+
+// goos: darwin
+// goarch: arm64
+// pkg: github.com/gopcua/opcua
+// BenchmarkUnmonitorItems/slice-8         	51153620	        24.03 ns/op	      20 B/op	       0 allocs/op
+// --- BENCH: BenchmarkUnmonitorItems/slice-8
+//     subscription_test.go:29: src 1 dst 0
+//     subscription_test.go:29: src 100 dst 50
+//     subscription_test.go:29: src 10000 dst 5000
+//     subscription_test.go:29: src 1000000 dst 500000
+//     subscription_test.go:29: src 51153620 dst 25576810
+// BenchmarkUnmonitorItems/slice_pre-alloc-8         	91635986	        22.77 ns/op	       8 B/op	       0 allocs/op
+// --- BENCH: BenchmarkUnmonitorItems/slice_pre-alloc-8
+//     subscription_test.go:51: src 1 dst 0
+//     subscription_test.go:51: src 100 dst 50
+//     subscription_test.go:51: src 10000 dst 5000
+//     subscription_test.go:51: src 1000000 dst 500000
+//     subscription_test.go:51: src 91635986 dst 45817993
+// BenchmarkUnmonitorItems/map-8                     	39885550	        43.72 ns/op	       0 B/op	       0 allocs/op
+// --- BENCH: BenchmarkUnmonitorItems/map-8
+//     subscription_test.go:75: src 0
+//     subscription_test.go:75: src 50
+//     subscription_test.go:75: src 5000
+//     subscription_test.go:75: src 500000
+//     subscription_test.go:75: src 19942775
+// PASS
+// ok  	github.com/gopcua/opcua	116.192s
+
+func BenchmarkUnmonitorItems(b *testing.B) {
+	b.Run("slice", func(b *testing.B) {
+		src := make([]*monitoredItem, b.N)
+		for i := 0; i < b.N; i++ {
+			src[i] = &monitoredItem{
+				res: &ua.MonitoredItemCreateResult{
+					MonitoredItemID: uint32(i),
+				},
+			}
+		}
+
+		b.ResetTimer()
+		var dst []*monitoredItem
+		for _, item := range src {
+			if item.res.MonitoredItemID%2 == 0 {
+				continue
+			}
+			dst = append(dst, item)
+		}
+
+		b.Log("src", len(src), "dst", len(dst)) // ensure src and dst are not GC'ed
+	})
+
+	b.Run("slice pre-alloc", func(b *testing.B) {
+		src := make([]*monitoredItem, b.N)
+		for i := 0; i < b.N; i++ {
+			src[i] = &monitoredItem{
+				res: &ua.MonitoredItemCreateResult{
+					MonitoredItemID: uint32(i),
+				},
+			}
+		}
+
+		b.ResetTimer()
+		dst := make([]*monitoredItem, 0, len(src))
+		for _, item := range src {
+			if item.res.MonitoredItemID%2 == 0 {
+				continue
+			}
+			dst = append(dst, item)
+		}
+
+		b.Log("src", len(src), "dst", len(dst)) // ensure src and dst are not GC'ed
+	})
+
+	b.Run("map", func(b *testing.B) {
+		idsToDelete := []uint32{}
+		src := make(map[uint32]*monitoredItem, b.N)
+		for i := 0; i < b.N; i++ {
+			id := uint32(i)
+			src[id] = &monitoredItem{
+				res: &ua.MonitoredItemCreateResult{
+					MonitoredItemID: id,
+				},
+			}
+
+			if id%2 == 0 {
+				idsToDelete = append(idsToDelete, id)
+			}
+		}
+
+		b.ResetTimer()
+		for _, id := range idsToDelete {
+			delete(src, id)
+		}
+
+		b.Log("src", len(src)) // ensure src and dst are not GC'ed
+	})
+}
diff --git a/uatest/read_test.go b/uatest/read_test.go
index eb09c56..4410580 100644
--- a/uatest/read_test.go
+++ b/uatest/read_test.go
@@ -1,4 +1,3 @@
-//go:build integration
 // +build integration
 
 package uatest
@@ -23,8 +22,6 @@ func TestRead(t *testing.T) {
 		{ua.NewStringNodeID(2, "rw_bool"), true},
 		{ua.NewStringNodeID(2, "ro_int32"), int32(5)},
 		{ua.NewStringNodeID(2, "rw_int32"), int32(5)},
-		{ua.NewStringNodeID(2, "array_int32"), []int32{1, 2, 3}},
-		{ua.NewStringNodeID(2, "2d_array_int32"), [][]int32{{1}, {2}, {3}}},
 	}
 
 	srv := NewServer("rw_server.py")
diff --git a/uatest/read_unknow_node_id_server.py b/uatest/read_unknow_node_id_server.py
new file mode 100755
index 0000000..71f91ed
--- /dev/null
+++ b/uatest/read_unknow_node_id_server.py
@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+
+from opcua import ua, Server
+from opcua.ua import uatypes
+
+from opcua.ua.object_ids import ObjectIds
+
+
+class IntVal(uatypes.FrozenClass):
+    ua_types = [
+        ('i', 'Int64'),
+    ]
+
+    def __init__(self):
+        self.i = 0
+        self._freeze = True
+
+    def __str__(self):
+        return 'IntVal(' + 'i:' + str(self.i) + ')'
+
+    __repr__ = __str__
+
+
+if __name__ == "__main__":
+    server = Server()
+    server.set_endpoint("opc.tcp://0.0.0.0:4840/")
+
+    ns = server.register_namespace("http://gopcua.com/")
+
+    uatypes.register_extension_object('IntVal', ua.StringNodeId("IntValType", ns), IntVal)
+
+    # definitely not clear why this is needed, but without it does not work
+    setattr(ua.ObjectIds, 'IntVal', 'IntValType')
+
+    main = server.nodes.objects.add_object(ua.NodeId("main", ns), "main")
+
+    main.add_variable(ua.NodeId("IntValZero", ns), "IntValZero", IntVal(), ua.VariantType.ExtensionObject)
+
+    server.start()
diff --git a/uatest/read_unknow_node_id_test.go b/uatest/read_unknow_node_id_test.go
new file mode 100644
index 0000000..ed11aba
--- /dev/null
+++ b/uatest/read_unknow_node_id_test.go
@@ -0,0 +1,54 @@
+//go:build integration
+// +build integration
+
+package uatest
+
+import (
+	"context"
+	"testing"
+
+	"github.com/gopcua/opcua"
+	"github.com/gopcua/opcua/id"
+	"github.com/gopcua/opcua/ua"
+)
+
+// TestRead performs an integration test to read values
+// from an OPC/UA server.
+func TestReadUnknowNodeID(t *testing.T) {
+	srv := NewServer("read_unknow_node_id_server.py")
+	defer srv.Close()
+
+	c := opcua.NewClient(srv.Endpoint, srv.Opts...)
+	if err := c.Connect(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+
+	// read node with unknown extension object
+	// This should be OK
+	nodeWithUnknownType := ua.NewStringNodeID(2, "IntValZero")
+	resp, err := c.Read(&ua.ReadRequest{
+		NodesToRead: []*ua.ReadValueID{
+			{NodeID: nodeWithUnknownType},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if got, want := resp.Results[0].Status, ua.StatusBadDataTypeIDUnknown; got != want {
+		t.Errorf("got status %v want %v for a node with an unknown type", got, want)
+	}
+
+	// check that the connection is still usable by reading another node.
+	_, err = c.Read(&ua.ReadRequest{
+		NodesToRead: []*ua.ReadValueID{
+			{
+				NodeID: ua.NewNumericNodeID(0, id.Server_ServerStatus_State),
+			},
+		},
+	})
+	if err != nil {
+		t.Error(err)
+	}
+}
diff --git a/uatest/rw_server.py b/uatest/rw_server.py
index 318c616..43cd03a 100755
--- a/uatest/rw_server.py
+++ b/uatest/rw_server.py
@@ -16,7 +16,4 @@ if __name__ == "__main__":
     rwInt32 = main.add_variable(ua.NodeId("rw_int32", ns), "rw_int32", 5, ua.VariantType.Int32)
     rwInt32.set_writable()
 
-    roArrayInt32 = main.add_variable(ua.NodeId("array_int32", ns), "array_int32", [1,2,3], ua.VariantType.Int32)
-    ro2DArrayInt32 = main.add_variable(ua.NodeId("2d_array_int32", ns), "2d_array_int32", [[1],[2],[3]], ua.VariantType.Int32)
-
     server.start()
diff --git a/uatest/timeout_test.go b/uatest/timeout_test.go
new file mode 100644
index 0000000..aad38f8
--- /dev/null
+++ b/uatest/timeout_test.go
@@ -0,0 +1,67 @@
+// +build integration
+
+package uatest
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/gopcua/opcua"
+)
+
+const (
+	// this _must_ be a "host" that will silently eat SYNs (no RSTs)
+	// 203.0.113.0/24 is IETF TEST-NET-3
+	tcpNoRstTestServer   = "opc.tcp://203.0.113.0:4840"
+	forceTimeoutDuration = time.Second * 5
+)
+
+func TestClientTimeoutViaOptions(t *testing.T) {
+	c := opcua.NewClient(tcpNoRstTestServer, opcua.DialTimeout(forceTimeoutDuration))
+
+	connectAndValidate(t, c, context.Background(), forceTimeoutDuration)
+}
+
+func TestClientTimeoutViaContext(t *testing.T) {
+	c := opcua.NewClient(tcpNoRstTestServer)
+
+	ctx, cancel := context.WithTimeout(context.Background(), forceTimeoutDuration)
+	defer cancel()
+
+	connectAndValidate(t, c, ctx, forceTimeoutDuration)
+}
+
+func connectAndValidate(t *testing.T, c *opcua.Client, ctx context.Context, d time.Duration) {
+	start := time.Now()
+
+	err := c.Connect(ctx)
+	if err == nil {
+		t.Fatal("err should not be nil")
+	}
+
+	elapsed := time.Since(start)
+
+	if oe, ok := err.(*net.OpError); ok {
+		if !oe.Timeout() {
+			t.Fatalf("got %#v, wanted net.timeoutError", oe.Unwrap())
+		}
+	} else {
+		t.Fatalf("got %T, wanted %T", err, net.OpError{})
+	}
+
+	pct := 0.05
+
+	if !within(elapsed, d, pct) {
+		t.Fatalf("took %s, expected %s +/- %v%%", elapsed, d, pct*100)
+	}
+}
+
+func within(x, y time.Duration, pct float64) bool {
+	if pct > 1 || pct < 0 {
+		panic("invalid pct")
+	}
+	p := float64(x) * pct
+	return float64(y) >= (float64(x)-p) && float64(y) <= (float64(x)+p)
+}