.

go/zodb/open.go

@@ -226,6 +226,7 @@ func (s *storage) watcher() {
 
 			// deliver event to all watchers
 			for watchq := range s.watchTab {
+				// XXX + select and handle DelWatch
 				watchq <- event
 			}
 		}

go/zodb/zodb.go

@@ -444,20 +444,38 @@ type Watcher interface {
 	// Whenever a new transaction is committed into the database,
 	// corresponding event will be sent to watchq.
 	//
+	// It is guaranteed that events are coming with ↑ .Tid .
+	//
 	// It will be only and all events in (at₀, +∞] range, that will be
-	// sent, where at₀ is database head that was current when AddWatch call was made. XXX
+	// sent, where at₀ is database head that was current when AddWatch call
+	// was made.
 	//
-	// Once registered, watchq must be read. Not doing so will stuck whole storage.
+	// Once registered, watchq must be read until DelWatch call.
+	// Not doing so will stuck whole storage.
 	//
 	// Multiple AddWatch calls with the same watchq register watchq only once.	XXX
 	//
-	// XXX ↑ guaranteed
 	// XXX watchq closed when stor.watchq closed?
 	AddWatch(watchq chan<- CommitEvent) (at0 Tid)
 
 	// DelWatch unregisters watchq from being notified of database changes.
 	//
 	// After DelWatch call completes, no new events will be sent to watchq.
+	// It is safe to call DelWatch without sumultaneously reading watchq.
+	// In particular the following example is valid:
+	//
+	//	at0 := stor.AddWatch(watchq)
+	//	defer stor.DelWatch(watchq)
+	//
+	//	for {
+	//		select {
+	//		case <-ctx.Done():
+	//			return ctx.Err()
+	//
+	//		case <-watchq:
+	//			...
+	//		}
+	//	}
 	//
 	// DelWatch is noop if watchq was not registered.
 	//

go/zodb/zodbtools/main.go

@@ -30,6 +30,7 @@ var commands = prog.CommandRegistry{
 	{"info", infoSummary, infoUsage, infoMain},
 	{"dump", dumpSummary, dumpUsage, dumpMain},
 	{"catobj", catobjSummary, catobjUsage, catobjMain},
+	{"watch", watchSummary, watchUsage, watchMain},
 }
 
 // main zodbtools driver

go/zodb/zodbtools/watch.go

+// Copyright (C) 2019  Nexedi SA and Contributors.
+//                     Kirill Smelkov <kirr@nexedi.com>
+//
+// This program is free software: you can Use, Study, Modify and Redistribute
+// it under the terms of the GNU General Public License version 3, or (at your
+// option) any later version, as published by the Free Software Foundation.
+//
+// You can also Link and Combine this program with other software covered by
+// the terms of any of the Free Software licenses or any of the Open Source
+// Initiative approved licenses and Convey the resulting work. Corresponding
+// source of such a combination shall include the source code for all other
+// software used.
+//
+// This program is distributed WITHOUT ANY WARRANTY; without even the implied
+// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+//
+// See COPYING file for full licensing terms.
+// See https://www.nexedi.com/licensing for rationale and options.
+
+// Zodbwatch - watch database for changes
+//
+// Zodbwatch watches deatbase for changes and prints information about
+// committed transactions. Output formats:
+//
+// Plain:
+//
+//	# at <tid>
+//	txn <tid>
+//	txn <tid>
+//	...
+//
+// Verbose:
+//
+//	# at <tid>
+//	txn <tid>
+//	obj <oid>
+//	obj ...
+//	...
+//	LF
+//	txn <tid>
+//	...
+//
+// TODO add support for emitting transaction in zodbdump format.
+
+package zodbtools
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+
+	"lab.nexedi.com/kirr/go123/prog"
+	"lab.nexedi.com/kirr/go123/xerr"
+	"lab.nexedi.com/kirr/neo/go/zodb"
+)
+
+// Watch watches for database changes and prints them to w.
+//
+// see top-level documentation for output format.
+func Watch(ctx context.Context, stor zodb.IStorage, w io.Writer, verbose bool) (err error) {
+	defer xerr.Contextf(&err, "%s: watch", stor.URL())
+
+	emitf := func(format string, argv ...interface{}) error {
+		_, err := fmt.Fprintf(w, format, argv)
+		return err
+	}
+
+	watchq := make(chan zodb.CommitEvent)
+	at0 := stor.AddWatch(watchq)
+	defer stor.DelWatch(watchq)
+
+	err = emitf("# at %s\n", at0)
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
+		case δ, ok := <-watchq:
+			if !ok {
+				// XXX correct?
+				err = emitf("# storage closed")
+				return err
+			}
+
+			err = emitf("txn %s\n", δ.Tid)
+			if err != nil {
+				return err
+			}
+
+			if verbose {
+				for _, oid := range δ.Changev {
+					err = emitf("obj %s\n", oid)
+					if err != nil {
+						return err
+					}
+				}
+
+				err = emitf("\n")
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+}
+
+// ----------------------------------------
+const watchSummary = "watch ZODB database for changes"
+
+func watchUsage(w io.Writer) {
+	fmt.Fprintf(w,
+`Usage: zodb watch [OPTIONS] <storage>
+Watch ZODB database for changes.
+
+<storage> is an URL (see 'zodb help zurl') of a ZODB-storage.
+
+Options:
+
+	-h --help       this help text.
+	-v		verbose mode.
+`)
+}
+
+func watchMain(argv []string) {
+	verbose := false
+	flags := flag.FlagSet{Usage: func() { watchUsage(os.Stderr) }}
+	flags.Init("", flag.ExitOnError)
+	flags.BoolVar(&verbose, "v", verbose, "verbose mode")
+	flags.Parse(argv[1:])
+
+	argv = flags.Args()
+	if len(argv) != 1 {
+		flags.Usage()
+		prog.Exit(2)
+	}
+	zurl := argv[0]
+
+	ctx := context.Background()
+
+	stor, err := zodb.OpenStorage(ctx, zurl, &zodb.OpenOptions{ReadOnly: true})
+	if err != nil {
+		prog.Fatal(err)
+	}
+	// TODO defer stor.Close()
+
+	err = Watch(ctx, stor, os.Stdout, verbose)
+	if err != nil { // XXX & not canceled
+		prog.Fatal(err)
+	}
+}

go/zodb/zodbtools/watch_test.go

+// Copyright (C) 2019  Nexedi SA and Contributors.
+//                     Kirill Smelkov <kirr@nexedi.com>
+//
+// This program is free software: you can Use, Study, Modify and Redistribute
+// it under the terms of the GNU General Public License version 3, or (at your
+// option) any later version, as published by the Free Software Foundation.
+//
+// You can also Link and Combine this program with other software covered by
+// the terms of any of the Free Software licenses or any of the Open Source
+// Initiative approved licenses and Convey the resulting work. Corresponding
+// source of such a combination shall include the source code for all other
+// software used.
+//
+// This program is distributed WITHOUT ANY WARRANTY; without even the implied
+// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+//
+// See COPYING file for full licensing terms.
+// See https://www.nexedi.com/licensing for rationale and options.
+
+package zodbtools
+
+import (
+	"testing"
+)
+
+
+func TestWatch(t *testing.T) {
+	// XXX
+}