hans/uni-channel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync tested.

Browse Source
This commit is contained in:
Hans Dijkema
2026-06-08 14:21:36 +02:00
parent 78f4583b6d
commit f0c4f5a4ce
6 changed files with 331 additions and 427 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+5
View File
@@ -16,3 +16,8 @@ compiled/
*.dep *.dep
/*.bak /*.bak
/scribblings/*.css
/scribblings/*.js
/scribblings/*.bak
/scribblings/*.html
README.md
+40 -18
View File
@@ -1,8 +1,46 @@
# uni-channel # 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 ```sh
raco pkg install --auto port-channel raco pkg install --auto port-channel
@@ -10,19 +48,3 @@ raco pkg install --auto ./uni-channel
raco test -x -p uni-channel raco test -x -p uni-channel
raco setup --pkgs 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
```
info.rkt
+4 -11
View File
@@ -2,20 +2,13 @@
(define collection "uni-channel") (define collection "uni-channel")
(define pkg-authors '(hnmdijkema)) (define pkg-authors '(hnmdijkema))
(define version "0.1") (define version "0.2")
(define license '(Apache-2.0 OR MIT)) (define license '(Apache-2.0 OR MIT))
(define pkg-desc "Uniform channel facade for place-channel, async-channel and port-channel.") (define pkg-desc "Uniform channel wrapper for place-channel, async-channel and port-channel.")
(define scribblings
'(
("scribblings/uni-channel.scrbl" () (library))
)
)
(define deps '("base" "port-channel")) (define deps '("base" "port-channel"))
(define build-deps '("racket-doc" "scribble-lib" "rackunit-lib")) (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 compile-omit-paths '("test"))
(define scribblings
'(("scribblings/uni-channel.scrbl" () (library))))
main.rkt
+71 -239
View File
@@ -7,15 +7,8 @@
(provide uni-channel? (provide uni-channel?
uni-channel-kind uni-channel-kind
uni-channel-direction uni-channel-direction
uni-channel-name
uni-channel-impl
make-uni-channel 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-put
uni-channel-get uni-channel-get
@@ -32,50 +25,31 @@
uni-channel-error-evt) uni-channel-error-evt)
(struct uni-channel (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) close-proc closed-box error-get-proc error-try-get-proc error-evt-proc)
#:property prop:evt #:property prop:evt
(lambda (ch) (lambda (ch)
(cond (if (can-get? ch) ((uni-channel-get-evt-proc ch) ch) never-evt)))
[(and (can-get? ch) (uni-channel-get-proc ch))
((uni-channel-get-evt-proc ch) ch)]
[else never-evt])))
(define (check-direction who v) (define (can-get? ch) (not (eq? (uni-channel-direction ch) 'output)))
(case v (define (can-put? ch) (not (eq? (uni-channel-direction ch) 'input)))
[(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 (raise-no-get who ch) (define (raise-no-get who ch)
(raise-arguments-error who "uni-channel does not support receiving" (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))) "direction" (uni-channel-direction ch)))
(define (raise-no-put who ch) (define (raise-no-put who ch)
(raise-arguments-error who "uni-channel does not support sending" (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))) "direction" (uni-channel-direction ch)))
(define (raise-closed who ch) (define (raise-closed who ch)
(raise-arguments-error who "uni-channel is closed" (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) (define (default-close ch)
(mark-closed! ch) (mark-closed! ch)
@@ -83,16 +57,14 @@
(define (no-error-get ch) (define (no-error-get ch)
(raise-arguments-error 'uni-channel-error-get "uni-channel has no error channel" (raise-arguments-error 'uni-channel-error-get "uni-channel has no error channel"
"name" (uni-channel-name ch)
"kind" (uni-channel-kind ch))) "kind" (uni-channel-kind ch)))
(define (no-error-try-get ch) #f) (define (no-error-try-get ch) #f)
(define (no-error-evt ch) never-evt) (define (no-error-evt ch) never-evt)
(define (make-oc #:kind kind (define (make-uc #:kind kind
#:backend backend #:impl impl
#:direction direction #:direction direction
#:name name
#:get get-proc #:get get-proc
#:put put-proc #:put put-proc
#:try-get try-get-proc #:try-get try-get-proc
@@ -102,234 +74,94 @@
#:error-get error-get-proc #:error-get error-get-proc
#:error-try-get error-try-get-proc #:error-try-get error-try-get-proc
#:error-evt error-evt-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 (uni-channel 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)) close-proc (box #f) error-get-proc error-try-get-proc error-evt-proc))
(define (make-uni-channel backend (define (make-uni-channel channel)
#:kind [kind 'auto] (cond
#:direction [direction 'auto] [(async-channel? channel) (make-async-uc channel)]
#:name [name #f]) [(place-channel? channel) (make-place-uc channel)]
(define detected-kind [(port-channel? channel) (make-port-uc channel)]
(case kind [else (raise-argument-error 'make-uni-channel
[(auto) "(or/c async-channel? place-channel? port-channel?)"
(cond channel)]))
[(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-async-channel [ch (make-async-channel)] (define (make-async-uc ch)
#:direction [direction 'bidirectional] (make-uc #:kind 'async
#:name [name 'async-channel]) #:impl ch
(define dir (if (eq? direction 'auto) 'bidirectional (check-direction 'make-uni-async-channel direction))) #:direction 'bidirectional
(define backend ch) #:get (lambda (_ch) (async-channel-get ch))
(make-oc #:kind 'async #:put (lambda (_ch v) (async-channel-put ch v))
#:backend backend #:try-get (lambda (_ch) (async-channel-try-get ch))
#:direction dir #:get-evt (lambda (_ch) ch)
#:name name #:put-evt (lambda (_ch v) (async-channel-put-evt ch v))
#: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))
#:close default-close #:close default-close
#:error-get no-error-get #:error-get no-error-get
#:error-try-get no-error-try-get #:error-try-get no-error-try-get
#:error-evt no-error-evt)) #:error-evt no-error-evt))
(define (make-uni-async-channel-pair #:name-a [name-a 'async-channel-a] (define (make-place-uc ch)
#:name-b [name-b 'async-channel-b]) (make-uc #:kind 'place
(define a-in (make-async-channel)) #:impl ch
(define b-in (make-async-channel)) #:direction 'bidirectional
(define (endpoint name incoming outgoing) #:get (lambda (_ch) (place-channel-get ch))
(make-oc #:kind 'async #:put (lambda (_ch v) (place-channel-put ch v))
#:backend (vector 'async-channel-pair incoming outgoing) #:try-get (lambda (_ch) (sync/timeout 0 ch))
#:direction 'bidirectional #:get-evt (lambda (_ch) ch)
#:name name #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (place-channel-put ch v))))
#: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))))
#:close default-close #:close default-close
#:error-get no-error-get #:error-get no-error-get
#:error-try-get no-error-try-get #:error-try-get no-error-try-get
#:error-evt no-error-evt)) #:error-evt no-error-evt))
(define (make-uni-place-channel-pair #:name-a [name-a 'place-channel-a] (define (make-port-uc pc)
#:name-b [name-b 'place-channel-b]) (define dir (port-channel-direction pc))
(define-values (a b) (place-channel)) (make-uc #:kind 'port
(values (make-uni-place-channel a #:name name-a) #:impl pc
(make-uni-place-channel b #:name name-b))) #:direction dir
#:get (lambda (_ch) (port-channel-get pc))
(define (make-port-oc in-pc out-pc #:direction direction #:name name) #:put (lambda (_ch v) (port-channel-put pc v))
(define backend #:try-get (lambda (_ch) (port-channel-try-get pc))
(cond #:get-evt (lambda (_ch) (port-channel-evt pc))
[(and in-pc out-pc) (vector 'port-channel-pair in-pc out-pc)] #:put-evt (lambda (_ch v) (handle-evt always-evt (lambda (_) (port-channel-put pc v))))
[in-pc in-pc] #:close (lambda (ch) (mark-closed! ch) (close-port-channel pc))
[out-pc out-pc] #:error-get (lambda (_ch) (port-channel-error-get pc))
[else (error 'make-port-oc "internal error: no port-channel backend")])) #:error-try-get (lambda (_ch) (port-channel-error-try-get pc))
(define (get* _ch) (port-channel-get in-pc)) #:error-evt (lambda (_ch) (port-channel-error-evt 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 (uni-channel-put ch v) (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)) (when (uni-channel-closed? ch) (raise-closed 'uni-channel-put ch))
(define put-proc (uni-channel-put-proc ch)) (unless (can-put? ch) (raise-no-put 'uni-channel-put ch))
(unless put-proc (raise-no-put 'uni-channel-put ch)) ((uni-channel-put-proc ch) ch v))
(put-proc ch v))
(define (uni-channel-get ch) (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)) (unless (can-get? ch) (raise-no-get 'uni-channel-get ch))
(define get-proc (uni-channel-get-proc ch)) ((uni-channel-get-proc ch) ch))
(unless get-proc (raise-no-get 'uni-channel-get ch))
(get-proc ch))
(define uni-channel-send uni-channel-put) (define (uni-channel-send ch v) (uni-channel-put ch v))
(define uni-channel-recv uni-channel-get) (define (uni-channel-recv ch) (uni-channel-get ch))
(define (uni-channel-try-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)) (unless (can-get? ch) (raise-no-get 'uni-channel-try-get ch))
(define try-get-proc (uni-channel-try-get-proc ch)) ((uni-channel-try-get-proc ch) ch))
(unless try-get-proc (raise-no-get 'uni-channel-try-get ch))
(try-get-proc ch))
(define (uni-channel-get-evt ch) (define (uni-channel-get-evt ch)
(unless (uni-channel? ch) (raise-argument-error 'uni-channel-get-evt "uni-channel?" ch)) (when (uni-channel-closed? ch) (raise-closed 'uni-channel-get-evt ch))
(unless (can-get? ch) (raise-no-get 'uni-channel-get-evt ch)) (if (can-get? ch) ((uni-channel-get-evt-proc ch) ch) never-evt))
(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))
(define (uni-channel-put-evt ch v) (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)) (when (uni-channel-closed? ch) (raise-closed 'uni-channel-put-evt ch))
(define put-evt-proc (uni-channel-put-evt-proc ch)) (unless (can-put? ch) (raise-no-put 'uni-channel-put-evt ch))
(unless put-evt-proc (raise-no-put 'uni-channel-put-evt ch)) ((uni-channel-put-evt-proc ch) ch v))
(put-evt-proc ch v))
(define (uni-channel-close ch) (define (uni-channel-close ch)
(unless (uni-channel? ch) (raise-argument-error 'uni-channel-close "uni-channel?" ch)) (unless (uni-channel-closed? ch) ((uni-channel-close-proc ch) ch))
((uni-channel-close-proc ch) ch)) (void))
(define (uni-channel-error-get ch) (define (uni-channel-closed? ch) (unbox (uni-channel-closed-box 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-error-try-get ch) (define (uni-channel-error-get ch) ((uni-channel-error-get-proc ch) ch))
(unless (uni-channel? ch) (raise-argument-error 'uni-channel-error-try-get "uni-channel?" ch)) (define (uni-channel-error-try-get ch) ((uni-channel-error-try-get-proc ch) ch))
((uni-channel-error-try-get-proc ch) ch)) (define (uni-channel-error-evt ch) ((uni-channel-error-evt-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))
scribblings/uni-channel.scrbl
+130 -111
View File
@@ -4,7 +4,6 @@
racket/contract/base racket/contract/base
racket/async-channel racket/async-channel
racket/place racket/place
racket/serialize
port-channel port-channel
uni-channel)) uni-channel))
@@ -13,155 +12,175 @@
@defmodule[uni-channel] @defmodule[uni-channel]
The @racketmodname[uni-channel] module provides a small uniform facade over The @racketmodname[uni-channel] module provides one small wrapper around three
three channel-like transports: @racketmodname[racket/async-channel] Racket channel-like transports: @racket[async-channel?], @racket[place-channel?]
asynchronous channels, @racketmodname[racket/place] place channels, and and @racket[port-channel?].
@racketmodname[port-channel] port channels.
The purpose is not to hide transport semantics completely. A place channel still The constructor is deliberately simple:
uses place-channel message constraints and a port channel still serializes values
with @racketmodname[racket/serialize]. The purpose is to make code that only @racketblock[
needs send, receive, event synchronization and close operations independent from (make-uni-channel channel)
the concrete channel implementation. ]
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} @section{Model}
A uni-channel has a kind, a direction, a name and an underlying backend A uni-channel has a kind and a direction.
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 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 The direction is one of:
@racket[never-evt].
@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] A uni-channel that supports receiving is itself a synchronizable event.
[#:kind kind (or/c 'auto 'async 'place 'port) 'auto] Synchronizing the wrapper returns the value received from the underlying channel:
[#:direction direction (or/c 'auto 'input 'output 'bidirectional) 'auto]
[#:name name any/c #f]) @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?]{ uni-channel?]{
Wraps an existing backend channel. With @racket['auto], the kind is inferred from Wraps @racket[channel] as a uni-channel. The kind and direction are inferred from
@racket[async-channel?], @racket[place-channel?] or @racket[port-channel?]. the supplied channel.
For a @racket[port-channel?] backend, the direction is read from For @racket[async-channel?] and @racket[place-channel?], the result is
@racket[port-channel-direction]. A single port-channel endpoint is therefore @racket['bidirectional]. For @racket[port-channel?], the result has the same
input-only or output-only. Use @racket[make-uni-port-channel] when one uni direction as the port-channel endpoint. To represent a full-duplex port-based
channel should combine an input port and an output port into a bidirectional connection, create one input @racket[port-channel?] and one output
endpoint.} @racket[port-channel?], then wrap each endpoint separately with
@racket[make-uni-channel].}
@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.}
@section{Predicates and accessors} @section{Predicates and accessors}
@defproc[(uni-channel? [v any/c]) boolean?]{Returns true when @racket[v] is a uni-channel.} @defproc[(uni-channel? [v any/c]) boolean?]{
@defproc[(uni-channel-kind [ch uni-channel?]) (or/c 'async 'place 'port)]{Returns the backend kind.} Returns true when @racket[v] is a uni-channel.}
@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-kind [ch uni-channel?]) (or/c 'async 'place 'port)]{
@defproc[(uni-channel-impl [ch uni-channel?]) any/c]{Returns the underlying backend object.} 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} @section{Sending and receiving}
@defproc[(uni-channel-put [ch uni-channel?] [v any/c]) void?]{ @defproc[(uni-channel-put [ch uni-channel?] [v any/c]) void?]{
Sends @racket[v] on @racket[ch]. The channel must be output-capable. For Sends @racket[v] on @racket[ch]. The channel must support output. For
@racket['port] channels, the value must be serializable by @racket[serialize]. @racket['port] channels, the value must be serializable by
For @racket['place] channels, the value must be allowed as a place message.} @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-send [ch uni-channel?] [v any/c]) void?]{
@defproc[(uni-channel-get [ch uni-channel?]) any/c]{Receives the next value from @racket[ch].} Alias for @racket[uni-channel-put].}
@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 [ch uni-channel?]) any/c]{
@defproc[(uni-channel-get-evt [ch uni-channel?]) evt?]{Returns a synchronizable receive event.} Receives the next value from @racket[ch]. The channel must support input.}
@defproc[(uni-channel-put-evt [ch uni-channel?] [v any/c]) evt?]{Returns an event that sends @racket[v] when selected.}
@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} @section{Closing}
@defproc[(uni-channel-close [ch uni-channel?]) void?]{ @defproc[(uni-channel-close [ch uni-channel?]) void?]{
Closes @racket[ch]. For async-channel and place-channel backends this marks the Closes the uni-channel wrapper. For async-channel and place-channel wrappers this
uni-channel wrapper as closed; Racket does not provide a native close operation for marks only the wrapper as closed, because those Racket channel types do not have a
those channel types. For port-channel backends this delegates to native close operation. For port-channel wrappers this delegates to
@racket[close-port-channel]. For a bidirectional port channel only the output @racket[close-port-channel]. Calling @racket[uni-channel-close] more than once is
side is closed, so already queued values can still be received from the input safe.}
side until end-of-file.}
@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-get [ch uni-channel?]) any/c]{
@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.} Blocks until a transport error is available. Some wrapped transports can
@defproc[(uni-channel-error-evt [ch uni-channel?]) evt?]{Returns a synchronizable backend-error event. For non-port backends this is @racket[never-evt].} 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} @section{Examples}
@subsection{Async-channel pair} @subsection{Async channel}
@racketblock[ @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-send a '(hello from a))
(uni-channel-recv b) (uni-channel-recv b)
] ]
@subsection{Place-channel pair} @subsection{Port-channel endpoints over a pipe}
@racketblock[ @racketblock[
(require uni-channel) (require port-channel
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)
(define-values (in out) (make-pipe)) (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-send writer '(hello 1 2 3))
(uni-channel-recv ch) (uni-channel-recv reader)
] ]
test/main.rkt
+81 -48
View File
@@ -1,65 +1,98 @@
#lang racket/base #lang racket/base
(require rackunit (require rackunit
rackunit/text-ui
racket/async-channel racket/async-channel
racket/place
racket/serialize racket/serialize
port-channel port-channel
uni-channel) uni-channel)
(serializable-struct msg (id payload) #:transparent) (serializable-struct msg (id payload) #:transparent)
(test-case "single async channel" (define (not-exported? name)
(define ch (make-uni-async-channel #:name 'single)) (with-handlers ([exn:fail? (lambda (_e) #t)])
(check-true (uni-channel? ch)) (dynamic-require 'uni-channel name)
(check-equal? (uni-channel-kind ch) 'async) #f))
(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))))
(test-case "async channel pair" (define tests
(define-values (a b) (make-uni-async-channel-pair)) (test-suite
(uni-channel-send a '(from a)) "uni-channel"
(uni-channel-send b '(from b))
(check-equal? (uni-channel-recv b) '(from a))
(check-equal? (uni-channel-recv a) '(from b)))
(test-case "place channel pair" (test-case "public API has one constructor"
(define-values (a b) (make-uni-place-channel-pair)) (check-true (procedure? make-uni-channel))
(uni-channel-send a '(place a)) (check-true (not-exported? 'make-uni-async-channel))
(check-equal? (uni-channel-recv b) '(place a)) (check-true (not-exported? 'make-uni-async-channel-pair))
(uni-channel-send b '(place b)) (check-true (not-exported? 'make-uni-place-channel))
(check-equal? (sync a) '(place b))) (check-true (not-exported? 'make-uni-place-channel-pair))
(check-true (not-exported? 'make-uni-port-channel)))
(test-case "port channel over pipe" (test-case "async-channel wrapper"
(define-values (in out) (make-pipe)) (define ch (make-uni-channel (make-async-channel)))
(define ch (make-uni-port-channel #:input in #:output out #:source 'pipe-test #:name 'pipe)) (check-true (uni-channel? ch))
(check-equal? (uni-channel-kind ch) 'port) (check-equal? (uni-channel-kind ch) 'async)
(check-equal? (uni-channel-direction ch) 'bidirectional) (check-equal? (uni-channel-direction ch) 'bidirectional)
(uni-channel-put ch '(hello 1 2 3)) (check-false (uni-channel-try-get ch))
(uni-channel-put ch (msg 7 '(a b c))) (uni-channel-put ch 'hello)
(check-equal? (uni-channel-get ch) '(hello 1 2 3)) (check-equal? (uni-channel-get ch) 'hello)
(check-equal? (uni-channel-get ch) (msg 7 '(a b c))) (sync (uni-channel-put-evt ch 'via-evt))
(uni-channel-close ch) (check-equal? (sync ch) 'via-evt)
(check-true (uni-channel-closed? ch)) (uni-channel-put ch '(sync returns this value))
(check-true (eof-object? (uni-channel-get ch)))) (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" (test-case "two async-channel wrappers can form a pair explicitly"
(define-values (in out) (make-pipe)) (define a-in (make-async-channel))
(define reader (make-port-channel in #:source 'reader)) (define b-in (make-async-channel))
(define writer (make-port-channel out #:source 'writer)) (define a (make-uni-channel a-in))
(define in-ch (make-uni-channel reader #:name 'reader)) (define b (make-uni-channel b-in))
(define out-ch (make-uni-channel writer #:name 'writer)) (uni-channel-send a 'local-a)
(check-equal? (uni-channel-direction in-ch) 'input) (uni-channel-send b 'local-b)
(check-equal? (uni-channel-direction out-ch) 'output) (check-equal? (uni-channel-recv a) 'local-a)
(uni-channel-put out-ch 'wrapped) (check-equal? (uni-channel-recv b) 'local-b))
(check-equal? (uni-channel-get in-ch) 'wrapped)
(uni-channel-close out-ch) (test-case "place-channel wrapper"
(check-true (eof-object? (uni-channel-get in-ch)))) (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 (module+ main
(define failed (run-tests tests))
(unless (zero? failed) (exit 1))
(displayln "uni-channel tests ok")) (displayln "uni-channel tests ok"))