basis/io/pipes/pipes.factor

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