.

wcfs/wcfs.go

@@ -497,9 +497,9 @@ type Watcher struct {
 	fileTab map[*FileWatch]struct{}
 
 	// IO
-	acceptq chan string // (stream, msg)   // client-initiated messages go here
+//	acceptq chan string // (stream, msg)   // client-initiated messages go here
 	rxMu    sync.Mutex
-	rxTab   map[uint32]chan msg // client replies go via here
+	rxTab   map[uint32]chan string // client replies go via here
 }
 
 // FileWatch represents watching for 1 BigFile.
@@ -1250,7 +1250,7 @@ func (watch *Watch) Open(flags uint32, fctx *fuse.Context) (nodefs.File, fuse.St
 	// XXX check flags?
 	w := &Watcher{
 		sk:      NewFileSock(),
-		id:      atomic.AddInt32(&watch.id, +1)
+		id:      atomic.AddInt32(&watch.id, +1),
 		fileTab: make(map[*FileWatch]struct{}),
 	}
 
@@ -1286,32 +1286,58 @@ func (w *Watcher) _serve() (err error) {
 
 		fmt.Printf("watch: rx: %q\n", l)
 
-		// <stream> ...
-		var req string
-		n, err := fmt.Sscanf(l, "%d %s\n", &stream, &req)
-		if err == nil && n != 2 {
-			err = fmt.Errorf("invalid frame: %q", l)
+		// <stream> <msg...>
+		sp := strings.IndexByte(l, ' ')
+		if sp == -1 {
+			// XXX write to peer too? (on which stream? -1?)
+			return fmt.Errorf("rx: invalid frame: %q", l)
 		}
+
+		stream, err := strconv.ParseUint(l[:sp], 10, 64)
 		if err != nil {
-			return fmt.Errorf("rx: %s", err)
+			return fmt.Errorf("rx: invalid frame (stream): %q", l)
 		}
 
+		msg := l[sp+1:]
+
+		// reply from client to to wcfs
 		reply := (stream % 2 == 0)
-		// reply to wcfs message
 		if reply {
 			w.rxMu.Lock()
 			rxq := w.rxTab[stream]
+			delete(w.rxTab, stream)
 			w.rxMu.Unlock()
 
 			if rxq == nil {
-				return fmt.Errorf("rx: reply on unexpected stream %d", stream)
+				return fmt.Errorf("rx %d: reply on unexpected streamd", stream)
+			}
+			rxq <- msg
+			continue
 		}
-			rxq <- req
 
-		// client-initiated message
-		} else {
-			fmt.Sscanf(req, "watch %s %s\n", &oid, &ats
+		// client-initiated request
+		msg = strings.TrimSuffix(msg, "\n")
+		msgv = strings.Split(msg, " ")
+		if !(len(msgv) == 3 && msgv[0] != "watch") {
+			// XXX write to peer too
+			return fmt.Errorf("rx %d: invalid request", stream)
+		}
+		oid, err := zodb.ParseOid(msgv[1])
+		if err != nil {
+			return fmt.Errorf("rx %d: bad watch: invalid oid")
+		}
 
+		var at zodb.Tid
+		switch {
+		case msgv[2] == "-":
+			at = zodb.InvalidTid
+		case strings.HasPrefix(msgv[2], "@"):
+			at, err = zodb.ParseTid(msgv[2][1:])
+		default:
+			err = fmt.Errorf("x") // XXX just anything
+		}
+		if err != nil {
+			return fmt.Errorf("rx %d: bad watch: invalid at")
 		}
 
 	}

wcfs/wcfs_test.py

@@ -408,7 +408,7 @@ class tWatch:
             if reply:
                 with t._rxmu:
                     assert stream in t._rxtab
-                    rxq = t._rxtabs[stream]
+                    rxq = t._rxtab.pop(stream)
                 rxq.send(msg)
             else:
                 with t._rxmu: