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

mt.scm

; スレッドを使ってhttp-getからキャラクタジェネレータを作る。
; そして仮想ポートをつくる。
; https://gist.github.com/3174805
(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 (http-hook-receiver proc) 
  (lambda (code hdrs total retr)
    (let loop [ ]
      (receive (port size) (retr) 
        (cond
          [ (eqv? size 0) #t]
          [ (or (not size) (> size 0) )
           (let loop-rb [(data (read-block size port) ) 
                         (req-size  size) ]
             
             (proc data) ; hook
             (let1 nrest (- req-size (string-length data))
               (if (= nrest 0)
                 (loop)
                 (loop-rb (read-block nrest port) nrest ) ) ) ) ]
          ) ) ) ) )

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

  (define (make-producer mtq http-param)
    ;#?= http-param
    (^ []
      (guard (e [else 
                  (print (standard-error-port) (ref e 'message) ) 
                  (enqueue/wait! mtq (eof-object)) ] )
        (apply http-get (append 
                          http-param
                          `(:receiver
                             ,(http-hook-receiver
                                (^(xp) 
                                  ;#?= xp
                                  (enqueue/wait! mtq xp) ) ) ) ) )
        (enqueue/wait! mtq (eof-object))
        ) ) )

  (define (make-consumer mtq proc)
    (^ []
      (guard (e [else (print (standard-error-port) (ref e 'message) ) ]) 
        (let* [( g ($ gconcatenate 
                     $ gmap  string->generator
                     $ 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 <virtual-input-port> :getc g ) )
               ]
          (proc vp)
          ) ) ) )    

  (and-let* [ ( mtq (make-mtqueue :max-length queue-size))
              ( c (make-thread (make-consumer mtq proc) ) )
              ( p (make-thread (make-producer mtq 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
    ))