port-channel/main.rkt

#lang racket/base

(require racket/async-channel
         racket/serialize)

(provide port-channel?
         port-channel-direction
         port-channel-error?
         port-channel-error-source
         port-channel-error-message

         make-port-channel
         port-channel-put
         port-channel-get
         port-channel-try-get
         port-channel-evt

         port-channel-error-get
         port-channel-error-try-get
         port-channel-error-evt

         close-port-channel
         port-channel-wait)

(struct port-channel-error (source message) #:transparent)

(struct port-channel
  (direction port channel errors thread close-token closing?)
  #:property prop:evt
  (lambda (pc)
    (case (port-channel-direction pc)
      [(input) (port-channel-channel pc)]
      [else never-evt])))

(define failed-token (gensym 'port-channel-failed))

(define (put-error! source errors e)
  (async-channel-put errors (port-channel-error source (exn-message e))))

(define (read-serialized in errors source closing?)
  (define serialized
    (with-handlers ([exn:fail?
                     (lambda (e)
                       (unless (unbox closing?) (put-error! source errors e))
                       eof)])
      (read in)))
  (cond
    [(eof-object? serialized) eof]
    [else
     (with-handlers ([exn:fail?
                      (lambda (e)
                        (put-error! source errors e)
                        failed-token)])
       (deserialize serialized))]))

(define (write-serialized out v errors source)
  (define serialized
    (with-handlers ([exn:fail?
                     (lambda (e)
                       (put-error! source errors e)
                       failed-token)])
      (serialize v)))
  (cond
    [(eq? serialized failed-token) #t]
    [else
     (with-handlers ([exn:fail?
                      (lambda (e)
                        (put-error! source errors e)
                        #f)])
       (write serialized out)
       (newline out)
       (flush-output out)
       #t)]))

(define (make-input-port-channel in source close?)
  (define ch (make-async-channel))
  (define errors (make-async-channel))
  (define closing? (box #f))
  (define close-token (gensym 'close-input-port-channel))
  (define reader
    (thread
     (lambda ()
       (let loop ()
         (define v (read-serialized in errors source closing?))
         (cond
           [(eof-object? v)
            (when close? (with-handlers ([exn:fail? void]) (close-input-port in)))
            (async-channel-put ch eof)]
           [(eq? v failed-token) (loop)]
           [else
            (async-channel-put ch v)
            (loop)])))))
  (port-channel 'input in ch errors reader close-token closing?))

(define (make-output-port-channel out source close?)
  (define ch (make-async-channel))
  (define errors (make-async-channel))
  (define closing? (box #f))
  (define close-token (gensym 'close-output-port-channel))
  (define writer
    (thread
     (lambda ()
       (let loop ()
         (define v (async-channel-get ch))
         (cond
           [(eq? v close-token)
            (set-box! closing? #t)
            (when close? (with-handlers ([exn:fail? void]) (close-output-port out)))]
           [else
            (define ok? (write-serialized out v errors source))
            (when ok? (loop))])))))
  (port-channel 'output out ch errors writer close-token closing?))

(define (make-port-channel port
                           #:direction [direction 'auto]
                           #:source [source 'port]
                           #:close? [close? #t])
  (define in? (input-port? port))
  (define out? (output-port? port))
  (case direction
    [(input)
     (unless in? (raise-argument-error 'make-port-channel "input-port?" port))
     (make-input-port-channel port source close?)]
    [(output)
     (unless out? (raise-argument-error 'make-port-channel "output-port?" port))
     (make-output-port-channel port source close?)]
    [(auto)
     (cond
       [(and in? (not out?)) (make-input-port-channel port source close?)]
       [(and out? (not in?)) (make-output-port-channel port source close?)]
       [(and in? out?)
        (raise-arguments-error
         'make-port-channel
         "port is both input and output; use #:direction 'input or #:direction 'output"
         "port" port)]
       [else (raise-argument-error 'make-port-channel "port?" port)])]
    [else
     (raise-argument-error 'make-port-channel "(or/c 'auto 'input 'output)" direction)]))

(define (port-channel-put pc v)
  (unless (eq? (port-channel-direction pc) 'output)
    (raise-argument-error 'port-channel-put "output port-channel?" pc))
  (async-channel-put (port-channel-channel pc) v))

(define (port-channel-get pc)
  (unless (eq? (port-channel-direction pc) 'input)
    (raise-argument-error 'port-channel-get "input port-channel?" pc))
  (sync pc))

(define (port-channel-try-get pc)
  (unless (eq? (port-channel-direction pc) 'input)
    (raise-argument-error 'port-channel-try-get "input port-channel?" pc))
  (async-channel-try-get (port-channel-channel pc)))

(define (port-channel-evt pc)
  (unless (eq? (port-channel-direction pc) 'input)
    (raise-argument-error 'port-channel-evt "input port-channel?" pc))
  pc)

(define (port-channel-error-get pc)
  (async-channel-get (port-channel-errors pc)))

(define (port-channel-error-try-get pc)
  (async-channel-try-get (port-channel-errors pc)))

(define (port-channel-error-evt pc)
  (port-channel-errors pc))

(define (close-port-channel pc)
  (set-box! (port-channel-closing? pc) #t)
  (case (port-channel-direction pc)
    [(input)
     (with-handlers ([exn:fail? void]) (close-input-port (port-channel-port pc)))]
    [(output)
     (async-channel-put (port-channel-channel pc) (port-channel-close-token pc))]
    [else (void)]))

(define (port-channel-wait pc)
  (thread-wait (port-channel-thread pc)))