()
| 282 | } |
| 283 | |
| 284 | func (w *watcher) Next() (reader.Value, error) { |
| 285 | for { |
| 286 | s, err := w.lw.Next() |
| 287 | if err != nil { |
| 288 | return nil, err |
| 289 | } |
| 290 | |
| 291 | // only process changes |
| 292 | if bytes.Equal(w.value.Bytes(), s.ChangeSet.Data) { |
| 293 | continue |
| 294 | } |
| 295 | |
| 296 | v, err := w.rd.Values(s.ChangeSet) |
| 297 | if err != nil { |
| 298 | return nil, err |
| 299 | } |
| 300 | |
| 301 | return v.Get() |
| 302 | } |
| 303 | } |
| 304 | |
| 305 | func (w *watcher) Stop() error { |
| 306 | return w.lw.Stop() |