api: add events subscription support

A user can create watcher by the Connection.NewWatcher() call:

    watcher = conn.NewWatcker("key", func(event WatchEvent) {
		// The callback code.
	})

After that, the watcher callback is invoked for the first time. In
this case, the callback is triggered whether or not the key has
already been broadcast. All subsequent invocations are triggered with
box.broadcast() called on the remote host. If a watcher is subscribed
for a key that has not been broadcast yet, the callback is triggered
only once, after the registration of the watcher.

If the key is updated while the watcher callback is running, the
callback will be invoked again with the latest value as soon as it
returns.

Multiple watchers can be created for one key.

If you don’t need the watcher anymore, you can unregister it using
the Unregister method:

	watcher.Unregister()

The api is similar to net.box implementation [1].

It also adds a BroadcastRequest to make it easier to send broadcast
messages.

1. https://www.tarantool.io/ru/doc/latest/reference/reference_lua/net_box/#conn-watch

Closes #119

Author: oleg-jukovec

CHANGELOG.md

@@ -10,6 +10,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 
 ### Added
 
+- Event subscription support (#119)
+
 ### Changed
 
 ### Fixed

connection.go

@@ -53,6 +53,8 @@ const (
 	// LogUnexpectedResultId is logged when response with unknown id was received.
 	// Most probably it is due to request timeout.
 	LogUnexpectedResultId
+	// LogReadWatchEventFailed is logged when failed to read a watch event.
+	LogReadWatchEventFailed
 )
 
 // ConnEvent is sent throw Notify channel specified in Opts.
@@ -62,6 +64,12 @@ type ConnEvent struct {
 	When time.Time
 }
 
+// A raw watch event.
+type connWatchEvent struct {
+	key   string
+	value interface{}
+}
+
 var epoch = time.Now()
 
 // Logger is logger type expected to be passed in options.
@@ -83,6 +91,9 @@ func (d defaultLogger) Report(event ConnLogKind, conn *Connection, v ...interfac
 	case LogUnexpectedResultId:
 		resp := v[0].(*Response)
 		log.Printf("tarantool: connection %s got unexpected resultId (%d) in response", conn.addr, resp.RequestId)
+	case LogReadWatchEventFailed:
+		err := v[0].(error)
+		log.Printf("tarantool: unable to parse watch event: %s\n", err)
 	default:
 		args := append([]interface{}{"tarantool: unexpected event ", event, conn}, v...)
 		log.Print(args...)
@@ -146,6 +157,9 @@ type Connection struct {
 	lenbuf  [PacketLengthBytes]byte
 
 	lastStreamId uint64
+
+	eventMap sync.Map
+	eventCh  chan connWatchEvent
 }
 
 var _ = Connector(&Connection{}) // Check compatibility with connector interface.
@@ -321,6 +335,7 @@ func Connect(addr string, opts Opts) (conn *Connection, err error) {
 		control:          make(chan struct{}),
 		opts:             opts,
 		dec:              newDecoder(&smallBuf{}),
+		eventCh:          make(chan connWatchEvent, 1024),
 	}
 	maxprocs := uint32(runtime.GOMAXPROCS(-1))
 	if conn.opts.Concurrency == 0 || conn.opts.Concurrency > maxprocs*128 {
@@ -355,6 +370,8 @@ func Connect(addr string, opts Opts) (conn *Connection, err error) {
 		conn.opts.Logger = defaultLogger{}
 	}
 
+	go conn.eventer(conn.eventCh)
+
 	if err = conn.createConnection(false); err != nil {
 		ter, ok := err.(Error)
 		if conn.opts.Reconnect <= 0 {
@@ -520,7 +537,20 @@ func (conn *Connection) dial() (err error) {
 		}
 	}
 
-	// Only if connected and authenticated.
+	// Watchers
+	conn.eventMap.Range(func(key, value interface{}) bool {
+		req := newWatchRequest(key.(string))
+		if err = conn.writeRequest(w, req); err != nil {
+			return false
+		}
+		return true
+	})
+
+	if err != nil {
+		return fmt.Errorf("unable to register watch: %w", err)
+	}
+
+	// Only if connected and fully initialized.
 	conn.lockShards()
 	conn.c = connection
 	atomic.StoreUint32(&conn.state, connConnected)
@@ -581,23 +611,33 @@ func pack(h *smallWBuf, enc *encoder, reqid uint32,
 	return
 }
 
-func (conn *Connection) writeAuthRequest(w *bufio.Writer, scramble []byte) (err error) {
+func (conn *Connection) writeRequest(w *bufio.Writer, req Request) (err error) {
 	var packet smallWBuf
-	req := newAuthRequest(conn.opts.User, string(scramble))
 	err = pack(&packet, newEncoder(&packet), 0, req, ignoreStreamId, conn.Schema)
 
 	if err != nil {
-		return errors.New("auth: pack error " + err.Error())
+		return fmt.Errorf("pack error %w", err)
 	}
 	if err := write(w, packet.b); err != nil {
-		return errors.New("auth: write error " + err.Error())
+		return fmt.Errorf("write error %w", err)
 	}
 	if err = w.Flush(); err != nil {
-		return errors.New("auth: flush error " + err.Error())
+		return fmt.Errorf("flush error %w", err)
 	}
 	return
 }
 
+func (conn *Connection) writeAuthRequest(w *bufio.Writer, scramble []byte) (err error) {
+	req := newAuthRequest(conn.opts.User, string(scramble))
+
+	err = conn.writeRequest(w, req)
+	if err != nil {
+		return fmt.Errorf("auth: %w", err)
+	}
+
+	return nil
+}
+
 func (conn *Connection) readAuthResponse(r io.Reader) (err error) {
 	respBytes, err := conn.read(r)
 	if err != nil {
@@ -655,6 +695,7 @@ func (conn *Connection) closeConnection(neterr error, forever bool) (err error)
 	defer conn.unlockShards()
 	if forever {
 		if conn.state != connClosed {
+			close(conn.eventCh)
 			close(conn.control)
 			atomic.StoreUint32(&conn.state, connClosed)
 			conn.notify(Closed)
@@ -774,6 +815,44 @@ func (conn *Connection) writer(w *bufio.Writer, c net.Conn) {
 	}
 }
 
+func readWatchEvent(reader io.Reader) (connWatchEvent, error) {
+	keyExist := false
+	event := connWatchEvent{}
+	d := newDecoder(reader)
+
+	if l, err := d.DecodeMapLen(); err == nil {
+		for ; l > 0; l-- {
+			if cd, err := d.DecodeInt(); err == nil {
+				switch cd {
+				case KeyEvent:
+					if event.key, err = d.DecodeString(); err != nil {
+						return event, err
+					}
+					keyExist = true
+				case KeyEventData:
+					if event.value, err = d.DecodeInterface(); err != nil {
+						return event, err
+					}
+				default:
+					if err = d.Skip(); err != nil {
+						return event, err
+					}
+				}
+			} else {
+				return event, err
+			}
+		}
+	} else {
+		return event, err
+	}
+
+	if !keyExist {
+		return event, errors.New("watch event does not have a key")
+	}
+
+	return event, nil
+}
+
 func (conn *Connection) reader(r *bufio.Reader, c net.Conn) {
 	for atomic.LoadUint32(&conn.state) != connClosed {
 		respBytes, err := conn.read(r)
@@ -789,7 +868,14 @@ func (conn *Connection) reader(r *bufio.Reader, c net.Conn) {
 		}
 
 		var fut *Future = nil
-		if resp.Code == PushCode {
+		if resp.Code == EventCode {
+			if event, err := readWatchEvent(&resp.buf); err == nil {
+				conn.eventCh <- event
+			} else {
+				conn.opts.Logger.Report(LogReadWatchEventFailed, conn, err)
+			}
+			continue
+		} else if resp.Code == PushCode {
 			if fut = conn.peekFuture(resp.RequestId); fut != nil {
 				fut.AppendPush(resp)
 			}
@@ -799,12 +885,36 @@ func (conn *Connection) reader(r *bufio.Reader, c net.Conn) {
 				conn.markDone(fut)
 			}
 		}
+
 		if fut == nil {
 			conn.opts.Logger.Report(LogUnexpectedResultId, conn, resp)
 		}
 	}
 }
 
+func (conn *Connection) eventer(ch chan connWatchEvent) {
+	for {
+		event, ok := <-ch
+		if !ok {
+			break
+		}
+
+		if value, ok := conn.eventMap.Load(event.key); ok {
+			data := value.(*watchDataShared)
+			data.condMutex.Lock()
+			data.value = event.value
+			data.version += 1
+			data.condMutex.Unlock()
+
+			data.cond.Broadcast()
+
+			if atomic.LoadUint32(&conn.state) == connConnected {
+				conn.Do(newWatchRequest(event.key))
+			}
+		}
+	}
+}
+
 func (conn *Connection) newFuture(ctx context.Context) (fut *Future) {
 	fut = NewFuture()
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
@@ -960,6 +1070,18 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
 		return
 	}
 	shard.bufmut.Unlock()
+
+	if req.Async() {
+		if fut = conn.fetchFuture(reqid); fut != nil {
+			resp := &Response{
+				RequestId: reqid,
+				Code:      OkCode,
+			}
+			fut.SetResponse(resp)
+			conn.markDone(fut)
+		}
+	}
+
 	if firstWritten {
 		conn.dirtyShard <- shardn
 	}
@@ -1163,3 +1285,121 @@ func (conn *Connection) NewStream() (*Stream, error) {
 		Conn: conn,
 	}, nil
 }
+
+type watchDataShared struct {
+	value     interface{}
+	version   uint
+	cond      *sync.Cond
+	condMutex sync.RWMutex
+
+	watchCnt   int32
+	watchMutex sync.Mutex
+}
+
+type connWatcher struct {
+	shared *watchDataShared
+	done   chan struct{}
+}
+
+// Unregister unregisters the connection watcher.
+//
+// Keep in mind that garbage collection of a watcher handle doesn’t lead to the
+// watcher’s destruction. In this case, the watcher remains registered. You
+// need to call Unregister() directly.
+func (w *connWatcher) Unregister() {
+	if w.shared != nil {
+		close(w.done)
+		w.shared.cond.Broadcast()
+		w.shared = nil
+		w.done = nil
+	}
+}
+
+// NewWatcher creates a new Watcher object for the connection.
+//
+// After watcher creation, the watcher callback is invoked for the first time.
+// In this case, the callback is triggered whether or not the key has already
+// been broadcast. All subsequent invocations are triggered with
+// box.broadcast() called on the remote host. If a watcher is subscribed for a
+// key that has not been broadcast yet, the callback is triggered only once,
+// after the registration of the watcher.
+//
+// The watcher callbacks are always invoked in a separate goroutine. A watcher
+// callback is never executed in parallel with itself, but they can be executed
+// in parallel to other watchers.
+//
+// If the key is updated while the watcher callback is running, the callback
+// will be invoked again with the latest value as soon as it returns.
+//
+// Watchers survive reconnection. All registered watchers are automatically
+// resubscribed when the connection is reestablished.
+//
+// See:
+// https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_events/#box-watchers
+//
+// Since 1.10.0
+func (conn *Connection) NewWatcher(key string, callback WatchCallback) (Watcher, error) {
+	var shared *watchDataShared
+
+	if val, ok := conn.eventMap.Load(key); !ok {
+		shared = &watchDataShared{
+			value:    nil,
+			version:  0,
+			watchCnt: 0,
+		}
+		shared.cond = sync.NewCond(shared.condMutex.RLocker())
+
+		if val, ok := conn.eventMap.LoadOrStore(key, shared); ok {
+			shared = val.(*watchDataShared)
+		}
+	} else {
+		shared = val.(*watchDataShared)
+	}
+
+	shared.watchMutex.Lock()
+	if shared.watchCnt == 0 {
+		if _, err := conn.Do(newWatchRequest(key)).Get(); err != nil {
+			return nil, err
+		}
+	}
+	atomic.AddInt32(&shared.watchCnt, 1)
+	shared.watchMutex.Unlock()
+
+	version := uint(0)
+	done := make(chan struct{})
+	go func() {
+		for {
+			shared.cond.L.Lock()
+			for {
+				select {
+				case <-done:
+					shared.cond.L.Unlock()
+					if atomic.AddInt32(&shared.watchCnt, -1) == 0 {
+						conn.Do(newUnwatchRequest(key))
+					}
+					return
+				default:
+				}
+				if version != shared.version {
+					break
+				}
+				shared.cond.Wait()
+			}
+
+			value := shared.value
+			version = shared.version
+			shared.cond.L.Unlock()
+
+			callback(WatchEvent{
+				Conn:  conn,
+				Key:   key,
+				Value: value,
+			})
+		}
+	}()
+
+	return &connWatcher{
+		shared: shared,
+		done:   done,
+	}, nil
+}