.

wcfs/internal/wcfs_misc.cpp

@@ -57,6 +57,37 @@ error File::close() {
     return nil;
 }
 
+tuple<int, error> File::read(void *buf, size_t count) {
+    File *f = this;
+    int n;
+
+    n = ::read(f->_fd, buf, count);
+    if (n == 0)
+        return make_tuple(n, io::EOF);
+    if (n < 0)
+        return make_tuple(0, f->_errno("read"));
+
+    return make_tuple(n, nil);
+}
+
+tuple <int, error> File::write(const void *buf, size_t count) {
+    File *f = this;
+    int n, wrote=0;
+
+    // NOTE contrary to write(2) we have to write all data as io.Writer requires.
+    while (count != 0) {
+        n = ::write(f->_fd, buf, count);
+        if (n < 0)
+            return make_tuple(wrote, f->_errno("write"));
+
+        wrote += n;
+        buf   += n;
+        count -= n;
+    }
+
+    return make_tuple(wrote, nil);
+}
+
 error File::stat(struct stat *st) {
     File *f = this;
 

wcfs/internal/wcfs_misc.h

@@ -66,6 +66,19 @@ public:
     int     fd()    const;
     string  name()  const;
     error   close();
+
+    // read implements io.Reader from Go: it reads into buf up-to count bytes.
+    //
+    // XXX buf -> slice<byte> ?
+    tuple<int, error> read(void *buf, size_t count);
+
+    // write implements io.Writer from Go: it writes all data from buf.
+    //
+    // NOTE write behaves like io.Writer in Go - it tries to write as much
+    // bytes as requested, and if it could write only less - it returns error.
+    // XXX buf -> slice<byte> ?
+    tuple<int, error> write(const void *buf, size_t count);
+
     error   stat(struct stat *st);
 
 private:

wcfs/internal/wcfs_virtmem.cpp

@@ -334,6 +334,20 @@ error _Mapping::_remmapblk(int64_t blk, Tid at) {
 }
 
 
+// ---- WatchLink ----
+
+// Pkt internally represents data of one message sent/received over WatchLink.
+struct Pkt {
+    // stream over which the data was received; used internally by send
+    StreamID stream;
+
+    // raw data received/to-be-sent.
+    // XXX not e.g. string as chan<T> currently does not support types with
+    //     non-trivial copy. Note: we anyway need to limit line length to avoid DoS
+    //     but just for DoS the limit would be higher.
+    char     rawdata[128-sizeof(StreamID)];
+};
+
 // _openwatch opens new watch link on wcfs.
 tuple<WatchLink*, error> WCFS::_openwatch() {
     WCFS *wc = this;
@@ -364,6 +378,36 @@ tuple<WatchLink*, error> WCFS::_openwatch() {
 
 // XXX close
 
+// _send sends raw message via specified stream.
+//
+// multiple _send can be called in parallel - _send serializes writes.
+// XXX +ctx?
+def WatchLink::_send(StreamID stream, const string &msg) {
+    WatchLink *wlink = this;
+    if (msg.find('\n') != string::npos)
+        panic("msg has \\n");
+    Pkt pkt;
+    pkt.stream  = stream;
+    pkt.rawdata = // XXX copy msg   XXX + "%d <stream> ...\n" ?
+    //pkt = b"%d %s\n" % (stream, msg)
+    wlink->_write(&pkt)
+}
+
+def WatchLink::_write(Pkt *pkt) {
+    WatchLink *wlink = this;
+    wlink->_txmu.lock();
+    defer([&]() {
+        wlink->_txmu.unlock();
+    })
+
+    //printf('C: watch  : tx: %r' % pkt)
+    wlink->_f.write(pkt);   // XXX all data
+#if 0
+    wlink->_wtx.write(pkt);
+    wlink->_wtx.flush();
+#endif
+}
+
 // ---- WCFS raw file access ----
 
 // _path returns path for object on wcfs.