Method process

src/pkg/store/store.go:218–289

(ops <-chan Op, seqns chan<- int64, watches chan<- int)

}

func (st *Store) process(ops <-chan Op, seqns chan<- int64, watches chan<- int) {
	defer st.closeWatches()

	for {
		var flush bool
		ver, values := st.state.ver, st.state.root

		// Take any incoming requests and queue them up.
		select {
		case a, ok := <-ops:
			if !ok {
				return
			}

			if a.Seqn > ver {
				heap.Push(st.todo, a)
			}
		case w := <-st.watchCh:
			n, ws := w.rev, []*watch{w}
			for ; len(ws) > 0 && n < st.head; n++ {
				ws = []*watch{}
			}
			for ; len(ws) > 0 && n <= ver; n++ {
				ws = st.notify(st.log[n], ws)
			}

			st.watches = append(st.watches, ws...)
		case seqn := <-st.cleanCh:
			for ; st.head <= seqn; st.head++ {
				st.log[st.head] = Event{}, false
			}
		case seqns <- ver:
			// nothing to do here
		case watches <- len(st.watches):
			// nothing to do here
		case flush = <-st.flush:
			// nothing
		}

		var ev Event
		// If we have any mutations that can be applied, do them.
		for st.todo.Len() > 0 {
			t := st.todo.At(0).(Op)
			if flush && ver < t.Seqn {
				ver = t.Seqn - 1
			}
			if t.Seqn > ver+1 {
				break
			}

			heap.Pop(st.todo)
			if t.Seqn < ver+1 {
				continue
			}

			values, ev = values.apply(t.Seqn, t.Mut)
			st.state = &state{ev.Seqn, values}
			ver = ev.Seqn