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

Initial import

Browse Source
This commit is contained in:
Hans Dijkema
2026-06-08 12:52:50 +02:00
parent 16a3566571
commit 4e6f922109
5 changed files with 610 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+26 -1
View File
@@ -1,3 +1,28 @@
# uni-channel
Encapsulating async-channel, port-channel and place-channel.
`uni-channel` provides a uniform facade over `async-channel`, `place-channel` and `port-channel`.
The package does not vendor `port-channel`; install `port-channel` from the Racket package index first.
```sh
raco pkg install --auto port-channel
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
```
info.rkt
+17
View File
@@ -0,0 +1,17 @@
#lang info
(define collection "uni-channel")
(define pkg-authors '(hnmdijkema))
(define version "0.1")
(define license '(Apache-2.0 OR MIT))
(define pkg-desc "Uniform channel facade 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))))
main.rkt
+335
View File
@@ -0,0 +1,335 @@
#lang racket/base
(require racket/async-channel
racket/place
port-channel)
(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
uni-channel-send
uni-channel-recv
uni-channel-try-get
uni-channel-get-evt
uni-channel-put-evt
uni-channel-close
uni-channel-closed?
uni-channel-error-get
uni-channel-error-try-get
uni-channel-error-evt)
(struct uni-channel
(kind impl 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)
#: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])))
(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 (raise-no-get who ch)
(raise-arguments-error who "uni-channel does not support receiving"
"name" (uni-channel-name 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)
"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)))
(define (closed-box) (box #f))
(define (mark-closed! ch)
(set-box! (uni-channel-closed-box ch) #t))
(define (default-close ch)
(mark-closed! ch)
(void))
(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
#:direction direction
#:name name
#:get get-proc
#:put put-proc
#:try-get try-get-proc
#:get-evt get-evt-proc
#:put-evt put-evt-proc
#:close close-proc
#: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))
(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-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))
#: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))))
#: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 (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))
(define (uni-channel-get ch)
(unless (uni-channel? ch) (raise-argument-error 'uni-channel-get "uni-channel?" 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))
(define uni-channel-send uni-channel-put)
(define uni-channel-recv uni-channel-get)
(define (uni-channel-try-get ch)
(unless (uni-channel? ch) (raise-argument-error 'uni-channel-try-get "uni-channel?" 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))
(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))
(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))
(define (uni-channel-close ch)
(unless (uni-channel? ch) (raise-argument-error 'uni-channel-close "uni-channel?" ch))
((uni-channel-close-proc ch) ch))
(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-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))
scribblings/uni-channel.scrbl
+167
View File
@@ -0,0 +1,167 @@
#lang scribble/manual
@(require (for-label racket/base
racket/contract/base
racket/async-channel
racket/place
racket/serialize
port-channel
uni-channel))
@title{uni-channel}
@author[@author+email["Hans Dijkema" "hans@dijkewijk.nl"]]
@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 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.
@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 that supports receiving is itself a synchronizable event:
@racketblock[(sync ch)]
For output-only channels, synchronizing on the uni-channel uses
@racket[never-evt].
@section{Constructors}
@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])
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?].
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.}
@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.}
@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.}
@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.}
@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.}
@defproc[(uni-channel-closed? [ch uni-channel?]) boolean?]{Returns true when @racket[uni-channel-close] has been called on the wrapper.}
@section{Port-channel 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].}
@section{Examples}
@subsection{Async-channel pair}
@racketblock[
(require uni-channel)
(define-values (a b) (make-uni-async-channel-pair))
(uni-channel-send a '(hello from a))
(uni-channel-recv b)
]
@subsection{Place-channel pair}
@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)
(define-values (in out) (make-pipe))
(define ch (make-uni-port-channel #:input in #:output out))
(uni-channel-send ch '(hello 1 2 3))
(uni-channel-recv ch)
]
test/main.rkt
+65
View File
@@ -0,0 +1,65 @@
#lang racket/base
(require rackunit
racket/async-channel
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))))
(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)))
(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 "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 "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))))
(module+ main
(displayln "uni-channel tests ok"))