fix of silently stop notifications (#183)

fixes #179
implemented according to https://tools.ietf.org/html/rfc7641#section-3.4

Author: jkralik

net/observation/observation.go

@@ -0,0 +1,22 @@
+package observation
+
+import (
+	"time"
+)
+
+// ObservationSequenceTimeout defines how long is sequence number is valid. https://tools.ietf.org/html/rfc7641#section-3.4
+const ObservationSequenceTimeout = 128 * time.Second
+
+// ValidSequenceNumber implements conditions in https://tools.ietf.org/html/rfc7641#section-3.4
+func ValidSequenceNumber(old, new uint32, lastEventOccurs time.Time, now time.Time) bool {
+	if old < new && (new-old) < (1<<23) {
+		return true
+	}
+	if old > new && (old-new) > (1<<23) {
+		return true
+	}
+	if now.Sub(lastEventOccurs) > ObservationSequenceTimeout {
+		return true
+	}
+	return false
+}

net/observation/observation_test.go

@@ -0,0 +1,107 @@
+package observation
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidSequenceNumber(t *testing.T) {
+	type args struct {
+		old             uint32
+		new             uint32
+		lastEventOccurs time.Time
+		now             time.Time
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "(1 << 25)-1, 0, now-1s, now",
+			args: args{
+				old:             (1 << 25)-1,
+				new:             0,
+				lastEventOccurs: time.Now().Add(-time.Second),
+				now:             time.Now(),
+			},
+			want: true,
+		},
+		{
+			name: "0, 1, 0, now",
+			args: args{
+				new: 1,
+				now: time.Now(),
+			},
+			want: true,
+		},
+		{
+			name: "1582, 1583, now-1s, now",
+			args: args{
+				old:             1582,
+				new:             1583,
+				lastEventOccurs: time.Now().Add(-time.Second),
+				now:             time.Now(),
+			},
+			want: true,
+		},
+		{
+			name: "1582, 1, now-129s, now",
+			args: args{
+				old:             1582,
+				new:             1,
+				lastEventOccurs: time.Now().Add(-time.Second * 129),
+				now:             time.Now(),
+			},
+			want: true,
+		},
+		{
+			name: "1582, 1, now-125s, now",
+			args: args{
+				old:             1582,
+				new:             1,
+				lastEventOccurs: time.Now().Add(-time.Second * 125),
+				now:             time.Now(),
+			},
+			want: false,
+		},
+		{
+			name: "1 << 23, 0, now-1s, now",
+			args: args{
+				old:             1 << 23,
+				new:             0,
+				lastEventOccurs: time.Now().Add(-time.Second),
+				now:             time.Now(),
+			},
+			want: false,
+		},
+		{
+			name: "0, 1 << 23+1, now-1s, now",
+			args: args{
+				old:             0,
+				new:             1 << 23+1,
+				lastEventOccurs: time.Now().Add(-time.Second),
+				now:             time.Now(),
+			},
+			want: false,
+		},
+		{
+			name: "1582, 1582, now-1s, now",
+			args: args{
+				old:             1582,
+				new:             1582,
+				lastEventOccurs: time.Now().Add(-time.Second),
+				now:             time.Now(),
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := ValidSequenceNumber(tt.args.old, tt.args.new, tt.args.lastEventOccurs, tt.args.now)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

tcp/clientobserve.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/plgd-dev/go-coap/v2/message"
 	"github.com/plgd-dev/go-coap/v2/message/codes"
+	"github.com/plgd-dev/go-coap/v2/net/observation"
 	"github.com/plgd-dev/go-coap/v2/tcp/message/pool"
 )
 
@@ -18,6 +20,7 @@ type Observation struct {
 	obsSequence uint32
 	etag        []byte
 	cc          *ClientConn
+	lastEvent   time.Time
 
 	mutex sync.Mutex
 }
@@ -67,16 +70,17 @@ func (o *Observation) wantBeNotified(r *pool.Message) bool {
 	if err != nil {
 		return true
 	}
+	now := time.Now()
 
 	o.mutex.Lock()
 	defer o.mutex.Unlock()
-	//obs starts with 0, after that check obsSequence
-	if obsSequence != 0 && o.obsSequence > obsSequence {
-		return false
+	if observation.ValidSequenceNumber(o.obsSequence, obsSequence, o.lastEvent, now) {
+		o.obsSequence = obsSequence
+		o.lastEvent = now
+		return true
 	}
-	o.obsSequence = obsSequence
 
-	return true
+	return false
 }
 
 // Observe subscribes for every change of resource on path.

udp/client/clientobserve.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/plgd-dev/go-coap/v2/message"
 	"github.com/plgd-dev/go-coap/v2/message/codes"
+	"github.com/plgd-dev/go-coap/v2/net/observation"
 	"github.com/plgd-dev/go-coap/v2/udp/message/pool"
 )
 
@@ -18,6 +20,7 @@ type Observation struct {
 	obsSequence uint32
 	etag        []byte
 	cc          *ClientConn
+	lastEvent   time.Time
 
 	mutex sync.Mutex
 }
@@ -72,16 +75,18 @@ func (o *Observation) wantBeNotified(r *pool.Message) bool {
 	if err != nil {
 		return true
 	}
+	now := time.Now()
 
 	o.mutex.Lock()
 	defer o.mutex.Unlock()
-	//obs starts with 0, after that check obsSequence
-	if obsSequence != 0 && o.obsSequence > obsSequence {
-		return false
+
+	if observation.ValidSequenceNumber(o.obsSequence, obsSequence, o.lastEvent, now) {
+		o.obsSequence = obsSequence
+		o.lastEvent = now
+		return true
 	}
-	o.obsSequence = obsSequence
 
-	return true
+	return false
 }
 
 // Observe subscribes for every change of resource on path.