Fix the watcher maybe lost the notify

watcher.go

@@ -2,6 +2,7 @@ package prophet
 
 import (
 	"sync"
+	"sync/atomic"
 
 	"github.com/fagongzi/goetty"
 )
@@ -50,6 +51,7 @@ type InitWatcher struct {
 
 // EventNotify event notify
 type EventNotify struct {
+	Seq   uint64 `json:"seq"`
 	Event int    `json:"event"`
 	Value []byte `json:"value"`
 }
@@ -175,12 +177,14 @@ func (evt *EventNotify) ReadLeaderChangerValue() (uint64, uint64) {
 }
 
 type watcher struct {
+	seq  uint64
 	info *InitWatcher
 	conn goetty.IOSession
 }
 
 func (wt *watcher) notify(evt *EventNotify) {
 	if MatchEvent(evt.Event, wt.info.Flag) {
+		evt.Seq = atomic.AddUint64(&wt.seq, 1)
 		err := wt.conn.WriteAndFlush(evt)
 		if err != nil {
 			wt.conn.Close()

watcher_client.go

@@ -101,8 +101,8 @@ func (w *Watcher) resetConnWithProphet(flag int) {
 			Flag: flag,
 		})
 		if err != nil {
-			time.Sleep(time.Millisecond * 200)
 			conn.Close()
+			time.Sleep(time.Millisecond * 200)
 			continue
 		}
 
@@ -125,6 +125,8 @@ func (w *Watcher) watchDog(flag int) {
 }
 
 func (w *Watcher) startReadLoop() {
+	expectSeq := uint64(0)
+
 	for {
 		msg, err := w.conn.Read()
 		if err != nil {
@@ -133,6 +135,17 @@ func (w *Watcher) startReadLoop() {
 		}
 
 		if evt, ok := msg.(*EventNotify); ok {
+			log.Infof("%d, %d", expectSeq, evt.Seq)
+			// we lost some event notify, close the conection, and retry
+			if expectSeq != evt.Seq {
+				log.Warnf("prophet: watch lost some event notify, expect seq %d, but %d, close and retry",
+					expectSeq,
+					evt.Seq)
+				w.conn.Close()
+				return
+			}
+
+			expectSeq = evt.Seq + 1
 			w.eventC <- evt
 		} else {
 			w.conn.Close()

watcher_test.go

@@ -0,0 +1,49 @@
+package prophet
+
+import (
+	"testing"
+	"time"
+
+	"github.com/fagongzi/goetty"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNotify(t *testing.T) {
+	addr := "127.0.0.1:12345"
+	bizCodec := &codec{}
+	c := make(chan *EventNotify, 10)
+	s := goetty.NewServer(addr,
+		goetty.WithServerDecoder(goetty.NewIntLengthFieldBasedDecoder(bizCodec)),
+		goetty.WithServerEncoder(goetty.NewIntLengthFieldBasedEncoder(bizCodec)))
+
+	conns := 0
+	go s.Start(func(conn goetty.IOSession) error {
+		conns++
+		for evt := range c {
+			log.Infof("write %d", evt.Seq)
+			conn.WriteAndFlush(evt)
+		}
+
+		log.Infof("exit")
+		return nil
+	})
+	<-s.Started()
+
+	defer s.Stop()
+
+	w := NewWatcher("127.0.0.1:12345")
+	ch := w.Watch(EventResourceChaned)
+	go func() {
+		for range ch {
+
+		}
+	}()
+
+	c <- &EventNotify{Event: EventInit, Seq: 0}
+	time.Sleep(time.Millisecond * 100)
+	assert.Equal(t, 1, conns, "test watcher failed")
+
+	c <- &EventNotify{Event: EventResourceChaned, Seq: 2}
+	time.Sleep(time.Millisecond * 200)
+	assert.Equal(t, 2, conns, "test watcher failed")
+}