ocapn: captp: Switch to netlayers providing messages one at a time

Previously the netlayer would provide i/o ports, but this is less
general than providing entirely parsed messages.

goblins/ocapn/captp.rkt

@@ -1189,8 +1189,8 @@
       (spawn ^hash))
 
     (define (^connection-establisher bcom netlayer netlayer-name)
-      (lambda (ip op incoming?)
-        ($ self 'new-connection netlayer netlayer-name ip op)))
+      (lambda (read-message write-message incoming?)
+        ($ self 'new-connection netlayer netlayer-name read-message write-message)))
 
     ;; Here's why all netlayers currently are in the same vat as the
     ;; router, at least currently... to make sure that they're all set up
@@ -1363,7 +1363,7 @@
      ;; somewhere...
      ;; TODO: Should this still be an exposed method?  Maybe it's something only
      ;; the ^connection-establisher should call...
-     [(new-connection netlayer netlayer-name network-in-port network-out-port)
+     [(new-connection netlayer netlayer-name read-message write-message)
       (define captp-outgoing-ch
         (make-async-channel))
       (define (send-to-remote msg)
@@ -1462,7 +1462,7 @@
       (syscaller-free-thread
        (lambda ()
          (let lp ()
-           (match (syrup-read network-in-port #:unmarshallers unmarshallers)
+           (match (read-message unmarshallers)
              [(? eof-object?)
               ;; (displayln "Shutting down captp session...")
               (<-np-extern incoming-forwarder
@@ -1476,10 +1476,7 @@
          (let lp ()
            (define msg
              (async-channel-get captp-outgoing-ch))
-           (syrup-write msg network-out-port #:marshallers marshallers)
-           ;; TODO: *should* we be flushing output each time we've written out
-           ;; a message?  It seems like "yes" but I'm a bit unsure
-           (flush-output network-out-port)
+           (write-message msg marshallers)
            (lp))))
 
       ;; Now we'll need to send our side of the start-session and get the

goblins/ocapn/netlayer/fake-intarwebs.rkt

@@ -20,6 +20,7 @@
          "../../actor-lib/methods.rkt"
          "../../actor-lib/sync-pr.rkt"
          "../structs-urls.rkt"
+         "utils/read-write-procs.rkt"
          racket/async-channel
          racket/match)
 
@@ -52,8 +53,10 @@
       (on (sync/pr new-conn-ch)
           (match-lambda
             [(list (vector ip op))
+             (define-values (read-message write-message)
+               (read-write-procs ip op))
              ;; TODO
-             (<- conn-establisher ip op #t)])
+             (<- conn-establisher read-message write-message #t)])
           #:finally listen-loop)))
 
   (define (^netlayer bcom)
@@ -84,7 +87,9 @@
            (on (<- network 'connect-to name)
                (match-lambda
                  [(vector ip op)
-                  (<- conn-establisher ip op #f)])
+                  (define-values (read-message write-message)
+                    (read-write-procs ip op))
+                  (<- conn-establisher read-message write-message #f)])
                #:promise? #t)])]))
     ;; TODO: return pre-setup-beh
     pre-setup-beh)

goblins/ocapn/netlayer/onion.rkt

@@ -21,6 +21,7 @@
          "../../actor-lib/methods.rkt"
          "../structs-urls.rkt"
          "onion-socks.rkt"
+         "utils/read-write-procs.rkt"
          racket/match
          racket/unix-socket
          racket/async-channel

goblins/ocapn/netlayer/utils/read-write-procs.rkt

@@ -0,0 +1,13 @@
+#lang racket/base
+
+(provide read-write-procs)
+
+(require syrup)
+
+(define (read-write-procs ip op)
+  (define (read-message unmarshallers)
+    (syrup-read ip #:unmarshallers unmarshallers))
+  (define (write-message msg marshallers)
+    (syrup-write msg op #:marshallers marshallers)
+    (flush-output op))
+  (values read-message write-message))