イテレータの反転をスレッドで実装すると、プロデューサ・コンシューマパターンになる。

mt.scm

; スレッドを使ってhttp-getからジェネレータを作る。
; そして仮想ポートをつくる。

(use rfc.http)
(use gauche.threads)
(use gauche.generator)
(use util.queue)
(use gauche.vport)
(use gauche.uvector)
(use sxml.ssax) 

;(use htmlprag)
;(use pretty-print)
;(use web-helper)


(define (call-with-input-http http-param proc :key (queue-size 100))

  (define (make-producer mtq buffer-size http-param)
    ;#?= http-param
    (^ []
      (guard (e [else 
                  (print (standard-error-port) (ref e 'message) ) 
                  (enqueue/wait! mtq (eof-object)) ] )
        (let1 outp (make <buffered-output-port> 
                         :buffer-size buffer-size
                         :flush (^ (u8v flag)
                                  (enqueue/wait! mtq (u8vector-copy u8v))
                                  (uvector-length u8v) ) )
          (apply http-get (append 
                            http-param `(:sink ,outp :flusher ,(^ _ (flush outp) #t) )))
          (enqueue/wait! mtq (eof-object))
          ) ) ) )

  (define (make-consumer mtq buffer-size proc)
    (^ []
      (guard (e [else (print (standard-error-port) (ref e 'message) ) ]) 
        (let* [( g (generate (^(yield)
                               (let loop [(xc (dequeue/wait! mtq))]
                                 (if (eof-object? xc)
                                   (yield (eof-object))
                                   (begin
                                     ;#?= xc
                                     (yield xc)
                                     (loop (dequeue/wait! mtq) ) ) ) ) ) ) )
               ( vp (make <buffered-input-port> 
                          :buffer-size buffer-size
                          :fill (^(u8v-dst)
                                  (let1 u8v-src (g)
                                    (if (eof-object? u8v-src)
                                      0
                                      (let [(ndst (uvector-length u8v-dst))
                                            (nsrc (uvector-length u8v-src ))]
                                        (if (>= ndst nsrc )
                                          (begin 
                                            (u8vector-copy! u8v-dst 0 u8v-src ) 
                                            ;#?= u8v-dst
                                            nsrc )
                                          (begin 
                                            ; ここには来ないはずだが念の為
                                            (error "ndst < nsrc") ) ) ) ) ) ) ) )
               ]
          (proc vp)
          ) ) ) )    

  (and-let* [ ( mtq (make-mtqueue :max-length queue-size))
             ( buf-size 40000)
             ( c (make-thread (make-consumer mtq buf-size proc) ) )
             ( p (make-thread (make-producer mtq buf-size http-param)))]
    (let [(tc (thread-start! c))
          (tp (thread-start! p)) ]
      (thread-join! tp)
      (thread-join! tc) ; <----consumerの値を返す 
      ) ) ) 

;(define (call-with-input-http-uri uri proc :key (queue-size 100))
;  (receive (host path) (uri->param uri)
;    (call-with-input-http 
;      (list host path) proc :queue-size queue-size )))

(define (main args)
  (call-with-input-http 
    (cdr args)
    (^p
      ;(print (port->string p))
      (print (ssax:xml->sxml p '() ) )
      ;(print (html->sxml p))
      )
    :queue-size 2
    ))