extra/io/pipes/pipes.factor

   1 ! Copyright (C) 2008 Slava Pestov.
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 USING: io.encodings io.backend io.ports io.streams.duplex
   4 io splitting grouping sequences namespaces kernel
   5 destructors math concurrency.combinators accessors
   6 arrays continuations quotations system vocabs.loader combinators ;
   7 IN: io.pipes
   8
   9 TUPLE: pipe in out ;
  10
  11 M: pipe dispose ( pipe -- )
  12     [ in>> dispose ] [ out>> dispose ] bi ;
  13
  14 HOOK: (pipe) io-backend ( -- pipe )
  15
  16 : <pipe> ( encoding -- stream )
  17     [
  18         >r (pipe) |dispose
  19         [ in>> <input-port> ] [ out>> <output-port> ] bi
  20         r> <encoder-duplex>
  21     ] with-destructors ;
  22
  23 <PRIVATE
  24
  25 : ?reader ( handle/f -- stream )
  26     [ <input-port> &dispose ] [ input-stream get ] if* ;
  27
  28 : ?writer ( handle/f -- stream )
  29     [ <output-port> &dispose ] [ output-stream get ] if* ;
  30
  31 GENERIC: run-pipeline-element ( input-fd output-fd obj -- quot )
  32
  33 M: callable run-pipeline-element
  34     [
  35         >r [ ?reader ] [ ?writer ] bi*
  36         r> with-streams*
  37     ] with-destructors ;
  38
  39 : <pipes> ( n -- pipes )
  40     [
  41         [ (pipe) |dispose ] replicate
  42         T{ pipe } [ prefix ] [ suffix ] bi
  43         2 <clumps>
  44     ] with-destructors ;
  45
  46 PRIVATE>
  47
  48 : run-pipeline ( seq -- results )
  49     [ length dup zero? [ drop { } ] [ 1- <pipes> ] if ] keep
  50     [
  51         >r [ first in>> ] [ second out>> ] bi
  52         r> run-pipeline-element
  53     ] 2parallel-map ;
  54
  55 {
  56     { [ os unix? ] [ "io.unix.pipes" require ] }
  57     { [ os winnt? ] [ "io.windows.nt.pipes" require ] }
  58     [ ]
  59 } cond