196 lines
6.7 KiB
Racket
196 lines
6.7 KiB
Racket
#lang scribble/manual
|
|
|
|
@(require (for-label racket/base
|
|
racket/contract/base
|
|
racket/async-channel
|
|
racket/place
|
|
port-channel
|
|
"../main.rkt"))
|
|
|
|
@title{uni-channel}
|
|
@author[@author+email["Hans Dijkema" "hans@dijkewijk.nl"]]
|
|
|
|
@defmodule[uni-channel]
|
|
|
|
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 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 and a direction.
|
|
|
|
The kind is one of:
|
|
|
|
@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?].}]
|
|
|
|
The direction is one of:
|
|
|
|
@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].}]
|
|
|
|
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 @racket[channel] as a uni-channel. The kind and direction are inferred from
|
|
the supplied 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 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 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]. 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 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-wait [ch uni-channel?]) void?]{
|
|
Waits until the underlying channel worker has stopped, when the backend has such
|
|
a worker. For @racket['port] channels this delegates to
|
|
@racket[port-channel-wait]. This is mainly useful for output port-channels: call
|
|
@racket[uni-channel-close] first to enqueue the close marker, then
|
|
@racket[uni-channel-wait] to wait until all queued values have been written and
|
|
the writer thread has stopped. For @racket['async] and @racket['place] channels,
|
|
this returns immediately.}
|
|
|
|
@defproc[(uni-channel-closed? [ch uni-channel?]) boolean?]{
|
|
Returns true when @racket[uni-channel-close] has been called on @racket[ch].}
|
|
|
|
@section{Transport errors}
|
|
|
|
@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}
|
|
|
|
@racketblock[
|
|
(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))
|
|
|
|
(uni-channel-send a '(hello from a))
|
|
(uni-channel-recv b)
|
|
]
|
|
|
|
@subsection{Port-channel endpoints over a pipe}
|
|
|
|
@racketblock[
|
|
(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)
|
|
]
|