Port implementation over racket ports.

.gitignore

@@ -15,3 +15,8 @@ compiled/
 # Dependency tracking files
 *.dep
 
+/scribblings/*.css
+/scribblings/*.js
+/*.bak
+/scribblings/*.bak
+/scribblings/*.html

README.md

@@ -1,3 +1,13 @@
 # port-channel
 
-Channels over ports for racket. 
+`port-channel` wraps a Racket input or output port as an asynchronous channel-like object.
+
+Input port-channels run a reader thread that reads serialized Racket values from the port,
+deserializes complete values and publishes them on an internal async channel. The
+`port-channel` itself is a synchronizable event, so `(sync pc)` returns the next complete value.
+
+Output port-channels expose `port-channel-put`, which queues a value immediately. A writer thread
+serializes and flushes queued values to the underlying output port.
+
+Values are encoded with `racket/serialize`. Custom structs must be declared with
+`serializable-struct`.

info.rkt

@@ -0,0 +1,11 @@
+#lang info
+
+(define collection "port-channel")
+(define pkg-desc "Asynchronous channels over Racket ports using racket/serialize.")
+(define version "0.1")
+(define pkg-authors '(hnmdijkema))
+(define scribblings '(("scribblings/port-channel.scrbl" ())))
+
+(define deps '("base"))
+(define build-deps '("racket-doc" "scribble-lib"))
+

main.rkt

@@ -0,0 +1,179 @@
+#lang racket/base
+
+(require racket/async-channel
+         racket/serialize)
+
+(provide port-channel?
+         port-channel-direction
+         port-channel-error?
+         port-channel-error-source
+         port-channel-error-message
+
+         make-port-channel
+         port-channel-put
+         port-channel-get
+         port-channel-try-get
+         port-channel-evt
+
+         port-channel-error-get
+         port-channel-error-try-get
+         port-channel-error-evt
+
+         close-port-channel
+         port-channel-wait)
+
+(struct port-channel-error (source message) #:transparent)
+
+(struct port-channel
+  (direction port channel errors thread close-token closing?)
+  #:property prop:evt
+  (lambda (pc)
+    (case (port-channel-direction pc)
+      [(input) (port-channel-channel pc)]
+      [else never-evt])))
+
+(define failed-token (gensym 'port-channel-failed))
+
+(define (put-error! source errors e)
+  (async-channel-put errors (port-channel-error source (exn-message e))))
+
+(define (read-serialized in errors source closing?)
+  (define serialized
+    (with-handlers ([exn:fail?
+                     (lambda (e)
+                       (unless (unbox closing?) (put-error! source errors e))
+                       eof)])
+      (read in)))
+  (cond
+    [(eof-object? serialized) eof]
+    [else
+     (with-handlers ([exn:fail?
+                      (lambda (e)
+                        (put-error! source errors e)
+                        failed-token)])
+       (deserialize serialized))]))
+
+(define (write-serialized out v errors source)
+  (define serialized
+    (with-handlers ([exn:fail?
+                     (lambda (e)
+                       (put-error! source errors e)
+                       failed-token)])
+      (serialize v)))
+  (cond
+    [(eq? serialized failed-token) #t]
+    [else
+     (with-handlers ([exn:fail?
+                      (lambda (e)
+                        (put-error! source errors e)
+                        #f)])
+       (write serialized out)
+       (newline out)
+       (flush-output out)
+       #t)]))
+
+(define (make-input-port-channel in source close?)
+  (define ch (make-async-channel))
+  (define errors (make-async-channel))
+  (define closing? (box #f))
+  (define close-token (gensym 'close-input-port-channel))
+  (define reader
+    (thread
+     (lambda ()
+       (let loop ()
+         (define v (read-serialized in errors source closing?))
+         (cond
+           [(eof-object? v)
+            (when close? (with-handlers ([exn:fail? void]) (close-input-port in)))
+            (async-channel-put ch eof)]
+           [(eq? v failed-token) (loop)]
+           [else
+            (async-channel-put ch v)
+            (loop)])))))
+  (port-channel 'input in ch errors reader close-token closing?))
+
+(define (make-output-port-channel out source close?)
+  (define ch (make-async-channel))
+  (define errors (make-async-channel))
+  (define closing? (box #f))
+  (define close-token (gensym 'close-output-port-channel))
+  (define writer
+    (thread
+     (lambda ()
+       (let loop ()
+         (define v (async-channel-get ch))
+         (cond
+           [(eq? v close-token)
+            (set-box! closing? #t)
+            (when close? (with-handlers ([exn:fail? void]) (close-output-port out)))]
+           [else
+            (define ok? (write-serialized out v errors source))
+            (when ok? (loop))])))))
+  (port-channel 'output out ch errors writer close-token closing?))
+
+(define (make-port-channel port
+                           #:direction [direction 'auto]
+                           #:source [source 'port]
+                           #:close? [close? #t])
+  (define in? (input-port? port))
+  (define out? (output-port? port))
+  (case direction
+    [(input)
+     (unless in? (raise-argument-error 'make-port-channel "input-port?" port))
+     (make-input-port-channel port source close?)]
+    [(output)
+     (unless out? (raise-argument-error 'make-port-channel "output-port?" port))
+     (make-output-port-channel port source close?)]
+    [(auto)
+     (cond
+       [(and in? (not out?)) (make-input-port-channel port source close?)]
+       [(and out? (not in?)) (make-output-port-channel port source close?)]
+       [(and in? out?)
+        (raise-arguments-error
+         'make-port-channel
+         "port is both input and output; use #:direction 'input or #:direction 'output"
+         "port" port)]
+       [else (raise-argument-error 'make-port-channel "port?" port)])]
+    [else
+     (raise-argument-error 'make-port-channel "(or/c 'auto 'input 'output)" direction)]))
+
+(define (port-channel-put pc v)
+  (unless (eq? (port-channel-direction pc) 'output)
+    (raise-argument-error 'port-channel-put "output port-channel?" pc))
+  (async-channel-put (port-channel-channel pc) v))
+
+(define (port-channel-get pc)
+  (unless (eq? (port-channel-direction pc) 'input)
+    (raise-argument-error 'port-channel-get "input port-channel?" pc))
+  (sync pc))
+
+(define (port-channel-try-get pc)
+  (unless (eq? (port-channel-direction pc) 'input)
+    (raise-argument-error 'port-channel-try-get "input port-channel?" pc))
+  (async-channel-try-get (port-channel-channel pc)))
+
+(define (port-channel-evt pc)
+  (unless (eq? (port-channel-direction pc) 'input)
+    (raise-argument-error 'port-channel-evt "input port-channel?" pc))
+  pc)
+
+(define (port-channel-error-get pc)
+  (async-channel-get (port-channel-errors pc)))
+
+(define (port-channel-error-try-get pc)
+  (async-channel-try-get (port-channel-errors pc)))
+
+(define (port-channel-error-evt pc)
+  (port-channel-errors pc))
+
+(define (close-port-channel pc)
+  (set-box! (port-channel-closing? pc) #t)
+  (case (port-channel-direction pc)
+    [(input)
+     (with-handlers ([exn:fail? void]) (close-input-port (port-channel-port pc)))]
+    [(output)
+     (async-channel-put (port-channel-channel pc) (port-channel-close-token pc))]
+    [else (void)]))
+
+(define (port-channel-wait pc)
+  (thread-wait (port-channel-thread pc)))

scribblings/port-channel.scrbl

@@ -0,0 +1,145 @@
+#lang scribble/manual
+
+@(require (for-label racket/base
+                     racket/contract
+                     racket/serialize
+                     "../main.rkt"))
+
+@title{port-channel}
+@author[@author+email["Hans Dijkema" "hans@dijkewijk.nl"]]
+
+@defmodule[port-channel]
+
+The @racketmodname[port-channel] module wraps a Racket port as an asynchronous,
+channel-like object. It is intended as a small building block for stream-based
+communication protocols, for example between a parent Racket process and a
+subprocess connected through standard input and standard output.
+
+A port-channel transports serialized Racket values. The implementation uses
+@racketmodname[racket/serialize]: values are passed through @racket[serialize],
+written to the port, read back as Racket data and reconstructed with
+@racket[deserialize]. Ordinary data values work directly. Custom structures must
+be defined with @racket[serializable-struct]. Ports, threads, custodians and
+ordinary procedures are not serializable values.
+
+@section{Model}
+
+A port-channel has one of two directions:
+
+@itemlist[
+ @item{@racket['input]: a reader thread reads complete serialized values from an
+       input port and puts the deserialized values on an internal asynchronous
+       channel. The port-channel is itself a synchronizable event, so
+       @racket[(sync pc)] returns the next complete value.}
+ @item{@racket['output]: @racket[port-channel-put] queues values on an internal
+       asynchronous channel. A writer thread serializes and writes queued values
+       to the output port.}]
+
+Buffered I/O alone is not enough to provide this behaviour. Synchronizing on a
+raw input port only says that some input is available; it does not guarantee that
+a complete serialized value is ready. The reader thread converts the byte stream
+into complete channel messages.
+
+@section{Reference}
+
+@defproc[(make-port-channel [port port?]
+                            [#:direction direction (or/c 'auto 'input 'output) 'auto]
+                            [#:source source any/c 'port]
+                            [#:close? close? any/c #t])
+         port-channel?]{
+Creates a port-channel for @racket[port]. With @racket['auto], the direction is
+inferred from the port. An input-only port becomes an input port-channel. An
+output-only port becomes an output port-channel. If the port is both an input and
+an output port, @racket[#:direction] must be supplied explicitly.
+
+The @racket[source] value is stored in reported @racket[port-channel-error]
+values. When @racket[close?] is true, @racket[close-port-channel] closes the
+underlying port.}
+
+@defproc[(port-channel? [v any/c]) boolean?]{
+Returns true when @racket[v] is a port-channel.}
+
+@defproc[(port-channel-direction [pc port-channel?]) (or/c 'input 'output)]{
+Returns the direction of @racket[pc].}
+
+@defproc[(port-channel-put [pc port-channel?] [v any/c]) void?]{
+Queues @racket[v] for writing to an output port-channel. The call only enqueues
+the value; serialization and port I/O are performed by the writer thread.
+
+The value must be serializable by @racket[serialize]. If serialization or writing
+fails, a @racket[port-channel-error] is published on the error channel.}
+
+@defproc[(port-channel-get [pc port-channel?]) any/c]{
+Returns the next value from an input port-channel. This is equivalent to
+@racket[(sync pc)]. When the underlying input stream reaches end-of-file,
+@racket[eof] is published as the final value.}
+
+@defproc[(port-channel-try-get [pc port-channel?]) any/c]{
+Attempts to get a value from an input port-channel without blocking. Returns
+@racket[#f] when no value is currently available.}
+
+@defproc[(port-channel-evt [pc port-channel?]) evt?]{
+Returns @racket[pc] as a synchronizable event. This is mainly a convenience for
+code that wants an explicit event-producing function.}
+
+@defproc[(close-port-channel [pc port-channel?]) void?]{
+Closes the port-channel. For an input port-channel, this closes the input port.
+For an output port-channel, a close marker is queued, so values already queued
+before the close marker are written first.}
+
+@defproc[(port-channel-wait [pc port-channel?]) void?]{
+Waits until the reader or writer thread of @racket[pc] has terminated.}
+
+@section{Errors}
+
+@defstruct*[port-channel-error ([source any/c]
+                                [message string?])]{
+Represents an error reported by the reader or writer thread.}
+
+@defproc[(port-channel-error-get [pc port-channel?]) port-channel-error?]{
+Blocks until the next error is available.}
+
+@defproc[(port-channel-error-try-get [pc port-channel?]) any/c]{
+Attempts to get the next error without blocking. Returns @racket[#f] when no
+error is currently available.}
+
+@defproc[(port-channel-error-evt [pc port-channel?]) evt?]{
+Returns the error channel as a synchronizable event.}
+
+@section{Example}
+
+@racketblock[
+(require port-channel)
+
+(define-values (in out) (make-pipe))
+
+(define reader (make-port-channel in))
+(define writer (make-port-channel out))
+
+(port-channel-put writer '(hello 1 2 3))
+(sync reader)
+]
+
+The result is:
+
+@racketblock[
+'(hello 1 2 3)
+]
+
+@section{Serializable structures}
+
+@racketblock[
+(require racket/serialize
+         port-channel)
+
+(serializable-struct message (id payload) #:transparent)
+
+(define-values (in out) (make-pipe))
+(define reader (make-port-channel in))
+(define writer (make-port-channel out))
+
+(port-channel-put writer (message 1 '(a b c)))
+(sync reader)
+]
+
+The received value is a reconstructed @racket[message] structure.

test/main.rkt

@@ -0,0 +1,25 @@
+#lang racket/base
+
+(require rackunit
+         racket/serialize
+         port-channel)
+
+(serializable-struct msg (id payload) #:transparent)
+
+(define-values (in out) (make-pipe))
+(define reader (make-port-channel in #:source 'test-reader))
+(define writer (make-port-channel out #:source 'test-writer))
+
+(port-channel-put writer '(hello 1 2 3))
+(port-channel-put writer (hasheq 'a 10 'b 20))
+(port-channel-put writer (msg 7 '(a b c)))
+
+(check-equal? (sync reader) '(hello 1 2 3))
+(check-equal? (sync reader) (hasheq 'a 10 'b 20))
+(check-equal? (sync reader) (msg 7 '(a b c)))
+
+(close-port-channel writer)
+(check-true (eof-object? (sync reader)))
+
+(module+ main
+  (displayln "port-channel tests ok"))