From f0c4f5a4ce1f6082203a729580f1e4948714f145 Mon Sep 17 00:00:00 2001 From: Hans Dijkema Date: Mon, 8 Jun 2026 14:21:36 +0200 Subject: [PATCH] sync tested. --- .gitignore | 5 + README.md | 58 +++++-- info.rkt | 15 +- main.rkt | 310 ++++++++-------------------------- scribblings/uni-channel.scrbl | 241 ++++++++++++++------------ test/main.rkt | 129 ++++++++------ 6 files changed, 331 insertions(+), 427 deletions(-) diff --git a/.gitignore b/.gitignore index 3bd7ab8..aef8e70 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,8 @@ compiled/ *.dep /*.bak + +/scribblings/*.css +/scribblings/*.js +/scribblings/*.bak +/scribblings/*.html diff --git a/README.md b/README.md index 126d277..449d82f 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,46 @@ # uni-channel -`uni-channel` provides a uniform facade over `async-channel`, `place-channel` and `port-channel`. +`uni-channel` provides one simple wrapper over `async-channel`, `place-channel` +and `port-channel`. -The package does not vendor `port-channel`; install `port-channel` from the Racket package index first. +```racket +(make-uni-channel channel) +``` + +The constructor detects the concrete channel kind and returns a `uni-channel?`. +No helper constructors are part of the public API; create the underlying channel +with the normal Racket or `port-channel` constructor and then wrap it. + +A wrapped channel that can receive is itself synchronizable. `(sync ch)` returns +the value received from the underlying channel. For an output-only endpoint, such +as an output `port-channel`, the receive event is `never-evt` because no value can +be received from that endpoint. + + +```racket +(require racket/async-channel + uni-channel) + +(define ch (make-uni-channel (make-async-channel))) +(uni-channel-send ch 'hello) +(uni-channel-recv ch) +``` + +For `port-channel`, wrap the input and output endpoints separately: + +```racket +(require port-channel + uni-channel) + +(define-values (in out) (make-pipe)) +(define reader (make-uni-channel (make-port-channel in))) +(define writer (make-uni-channel (make-port-channel out))) + +(uni-channel-send writer '(hello 1 2 3)) +(uni-channel-recv reader) +``` + +Install and test: ```sh raco pkg install --auto port-channel @@ -10,19 +48,3 @@ raco pkg install --auto ./uni-channel raco test -x -p uni-channel raco setup --pkgs uni-channel ``` - -Main API: - -```racket -make-uni-channel -make-uni-async-channel -make-uni-async-channel-pair -make-uni-place-channel -make-uni-place-channel-pair -make-uni-port-channel -uni-channel-put -uni-channel-get -uni-channel-send -uni-channel-recv -uni-channel-close -``` diff --git a/info.rkt b/info.rkt index ca35a59..978a695 100644 --- a/info.rkt +++ b/info.rkt @@ -2,20 +2,13 @@ (define collection "uni-channel") (define pkg-authors '(hnmdijkema)) -(define version "0.1") +(define version "0.2") (define license '(Apache-2.0 OR MIT)) -(define pkg-desc "Uniform channel facade for place-channel, async-channel and port-channel.") - -(define scribblings - '( - ("scribblings/uni-channel.scrbl" () (library)) - ) - ) +(define pkg-desc "Uniform channel wrapper for place-channel, async-channel and port-channel.") (define deps '("base" "port-channel")) (define build-deps '("racket-doc" "scribble-lib" "rackunit-lib")) - -;; Keep package installation from compiling the tests as runtime modules. -;; `rackunit-lib` is a build/test dependency, not a library dependency. (define compile-omit-paths '("test")) +(define scribblings + '(("scribblings/uni-channel.scrbl" () (library)))) diff --git a/main.rkt b/main.rkt index ca377fb..a55f47c 100644 --- a/main.rkt +++ b/main.rkt @@ -7,15 +7,8 @@ (provide uni-channel? uni-channel-kind uni-channel-direction - uni-channel-name - uni-channel-impl make-uni-channel - make-uni-async-channel - make-uni-async-channel-pair - make-uni-place-channel - make-uni-place-channel-pair - make-uni-port-channel uni-channel-put uni-channel-get @@ -32,50 +25,31 @@ uni-channel-error-evt) (struct uni-channel - (kind impl direction name get-proc put-proc try-get-proc get-evt-proc put-evt-proc + (kind impl direction get-proc put-proc try-get-proc get-evt-proc put-evt-proc close-proc closed-box error-get-proc error-try-get-proc error-evt-proc) #:property prop:evt (lambda (ch) - (cond - [(and (can-get? ch) (uni-channel-get-proc ch)) - ((uni-channel-get-evt-proc ch) ch)] - [else never-evt]))) + (if (can-get? ch) ((uni-channel-get-evt-proc ch) ch) never-evt))) -(define (check-direction who v) - (case v - [(input output bidirectional) v] - [else (raise-argument-error who "(or/c 'input 'output 'bidirectional)" v)])) - -(define (check-kind who v) - (case v - [(async place port custom) v] - [else (raise-argument-error who "(or/c 'async 'place 'port 'custom)" v)])) - -(define (can-get? ch) - (not (eq? (uni-channel-direction ch) 'output))) - -(define (can-put? ch) - (not (eq? (uni-channel-direction ch) 'input))) +(define (can-get? ch) (not (eq? (uni-channel-direction ch) 'output))) +(define (can-put? ch) (not (eq? (uni-channel-direction ch) 'input))) (define (raise-no-get who ch) (raise-arguments-error who "uni-channel does not support receiving" - "name" (uni-channel-name ch) + "kind" (uni-channel-kind ch) "direction" (uni-channel-direction ch))) (define (raise-no-put who ch) (raise-arguments-error who "uni-channel does not support sending" - "name" (uni-channel-name ch) + "kind" (uni-channel-kind ch) "direction" (uni-channel-direction ch))) (define (raise-closed who ch) (raise-arguments-error who "uni-channel is closed" - "name" (uni-channel-name ch) - "kind" (uni-channel-kind ch))) + "kind" (uni-channel-kind ch) + "direction" (uni-channel-direction ch))) -(define (closed-box) (box #f)) - -(define (mark-closed! ch) - (set-box! (uni-channel-closed-box ch) #t)) +(define (mark-closed! ch) (set-box! (uni-channel-closed-box ch) #t)) (define (default-close ch) (mark-closed! ch) @@ -83,16 +57,14 @@ (define (no-error-get ch) (raise-arguments-error 'uni-channel-error-get "uni-channel has no error channel" - "name" (uni-channel-name ch) "kind" (uni-channel-kind ch))) (define (no-error-try-get ch) #f) (define (no-error-evt ch) never-evt) -(define (make-oc #:kind kind - #:backend backend +(define (make-uc #:kind kind + #:impl impl #:direction direction - #:name name #:get get-proc #:put put-proc #:try-get try-get-proc @@ -102,234 +74,94 @@ #:error-get error-get-proc #:error-try-get error-try-get-proc #:error-evt error-evt-proc) - (uni-channel kind backend direction name get-proc put-proc try-get-proc get-evt-proc put-evt-proc - close-proc (closed-box) error-get-proc error-try-get-proc error-evt-proc)) + (uni-channel kind impl direction get-proc put-proc try-get-proc get-evt-proc put-evt-proc + close-proc (box #f) error-get-proc error-try-get-proc error-evt-proc)) -(define (make-uni-channel backend - #:kind [kind 'auto] - #:direction [direction 'auto] - #:name [name #f]) - (define detected-kind - (case kind - [(auto) - (cond - [(async-channel? backend) 'async] - [(place-channel? backend) 'place] - [(port-channel? backend) 'port] - [else (raise-argument-error 'make-uni-channel - "(or/c async-channel? place-channel? port-channel?)" - backend)])] - [else (check-kind 'make-uni-channel kind)])) - (case detected-kind - [(async) - (unless (async-channel? backend) - (raise-argument-error 'make-uni-channel "async-channel?" backend)) - (make-uni-async-channel backend #:direction direction #:name (or name 'async-channel))] - [(place) - (unless (place-channel? backend) - (raise-argument-error 'make-uni-channel "place-channel?" backend)) - (make-uni-place-channel backend #:direction direction #:name (or name 'place-channel))] - [(port) - (unless (port-channel? backend) - (raise-argument-error 'make-uni-channel "port-channel?" backend)) - (define pc-dir (port-channel-direction backend)) - (define dir (if (eq? direction 'auto) pc-dir (check-direction 'make-uni-channel direction))) - (unless (eq? pc-dir dir) - (raise-arguments-error 'make-uni-channel "direction does not match port-channel direction" - "requested direction" dir - "port-channel direction" pc-dir)) - (make-port-oc (and (eq? pc-dir 'input) backend) - (and (eq? pc-dir 'output) backend) - #:direction pc-dir - #:name (or name 'port-channel))] - [else - (raise-argument-error 'make-uni-channel "(or/c 'auto 'async 'place 'port)" kind)])) +(define (make-uni-channel channel) + (cond + [(async-channel? channel) (make-async-uc channel)] + [(place-channel? channel) (make-place-uc channel)] + [(port-channel? channel) (make-port-uc channel)] + [else (raise-argument-error 'make-uni-channel + "(or/c async-channel? place-channel? port-channel?)" + channel)])) -(define (make-uni-async-channel [ch (make-async-channel)] - #:direction [direction 'bidirectional] - #:name [name 'async-channel]) - (define dir (if (eq? direction 'auto) 'bidirectional (check-direction 'make-uni-async-channel direction))) - (define backend ch) - (make-oc #:kind 'async - #:backend backend - #:direction dir - #:name name - #:get (lambda (_ch) (async-channel-get backend)) - #:put (lambda (_ch v) (async-channel-put backend v)) - #:try-get (lambda (_ch) (async-channel-try-get backend)) - #:get-evt (lambda (_ch) backend) - #:put-evt (lambda (_ch v) (async-channel-put-evt backend v)) +(define (make-async-uc ch) + (make-uc #:kind 'async + #:impl ch + #:direction 'bidirectional + #:get (lambda (_ch) (async-channel-get ch)) + #:put (lambda (_ch v) (async-channel-put ch v)) + #:try-get (lambda (_ch) (async-channel-try-get ch)) + #:get-evt (lambda (_ch) ch) + #:put-evt (lambda (_ch v) (async-channel-put-evt ch v)) #:close default-close #:error-get no-error-get #:error-try-get no-error-try-get #:error-evt no-error-evt)) -(define (make-uni-async-channel-pair #:name-a [name-a 'async-channel-a] - #:name-b [name-b 'async-channel-b]) - (define a-in (make-async-channel)) - (define b-in (make-async-channel)) - (define (endpoint name incoming outgoing) - (make-oc #:kind 'async - #:backend (vector 'async-channel-pair incoming outgoing) - #:direction 'bidirectional - #:name name - #:get (lambda (_ch) (async-channel-get incoming)) - #:put (lambda (_ch v) (async-channel-put outgoing v)) - #:try-get (lambda (_ch) (async-channel-try-get incoming)) - #:get-evt (lambda (_ch) incoming) - #:put-evt (lambda (_ch v) (async-channel-put-evt outgoing v)) - #:close default-close - #:error-get no-error-get - #:error-try-get no-error-try-get - #:error-evt no-error-evt)) - (values (endpoint name-a a-in b-in) (endpoint name-b b-in a-in))) - -(define (make-uni-place-channel ch - #:direction [direction 'bidirectional] - #:name [name 'place-channel]) - (define dir (if (eq? direction 'auto) 'bidirectional (check-direction 'make-uni-place-channel direction))) - (define backend ch) - (make-oc #:kind 'place - #:backend backend - #:direction dir - #:name name - #:get (lambda (_ch) (place-channel-get backend)) - #:put (lambda (_ch v) (place-channel-put backend v)) - #:try-get (lambda (_ch) (sync/timeout 0 backend)) - #:get-evt (lambda (_ch) backend) - #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (place-channel-put backend v)))) +(define (make-place-uc ch) + (make-uc #:kind 'place + #:impl ch + #:direction 'bidirectional + #:get (lambda (_ch) (place-channel-get ch)) + #:put (lambda (_ch v) (place-channel-put ch v)) + #:try-get (lambda (_ch) (sync/timeout 0 ch)) + #:get-evt (lambda (_ch) ch) + #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (place-channel-put ch v)))) #:close default-close #:error-get no-error-get #:error-try-get no-error-try-get #:error-evt no-error-evt)) -(define (make-uni-place-channel-pair #:name-a [name-a 'place-channel-a] - #:name-b [name-b 'place-channel-b]) - (define-values (a b) (place-channel)) - (values (make-uni-place-channel a #:name name-a) - (make-uni-place-channel b #:name name-b))) - -(define (make-port-oc in-pc out-pc #:direction direction #:name name) - (define backend - (cond - [(and in-pc out-pc) (vector 'port-channel-pair in-pc out-pc)] - [in-pc in-pc] - [out-pc out-pc] - [else (error 'make-port-oc "internal error: no port-channel backend")])) - (define (get* _ch) (port-channel-get in-pc)) - (define (put* _ch v) (port-channel-put out-pc v)) - (define (try-get* _ch) (port-channel-try-get in-pc)) - (define (get-evt* _ch) (port-channel-evt in-pc)) - (define (put-evt* _ch v) (handle-evt always-evt (lambda (_) (port-channel-put out-pc v)))) - (define (close* ch) - (mark-closed! ch) - (cond - [(and in-pc out-pc) (close-port-channel out-pc)] - [in-pc (close-port-channel in-pc)] - [out-pc (close-port-channel out-pc)]) - (void)) - (define (error-get* _ch) - (cond - [in-pc (port-channel-error-get in-pc)] - [out-pc (port-channel-error-get out-pc)])) - (define (error-try-get* _ch) - (or (and in-pc (port-channel-error-try-get in-pc)) - (and out-pc (port-channel-error-try-get out-pc)))) - (define (error-evt* _ch) - (cond - [(and in-pc out-pc) (choice-evt (port-channel-error-evt in-pc) (port-channel-error-evt out-pc))] - [in-pc (port-channel-error-evt in-pc)] - [out-pc (port-channel-error-evt out-pc)])) - (make-oc #:kind 'port - #:backend backend - #:direction direction - #:name name - #:get (and in-pc get*) - #:put (and out-pc put*) - #:try-get (and in-pc try-get*) - #:get-evt (and in-pc get-evt*) - #:put-evt (and out-pc put-evt*) - #:close close* - #:error-get error-get* - #:error-try-get error-try-get* - #:error-evt error-evt*)) - -(define (make-uni-port-channel #:input [in #f] - #:output [out #f] - #:source [source 'port-channel] - #:close? [close? #t] - #:name [name 'port-channel]) - (unless (or in out) - (raise-arguments-error 'make-uni-port-channel - "expected at least one of #:input or #:output")) - (when (and in (not (input-port? in))) - (raise-argument-error 'make-uni-port-channel "input-port?" in)) - (when (and out (not (output-port? out))) - (raise-argument-error 'make-uni-port-channel "output-port?" out)) - (define in-pc (and in (make-port-channel in #:direction 'input #:source source #:close? close?))) - (define out-pc (and out (make-port-channel out #:direction 'output #:source source #:close? close?))) - (define direction - (cond - [(and in-pc out-pc) 'bidirectional] - [in-pc 'input] - [else 'output])) - (make-port-oc in-pc out-pc #:direction direction #:name name)) - -(define (uni-channel-closed? ch) - (unbox (uni-channel-closed-box ch))) +(define (make-port-uc pc) + (define dir (port-channel-direction pc)) + (make-uc #:kind 'port + #:impl pc + #:direction dir + #:get (lambda (_ch) (port-channel-get pc)) + #:put (lambda (_ch v) (port-channel-put pc v)) + #:try-get (lambda (_ch) (port-channel-try-get pc)) + #:get-evt (lambda (_ch) (port-channel-evt pc)) + #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (port-channel-put pc v)))) + #:close (lambda (ch) (mark-closed! ch) (close-port-channel pc)) + #:error-get (lambda (_ch) (port-channel-error-get pc)) + #:error-try-get (lambda (_ch) (port-channel-error-try-get pc)) + #:error-evt (lambda (_ch) (port-channel-error-evt pc)))) (define (uni-channel-put ch v) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-put "uni-channel?" ch)) - (unless (can-put? ch) (raise-no-put 'uni-channel-put ch)) (when (uni-channel-closed? ch) (raise-closed 'uni-channel-put ch)) - (define put-proc (uni-channel-put-proc ch)) - (unless put-proc (raise-no-put 'uni-channel-put ch)) - (put-proc ch v)) + (unless (can-put? ch) (raise-no-put 'uni-channel-put ch)) + ((uni-channel-put-proc ch) ch v)) (define (uni-channel-get ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-get "uni-channel?" ch)) + (when (uni-channel-closed? ch) (raise-closed 'uni-channel-get ch)) (unless (can-get? ch) (raise-no-get 'uni-channel-get ch)) - (define get-proc (uni-channel-get-proc ch)) - (unless get-proc (raise-no-get 'uni-channel-get ch)) - (get-proc ch)) + ((uni-channel-get-proc ch) ch)) -(define uni-channel-send uni-channel-put) -(define uni-channel-recv uni-channel-get) +(define (uni-channel-send ch v) (uni-channel-put ch v)) +(define (uni-channel-recv ch) (uni-channel-get ch)) (define (uni-channel-try-get ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-try-get "uni-channel?" ch)) + (when (uni-channel-closed? ch) (raise-closed 'uni-channel-try-get ch)) (unless (can-get? ch) (raise-no-get 'uni-channel-try-get ch)) - (define try-get-proc (uni-channel-try-get-proc ch)) - (unless try-get-proc (raise-no-get 'uni-channel-try-get ch)) - (try-get-proc ch)) + ((uni-channel-try-get-proc ch) ch)) (define (uni-channel-get-evt ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-get-evt "uni-channel?" ch)) - (unless (can-get? ch) (raise-no-get 'uni-channel-get-evt ch)) - (define get-evt-proc (uni-channel-get-evt-proc ch)) - (unless get-evt-proc (raise-no-get 'uni-channel-get-evt ch)) - (get-evt-proc ch)) + (when (uni-channel-closed? ch) (raise-closed 'uni-channel-get-evt ch)) + (if (can-get? ch) ((uni-channel-get-evt-proc ch) ch) never-evt)) (define (uni-channel-put-evt ch v) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-put-evt "uni-channel?" ch)) - (unless (can-put? ch) (raise-no-put 'uni-channel-put-evt ch)) (when (uni-channel-closed? ch) (raise-closed 'uni-channel-put-evt ch)) - (define put-evt-proc (uni-channel-put-evt-proc ch)) - (unless put-evt-proc (raise-no-put 'uni-channel-put-evt ch)) - (put-evt-proc ch v)) + (unless (can-put? ch) (raise-no-put 'uni-channel-put-evt ch)) + ((uni-channel-put-evt-proc ch) ch v)) (define (uni-channel-close ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-close "uni-channel?" ch)) - ((uni-channel-close-proc ch) ch)) + (unless (uni-channel-closed? ch) ((uni-channel-close-proc ch) ch)) + (void)) -(define (uni-channel-error-get ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-error-get "uni-channel?" ch)) - ((uni-channel-error-get-proc ch) ch)) +(define (uni-channel-closed? ch) (unbox (uni-channel-closed-box ch))) -(define (uni-channel-error-try-get ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-error-try-get "uni-channel?" ch)) - ((uni-channel-error-try-get-proc ch) ch)) - -(define (uni-channel-error-evt ch) - (unless (uni-channel? ch) (raise-argument-error 'uni-channel-error-evt "uni-channel?" ch)) - ((uni-channel-error-evt-proc ch) ch)) +(define (uni-channel-error-get ch) ((uni-channel-error-get-proc ch) ch)) +(define (uni-channel-error-try-get ch) ((uni-channel-error-try-get-proc ch) ch)) +(define (uni-channel-error-evt ch) ((uni-channel-error-evt-proc ch) ch)) diff --git a/scribblings/uni-channel.scrbl b/scribblings/uni-channel.scrbl index fdfde66..39e8750 100644 --- a/scribblings/uni-channel.scrbl +++ b/scribblings/uni-channel.scrbl @@ -4,7 +4,6 @@ racket/contract/base racket/async-channel racket/place - racket/serialize port-channel uni-channel)) @@ -13,155 +12,175 @@ @defmodule[uni-channel] -The @racketmodname[uni-channel] module provides a small uniform facade over -three channel-like transports: @racketmodname[racket/async-channel] -asynchronous channels, @racketmodname[racket/place] place channels, and -@racketmodname[port-channel] port channels. +The @racketmodname[uni-channel] module provides one small wrapper around three +Racket channel-like transports: @racket[async-channel?], @racket[place-channel?] +and @racket[port-channel?]. -The purpose is not to hide transport semantics completely. A place channel still -uses place-channel message constraints and a port channel still serializes values -with @racketmodname[racket/serialize]. The purpose is to make code that only -needs send, receive, event synchronization and close operations independent from -the concrete channel implementation. +The constructor is deliberately simple: + +@racketblock[ +(make-uni-channel channel) +] + +The wrapper detects the concrete channel kind and exposes the same small set of +operations for sending, receiving, event synchronization and closing. The wrapper +does not change the transport semantics of the underlying channel. A +@racket[place-channel?] still accepts only place messages, and a +@racket[port-channel?] still transports serialized values as defined by +@racketmodname[port-channel]. @section{Model} -A uni-channel has a kind, a direction, a name and an underlying backend -object. The kind is one of @racket['async], @racket['place] or @racket['port]. -The direction is one of @racket['input], @racket['output] or -@racket['bidirectional]. Receiving is supported for input and bidirectional -channels. Sending is supported for output and bidirectional channels. +A uni-channel has a kind and a direction. -A uni-channel that supports receiving is itself a synchronizable event: +The kind is one of: -@racketblock[(sync ch)] +@itemlist[ + @item{@racket['async] for an @racket[async-channel?].} + @item{@racket['place] for a @racket[place-channel?].} + @item{@racket['port] for a @racket[port-channel?].}] -For output-only channels, synchronizing on the uni-channel uses -@racket[never-evt]. +The direction is one of: -@section{Constructors} +@itemlist[ + @item{@racket['bidirectional] for async-channel and place-channel wrappers.} + @item{@racket['input] or @racket['output] for port-channel wrappers, copied from + @racket[port-channel-direction].}] -@defproc[(make-uni-channel [backend any/c] - [#:kind kind (or/c 'auto 'async 'place 'port) 'auto] - [#:direction direction (or/c 'auto 'input 'output 'bidirectional) 'auto] - [#:name name any/c #f]) +A uni-channel that supports receiving is itself a synchronizable event. +Synchronizing the wrapper returns the value received from the underlying channel: + +@racketblock[ +(sync ch) +] + +For output-only channels, synchronizing on the uni-channel uses @racket[never-evt]. +Such a channel has no receive side, so it can never become ready as a receive +event. + +@section{Constructor} + +@defproc[(make-uni-channel [channel (or/c async-channel? place-channel? port-channel?)]) uni-channel?]{ -Wraps an existing backend channel. With @racket['auto], the kind is inferred from -@racket[async-channel?], @racket[place-channel?] or @racket[port-channel?]. +Wraps @racket[channel] as a uni-channel. The kind and direction are inferred from +the supplied channel. -For a @racket[port-channel?] backend, the direction is read from -@racket[port-channel-direction]. A single port-channel endpoint is therefore -input-only or output-only. Use @racket[make-uni-port-channel] when one uni -channel should combine an input port and an output port into a bidirectional -endpoint.} - -@defproc[(make-uni-async-channel [ch async-channel? (make-async-channel)] - [#:direction direction (or/c 'auto 'input 'output 'bidirectional) 'bidirectional] - [#:name name any/c 'async-channel]) - uni-channel?]{ -Wraps a single asynchronous channel. The same underlying queue is used for both -sending and receiving when the direction is @racket['bidirectional].} - -@defproc[(make-uni-async-channel-pair [#:name-a name-a any/c 'async-channel-a] - [#:name-b name-b any/c 'async-channel-b]) - (values uni-channel? uni-channel?)]{ -Creates two bidirectional uni-channel endpoints backed by two asynchronous channels. -A value sent on the first endpoint is received on the second endpoint, and the -other way around.} - -@defproc[(make-uni-place-channel [ch place-channel?] - [#:direction direction (or/c 'auto 'input 'output 'bidirectional) 'bidirectional] - [#:name name any/c 'place-channel]) - uni-channel?]{ -Wraps an existing place-channel endpoint. Values sent through this backend must -satisfy Racket's place-message constraints.} - -@defproc[(make-uni-place-channel-pair [#:name-a name-a any/c 'place-channel-a] - [#:name-b name-b any/c 'place-channel-b]) - (values uni-channel? uni-channel?)]{ -Creates two bidirectional uni-channel endpoints using @racket[place-channel].} - -@defproc[(make-uni-port-channel [#:input in (or/c #f input-port?) #f] - [#:output out (or/c #f output-port?) #f] - [#:source source any/c 'port-channel] - [#:close? close? any/c #t] - [#:name name any/c 'port-channel]) - uni-channel?]{ -Creates a uni-channel backed by one or two @racketmodname[port-channel] -endpoints. At least one of @racket[#:input] or @racket[#:output] must be -provided. With both, the result is bidirectional: receives come from the input -port-channel and sends go to the output port-channel.} +For @racket[async-channel?] and @racket[place-channel?], the result is +@racket['bidirectional]. For @racket[port-channel?], the result has the same +direction as the port-channel endpoint. To represent a full-duplex port-based +connection, create one input @racket[port-channel?] and one output +@racket[port-channel?], then wrap each endpoint separately with +@racket[make-uni-channel].} @section{Predicates and accessors} -@defproc[(uni-channel? [v any/c]) boolean?]{Returns true when @racket[v] is a uni-channel.} -@defproc[(uni-channel-kind [ch uni-channel?]) (or/c 'async 'place 'port)]{Returns the backend kind.} -@defproc[(uni-channel-direction [ch uni-channel?]) (or/c 'input 'output 'bidirectional)]{Returns the channel direction.} -@defproc[(uni-channel-name [ch uni-channel?]) any/c]{Returns the diagnostic name supplied when the uni-channel was created.} -@defproc[(uni-channel-impl [ch uni-channel?]) any/c]{Returns the underlying backend object.} +@defproc[(uni-channel? [v any/c]) boolean?]{ +Returns true when @racket[v] is a uni-channel.} + +@defproc[(uni-channel-kind [ch uni-channel?]) (or/c 'async 'place 'port)]{ +Returns the detected backend kind.} + +@defproc[(uni-channel-direction [ch uni-channel?]) + (or/c 'input 'output 'bidirectional)]{ +Returns the direction of @racket[ch].} @section{Sending and receiving} @defproc[(uni-channel-put [ch uni-channel?] [v any/c]) void?]{ -Sends @racket[v] on @racket[ch]. The channel must be output-capable. For -@racket['port] channels, the value must be serializable by @racket[serialize]. -For @racket['place] channels, the value must be allowed as a place message.} +Sends @racket[v] on @racket[ch]. The channel must support output. For +@racket['port] channels, the value must be serializable by +@racketmodname[racket/serialize]. For @racket['place] channels, the value must be +allowed as a place message.} -@defproc[(uni-channel-send [ch uni-channel?] [v any/c]) void?]{Alias for @racket[uni-channel-put].} -@defproc[(uni-channel-get [ch uni-channel?]) any/c]{Receives the next value from @racket[ch].} -@defproc[(uni-channel-recv [ch uni-channel?]) any/c]{Alias for @racket[uni-channel-get].} -@defproc[(uni-channel-try-get [ch uni-channel?]) any/c]{Attempts to receive a value without blocking. Returns @racket[#f] when no value is currently available.} -@defproc[(uni-channel-get-evt [ch uni-channel?]) evt?]{Returns a synchronizable receive event.} -@defproc[(uni-channel-put-evt [ch uni-channel?] [v any/c]) evt?]{Returns an event that sends @racket[v] when selected.} +@defproc[(uni-channel-send [ch uni-channel?] [v any/c]) void?]{ +Alias for @racket[uni-channel-put].} + +@defproc[(uni-channel-get [ch uni-channel?]) any/c]{ +Receives the next value from @racket[ch]. The channel must support input.} + +@defproc[(uni-channel-recv [ch uni-channel?]) any/c]{ +Alias for @racket[uni-channel-get].} + +@defproc[(uni-channel-try-get [ch uni-channel?]) any/c]{ +Attempts to receive a value without blocking. Returns @racket[#f] when no value +is currently available. As with the underlying Racket operations, a queued +@racket[#f] value cannot be distinguished from no value by this function.} + +@defproc[(uni-channel-get-evt [ch uni-channel?]) evt?]{ +Returns a synchronizable receive event. Synchronizing this event returns the +value received from the underlying channel. For output-only channels this returns +@racket[never-evt], because no value can be received from that endpoint.} + +@defproc[(uni-channel-put-evt [ch uni-channel?] [v any/c]) evt?]{ +Returns an event that sends @racket[v] when selected. The channel must support +output.} @section{Closing} @defproc[(uni-channel-close [ch uni-channel?]) void?]{ -Closes @racket[ch]. For async-channel and place-channel backends this marks the -uni-channel wrapper as closed; Racket does not provide a native close operation for -those channel types. For port-channel backends this delegates to -@racket[close-port-channel]. For a bidirectional port channel only the output -side is closed, so already queued values can still be received from the input -side until end-of-file.} +Closes the uni-channel wrapper. For async-channel and place-channel wrappers this +marks only the wrapper as closed, because those Racket channel types do not have a +native close operation. For port-channel wrappers this delegates to +@racket[close-port-channel]. Calling @racket[uni-channel-close] more than once is +safe.} -@defproc[(uni-channel-closed? [ch uni-channel?]) boolean?]{Returns true when @racket[uni-channel-close] has been called on the wrapper.} +@defproc[(uni-channel-closed? [ch uni-channel?]) boolean?]{ +Returns true when @racket[uni-channel-close] has been called on @racket[ch].} -@section{Port-channel errors} +@section{Transport errors} -@defproc[(uni-channel-error-get [ch uni-channel?]) any/c]{Blocks until a backend error is available. This is meaningful for @racket['port] channels and raises an exception for backends that do not have an error channel.} -@defproc[(uni-channel-error-try-get [ch uni-channel?]) any/c]{Attempts to get a backend error without blocking. Returns @racket[#f] when no error is available.} -@defproc[(uni-channel-error-evt [ch uni-channel?]) evt?]{Returns a synchronizable backend-error event. For non-port backends this is @racket[never-evt].} +@defproc[(uni-channel-error-get [ch uni-channel?]) any/c]{ +Blocks until a transport error is available. Some wrapped transports can +report asynchronous transport errors, for example I/O or serialization errors. +Wrappers that do not have an error stream raise an exception.} + +@defproc[(uni-channel-error-try-get [ch uni-channel?]) any/c]{ +Attempts to get a transport error without blocking. Returns @racket[#f] when no +error is available. Wrappers that do not have an error stream also return +@racket[#f].} + +@defproc[(uni-channel-error-evt [ch uni-channel?]) evt?]{ +Returns a synchronizable transport-error event. Wrappers that do not have an +error stream return @racket[never-evt].} @section{Examples} -@subsection{Async-channel pair} +@subsection{Async channel} @racketblock[ -(require uni-channel) +(require racket/async-channel + uni-channel) + +(define ch (make-uni-channel (make-async-channel))) + +(uni-channel-send ch 'hello) +(uni-channel-recv ch) +] + +@subsection{Place channel pair} + +@racketblock[ +(require racket/place + uni-channel) + +(define-values (raw-a raw-b) (place-channel)) +(define a (make-uni-channel raw-a)) +(define b (make-uni-channel raw-b)) -(define-values (a b) (make-uni-async-channel-pair)) (uni-channel-send a '(hello from a)) (uni-channel-recv b) ] -@subsection{Place-channel pair} +@subsection{Port-channel endpoints over a pipe} @racketblock[ -(require uni-channel) - -(define-values (a b) (make-uni-place-channel-pair)) -(uni-channel-send a '(hello from a)) -(sync b) -] - -@subsection{Port-channel over a pipe} - -@racketblock[ -(require uni-channel) +(require port-channel + uni-channel) (define-values (in out) (make-pipe)) -(define ch (make-uni-port-channel #:input in #:output out)) +(define reader (make-uni-channel (make-port-channel in))) +(define writer (make-uni-channel (make-port-channel out))) -(uni-channel-send ch '(hello 1 2 3)) -(uni-channel-recv ch) +(uni-channel-send writer '(hello 1 2 3)) +(uni-channel-recv reader) ] diff --git a/test/main.rkt b/test/main.rkt index 979ecf3..8e844f8 100644 --- a/test/main.rkt +++ b/test/main.rkt @@ -1,65 +1,98 @@ #lang racket/base (require rackunit + rackunit/text-ui racket/async-channel + racket/place racket/serialize port-channel uni-channel) (serializable-struct msg (id payload) #:transparent) -(test-case "single async channel" - (define ch (make-uni-async-channel #:name 'single)) - (check-true (uni-channel? ch)) - (check-equal? (uni-channel-kind ch) 'async) - (uni-channel-put ch 'hello) - (check-equal? (uni-channel-get ch) 'hello) - (check-false (uni-channel-try-get ch)) - (sync (uni-channel-put-evt ch 'via-evt)) - (check-equal? (sync ch) 'via-evt) - (uni-channel-close ch) - (check-true (uni-channel-closed? ch)) - (check-exn exn:fail? (lambda () (uni-channel-put ch 'after-close)))) +(define (not-exported? name) + (with-handlers ([exn:fail? (lambda (_e) #t)]) + (dynamic-require 'uni-channel name) + #f)) -(test-case "async channel pair" - (define-values (a b) (make-uni-async-channel-pair)) - (uni-channel-send a '(from a)) - (uni-channel-send b '(from b)) - (check-equal? (uni-channel-recv b) '(from a)) - (check-equal? (uni-channel-recv a) '(from b))) +(define tests + (test-suite + "uni-channel" -(test-case "place channel pair" - (define-values (a b) (make-uni-place-channel-pair)) - (uni-channel-send a '(place a)) - (check-equal? (uni-channel-recv b) '(place a)) - (uni-channel-send b '(place b)) - (check-equal? (sync a) '(place b))) + (test-case "public API has one constructor" + (check-true (procedure? make-uni-channel)) + (check-true (not-exported? 'make-uni-async-channel)) + (check-true (not-exported? 'make-uni-async-channel-pair)) + (check-true (not-exported? 'make-uni-place-channel)) + (check-true (not-exported? 'make-uni-place-channel-pair)) + (check-true (not-exported? 'make-uni-port-channel))) -(test-case "port channel over pipe" - (define-values (in out) (make-pipe)) - (define ch (make-uni-port-channel #:input in #:output out #:source 'pipe-test #:name 'pipe)) - (check-equal? (uni-channel-kind ch) 'port) - (check-equal? (uni-channel-direction ch) 'bidirectional) - (uni-channel-put ch '(hello 1 2 3)) - (uni-channel-put ch (msg 7 '(a b c))) - (check-equal? (uni-channel-get ch) '(hello 1 2 3)) - (check-equal? (uni-channel-get ch) (msg 7 '(a b c))) - (uni-channel-close ch) - (check-true (uni-channel-closed? ch)) - (check-true (eof-object? (uni-channel-get ch)))) + (test-case "async-channel wrapper" + (define ch (make-uni-channel (make-async-channel))) + (check-true (uni-channel? ch)) + (check-equal? (uni-channel-kind ch) 'async) + (check-equal? (uni-channel-direction ch) 'bidirectional) + (check-false (uni-channel-try-get ch)) + (uni-channel-put ch 'hello) + (check-equal? (uni-channel-get ch) 'hello) + (sync (uni-channel-put-evt ch 'via-evt)) + (check-equal? (sync ch) 'via-evt) + (uni-channel-put ch '(sync returns this value)) + (check-equal? (sync (uni-channel-get-evt ch)) '(sync returns this value)) + (uni-channel-close ch) + (check-true (uni-channel-closed? ch)) + (check-exn exn:fail? (lambda () (uni-channel-put ch 'after-close)))) -(test-case "wrap existing port-channel endpoints" - (define-values (in out) (make-pipe)) - (define reader (make-port-channel in #:source 'reader)) - (define writer (make-port-channel out #:source 'writer)) - (define in-ch (make-uni-channel reader #:name 'reader)) - (define out-ch (make-uni-channel writer #:name 'writer)) - (check-equal? (uni-channel-direction in-ch) 'input) - (check-equal? (uni-channel-direction out-ch) 'output) - (uni-channel-put out-ch 'wrapped) - (check-equal? (uni-channel-get in-ch) 'wrapped) - (uni-channel-close out-ch) - (check-true (eof-object? (uni-channel-get in-ch)))) + (test-case "two async-channel wrappers can form a pair explicitly" + (define a-in (make-async-channel)) + (define b-in (make-async-channel)) + (define a (make-uni-channel a-in)) + (define b (make-uni-channel b-in)) + (uni-channel-send a 'local-a) + (uni-channel-send b 'local-b) + (check-equal? (uni-channel-recv a) 'local-a) + (check-equal? (uni-channel-recv b) 'local-b)) + + (test-case "place-channel wrapper" + (define-values (raw-a raw-b) (place-channel)) + (define a (make-uni-channel raw-a)) + (define b (make-uni-channel raw-b)) + (check-equal? (uni-channel-kind a) 'place) + (check-equal? (uni-channel-direction a) 'bidirectional) + (uni-channel-send a '(from a)) + (check-equal? (uni-channel-recv b) '(from a)) + (uni-channel-send b '(from b)) + (check-equal? (sync a) '(from b)) + (uni-channel-send b '(place sync value)) + (check-equal? (sync (uni-channel-get-evt a)) '(place sync value))) + + (test-case "port-channel wrappers" + (define-values (in out) (make-pipe)) + (define reader (make-uni-channel (make-port-channel in #:source 'reader))) + (define writer (make-uni-channel (make-port-channel out #:source 'writer))) + (check-equal? (uni-channel-kind reader) 'port) + (check-equal? (uni-channel-direction reader) 'input) + (check-equal? (uni-channel-direction writer) 'output) + (check-eq? (uni-channel-get-evt writer) never-evt) + (check-false (sync/timeout 0 writer)) + (uni-channel-put writer '(hello 1 2 3)) + (uni-channel-put writer (msg 7 '(a b c))) + (check-equal? (uni-channel-get reader) '(hello 1 2 3)) + (check-equal? (uni-channel-get reader) (msg 7 '(a b c))) + (uni-channel-put writer (msg 9 '(via sync))) + (check-equal? (sync reader) (msg 9 '(via sync))) + (uni-channel-put writer '(via get-evt)) + (check-equal? (sync (uni-channel-get-evt reader)) '(via get-evt)) + (check-exn exn:fail? (lambda () (uni-channel-get writer))) + (check-exn exn:fail? (lambda () (uni-channel-put reader 'nope))) + (uni-channel-close writer) + (check-true (uni-channel-closed? writer)) + (check-true (eof-object? (uni-channel-get reader)))) + + (test-case "bad input" + (check-exn exn:fail? (lambda () (make-uni-channel 'not-a-channel)))))) (module+ main + (define failed (run-tests tests)) + (unless (zero? failed) (exit 1)) (displayln "uni-channel tests ok"))