1// Copyright 2015 CoreOS, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15package storage
16
17import (
18	"errors"
19	"sync"
20
21	"github.com/coreos/etcd/storage/storagepb"
22)
23
24var (
25	ErrWatcherNotExist = errors.New("storage: watcher does not exist")
26)
27
28type WatchID int64
29
30type WatchStream interface {
31	// Watch creates a watcher. The watcher watches the events happening or
32	// happened on the given key or range [key, end) from the given startRev.
33	//
34	// The whole event history can be watched unless compacted.
35	// If `startRev` <=0, watch observes events after currentRev.
36	//
37	// The returned `id` is the ID of this watcher. It appears as WatchID
38	// in events that are sent to the created watcher through stream channel.
39	//
40	Watch(key, end []byte, startRev int64) WatchID
41
42	// Chan returns a chan. All watch response will be sent to the returned chan.
43	Chan() <-chan WatchResponse
44
45	// RequestProgress requests the progress of the watcher with given ID. The response
46	// will only be sent if the watcher is currently synced.
47	// The responses will be sent through the WatchRespone Chan attached
48	// with this stream to ensure correct ordering.
49	// The responses contains no events. The revision in the response is the progress
50	// of the watchers since the watcher is currently synced.
51	RequestProgress(id WatchID)
52
53	// Cancel cancels a watcher by giving its ID. If watcher does not exist, an error will be
54	// returned.
55	Cancel(id WatchID) error
56
57	// Close closes Chan and release all related resources.
58	Close()
59
60	// Rev returns the current revision of the KV the stream watches on.
61	Rev() int64
62}
63
64type WatchResponse struct {
65	// WatchID is the WatchID of the watcher this response sent to.
66	WatchID WatchID
67
68	// Events contains all the events that needs to send.
69	Events []storagepb.Event
70
71	// Revision is the revision of the KV when the watchResponse is created.
72	// For a normal response, the revision should be the same as the last
73	// modified revision inside Events. For a delayed response to a unsynced
74	// watcher, the revision is greater than the last modified revision
75	// inside Events.
76	Revision int64
77
78	// CompactRevision is set when the watcher is cancelled due to compaction.
79	CompactRevision int64
80}
81
82// watchStream contains a collection of watchers that share
83// one streaming chan to send out watched events and other control events.
84type watchStream struct {
85	watchable watchable
86	ch        chan WatchResponse
87
88	mu sync.Mutex // guards fields below it
89	// nextID is the ID pre-allocated for next new watcher in this stream
90	nextID   WatchID
91	closed   bool
92	cancels  map[WatchID]cancelFunc
93	watchers map[WatchID]*watcher
94}
95
96// Watch creates a new watcher in the stream and returns its WatchID.
97// TODO: return error if ws is closed?
98func (ws *watchStream) Watch(key, end []byte, startRev int64) WatchID {
99	ws.mu.Lock()
100	defer ws.mu.Unlock()
101	if ws.closed {
102		return -1
103	}
104
105	id := ws.nextID
106	ws.nextID++
107
108	w, c := ws.watchable.watch(key, end, startRev, id, ws.ch)
109
110	ws.cancels[id] = c
111	ws.watchers[id] = w
112	return id
113}
114
115func (ws *watchStream) Chan() <-chan WatchResponse {
116	return ws.ch
117}
118
119func (ws *watchStream) Cancel(id WatchID) error {
120	cancel, ok := ws.cancels[id]
121	if !ok {
122		return ErrWatcherNotExist
123	}
124	cancel()
125	delete(ws.cancels, id)
126	delete(ws.watchers, id)
127	return nil
128}
129
130func (ws *watchStream) Close() {
131	ws.mu.Lock()
132	defer ws.mu.Unlock()
133
134	for _, cancel := range ws.cancels {
135		cancel()
136	}
137	ws.closed = true
138	close(ws.ch)
139	watchStreamGauge.Dec()
140}
141
142func (ws *watchStream) Rev() int64 {
143	ws.mu.Lock()
144	defer ws.mu.Unlock()
145	return ws.watchable.rev()
146}
147
148func (ws *watchStream) RequestProgress(id WatchID) {
149	ws.mu.Lock()
150	w, ok := ws.watchers[id]
151	ws.mu.Unlock()
152	if !ok {
153		return
154	}
155	ws.watchable.progress(w)
156}
157