;; Define yield macros
(define-syntax defstream
  (lambda (x)
    (syntax-case x ()
      ((defstream (name args ...) body ...)
       (with-syntax ((yield (datum->syntax #'defstream 'yield)))
	 (syntax
	   (define (name args ...)
	     (let ((exit #f)
		   (enter #f))
	       (let-syntax ((yield
			      (syntax-rules ()
				((_ var)
				 (call/cc
				   (lambda (cc)
				     (set! enter cc)
				     (exit var)))))))
		 (let ((name
			 (lambda ()
			   body ...
			   (exit 'stop-iteration))))
		   (set! enter (lambda (arg) (name)))
		   (lambda ()
		     (call/cc
		       (lambda (cc)
			 (set! exit cc)
			 (enter #f))))))))))))))

(define-syntax dostream
  (syntax-rules ()
    ((dostream (var stream) body ...)
     (let ((it stream))
       (do ((var (it) (it)))
	   ((eq? var 'stop-iteration))
	 body ...)))))

;; Push and pop (why does scheme not have those?)
(define-syntax push!
  (syntax-rules ()
    ((push! elt lst)
     (set! lst (cons elt lst)))))

(define-syntax pop!
  (syntax-rules ()
    ((pop! lst)
     (let ((elt (car lst)))
       (set! lst (cdr lst))
       elt))))

;; Timer
(define-syntax timeit
  (syntax-rules ()
    ((timeit func)
     (let* ((timer (current-time))
	    (rslt func))
       (display (time-difference (current-time) timer))
       (display "\n")
       rslt))))

;; Pipeline

(defstream (source)
  (do ((k 1 (add1 k)))
      ((= 100000 k))
    (yield (/ (* (add1 k) k) 2))))

(defstream  (make-pairs src)
  (let ((buffer '()))
    (dostream (chunk src)
      (if (null? buffer)
	  (push! chunk buffer)
	  (yield (cons (pop! buffer) chunk))))))

(defstream (split src)
  (dostream (chunk src)
    (yield chunk)
    (yield chunk)))

(defstream (multiply src)
  (dostream (chunk src)
	    (let ((x (car chunk))
		  (y (cdr chunk)))
	      (yield (* x y)))))

(defstream (toy-gcd src)
  (letrec ((helper (lambda (x y)
		     (if (< x y)
			 (helper y x)
			 (if (= 0 y) (yield x)
			     (helper y (modulo x y)))))))
    (dostream (chunk src)
	      (let ((x (car chunk))
		    (y (cdr chunk)))
		(helper x y)))))

(defstream (combine src1 src2)
  (dostream (x src1)
    (yield (cons x (src2)))))

(defstream (divide src)
  (dostream (chunk src)
	    (let ((x (car chunk))
		  (y (cdr chunk)))
	      (yield (quotient x y)))))

(defstream (get-digits src)
  (letrec ((helper (lambda (x)
		     (unless (= 0 x)
		       (helper (quotient x 10))
		       (yield (modulo x 10))))))
    (dostream (x src)
	      (helper x))))

(define (sink src)
  (dostream (chunk src)))

(define (pipeline)
  (let ((split (split (make-pairs (source)))))
    (sink (get-digits (divide (combine (multiply split) (toy-gcd split)))))))

(timeit (do ((i 0 (add1 i)))
	    ((= i 100))
	  (pipeline)))