Merge pull request #653 from bitomaxsp/error/handling

Improve error handling

client.go

@@ -193,7 +193,7 @@ const (
 )
 
 // Connect establishes a secure channel and creates a new session.
-func (c *Client) Connect(ctx context.Context) (err error) {
+func (c *Client) Connect(ctx context.Context) error {
 	// todo(fs): remove with v0.5.0
 	if c.cfgerr != nil {
 		return c.cfgerr
@@ -321,8 +321,6 @@ func (c *Client) monitor(ctx context.Context) {
 				action = createSecureChannel
 			}
 
-			c.setState(Disconnected)
-
 			c.pauseSubscriptions(ctx)
 
 			var (
@@ -387,6 +385,8 @@ func (c *Client) monitor(ctx context.Context) {
 						// This only works if the session is still open on the server
 						// otherwise recreate it
 
+						c.setState(Reconnecting)
+
 						s := c.Session()
 						if s == nil {
 							dlog.Printf("no session to restore")
@@ -416,6 +416,7 @@ func (c *Client) monitor(ctx context.Context) {
 					case recreateSession:
 						dlog.Printf("action: recreateSession")
 
+						c.setState(Reconnecting)
 						// create a new session to replace the previous one
 
 						dlog.Printf("trying to recreate session")
@@ -460,6 +461,12 @@ func (c *Client) monitor(ctx context.Context) {
 						// recreate them all if that fails.
 						res, err := c.transferSubscriptions(ctx, subIDs)
 						switch {
+
+						case errors.Is(err, ua.StatusBadServiceUnsupported):
+							dlog.Printf("transfer subscriptions not supported. Recreating all subscriptions: %v", err)
+							subsToRepublish = nil
+							subsToRecreate = subIDs
+
 						case err != nil:
 							dlog.Printf("transfer subscriptions failed. Recreating all subscriptions: %v", err)
 							subsToRepublish = nil
@@ -690,6 +697,15 @@ type Session struct {
 	// Session response. Used to generate the signatures for the ActivateSessionRequest
 	// and User Authorization
 	serverNonce []byte
+
+	// revisedTimeout is the actual maximum time that a Session shall remain open without activity.
+	revisedTimeout time.Duration
+}
+
+// RevisedTimeout return actual maximum time that a Session shall remain open without activity.
+// This value is provided by the server in response to CreateSession.
+func (s *Session) RevisedTimeout() time.Duration {
+	return s.revisedTimeout
 }
 
 // CreateSession creates a new session which is not yet activated and not
@@ -764,6 +780,7 @@ func (c *Client) CreateSessionWithContext(ctx context.Context, cfg *uasc.Session
 			resp:              res,
 			serverNonce:       res.ServerNonce,
 			serverCertificate: res.ServerCertificate,
+			revisedTimeout:    time.Duration(res.RevisedSessionTimeout) * time.Millisecond,
 		}
 
 		return nil

client_sub.go

@@ -408,6 +408,10 @@ func (c *Client) publish(ctx context.Context) error {
 		dlog.Printf("error: session not active. pausing publish loop")
 		return err
 
+	case err == ua.StatusBadSessionIDInvalid:
+		dlog.Printf("error: session not valid. pausing publish loop")
+		return err
+
 	case err == ua.StatusBadServerNotConnected:
 		dlog.Printf("error: no connection. pausing publish loop")
 		return err

examples/read/read.go

@@ -6,8 +6,11 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
+	"io"
 	"log"
+	"time"
 
 	"github.com/gopcua/opcua"
 	"github.com/gopcua/opcua/debug"
@@ -44,12 +47,44 @@ func main() {
 		TimestampsToReturn: ua.TimestampsToReturnBoth,
 	}
 
-	resp, err := c.ReadWithContext(ctx, req)
-	if err != nil {
-		log.Fatalf("Read failed: %s", err)
+	var resp *ua.ReadResponse
+	for {
+		resp, err = c.ReadWithContext(ctx, req)
+		if err == nil {
+			break
+		}
+
+		// Following switch contains known errors that can be retried by the user.
+		// Best practice is to do it on read operations.
+		switch {
+		case err == io.EOF && c.State() != opcua.Closed:
+			// has to be retried unless user closed the connection
+			time.After(1 * time.Second)
+			continue
+
+		case errors.Is(err, ua.StatusBadSessionIDInvalid):
+			// Session is not activated has to be retried. Session will be recreated internally.
+			time.After(1 * time.Second)
+			continue
+
+		case errors.Is(err, ua.StatusBadSessionNotActivated):
+			// Session is invalid has to be retried. Session will be recreated internally.
+			time.After(1 * time.Second)
+			continue
+
+		case errors.Is(err, ua.StatusBadSecureChannelIDInvalid):
+			// secure channel will be recreated internally.
+			time.After(1 * time.Second)
+			continue
+
+		default:
+			log.Fatalf("Read failed: %s", err)
+		}
 	}
-	if resp.Results[0].Status != ua.StatusOK {
+
+	if resp != nil && resp.Results[0].Status != ua.StatusOK {
 		log.Fatalf("Status not OK: %v", resp.Results[0].Status)
 	}
+
 	log.Printf("%#v", resp.Results[0].Value.Value())
 }

examples/subscribe/subscribe.go

@@ -47,6 +47,7 @@ func main() {
 	if ep == nil {
 		log.Fatal("Failed to find suitable endpoint")
 	}
+	ep.EndpointURL = *endpoint
 
 	fmt.Println("*", ep.SecurityPolicyURI, ep.SecurityMode)
 

node.go

@@ -31,7 +31,7 @@ func (n *Node) String() string {
 // Note: Starting with v0.5 this method will require a context
 // and the corresponding XXXWithContext(ctx) method will be removed.
 func (n *Node) NodeClass(ctx context.Context) (ua.NodeClass, error) {
-	return n.NodeClassWithContext(context.Background())
+	return n.NodeClassWithContext(ctx)
 }
 
 // Note: Starting with v0.5 this method is superseded by the non 'WithContext' method.