Обработка бесконечных (Lazy) последовательностей

Question

Я унаследовал некоторый код, который опросает обновления, добавляя эти обновления в ленивую последовательность и обрабатывая. После обновления от clojure 1.7.0-alpha5 до любой более поздней версии код выглядит сломанным из-за chunking ленивой последовательности. Я написал пример, чтобы показать проблему:

(defn infinite-updates 
    [] 
    (letfn [(step [n] 
      (lazy-seq 
      ;; Poll for an update 
      (Thread/sleep 3000) 
      (cons n (step (rand-int 5)))))] 
    (lazy-seq 
     (step (rand-int 5))))) 

;; Run with: 
(doseq [t (sequence (map inc) (infinite-updates))] (println t)) 

Проект был на Clojure 1.7.0-alpha5 и работает как например: каждые 3 секунды t печатается.

После того, как я обновляю эту ревизию, она решает результаты, поэтому я получаю 32 t, напечатанный примерно через 1,5 минуты.

Я попытался с помощью следующих действий:

(defn unchunk [s] 
    (when (seq s) 
    (lazy-seq 
    (cons (first s) 
     (unchunk (next s)))))) 

К unchunk данные не повезло.

Как обрабатывать эти обновления по мере их доступности ИЛИ существует ли более идиоматический способ записи infinite-updates, чтобы обрабатывать все обновления по мере их поступления, не полагаясь на ленивый-seq?

Answer 1

Ваша проблема индивидуальна для clojure.core.async. Этот код показывает контуры:

(ns tst.clj.core 
    (:use clj.core 
     clojure.test) 
    (:require 
    [clojure.core.async :as async] 
)) 

; create a buffer of arbitrary size 99 (any size, even zero, would work) 
(let [buffer (async/chan 99) ] 
    (async/go  ; start sending loop in another thread 
    (while true 
     (Thread/sleep 3000) 
     (let [val (rand-int 5) ] 
     (println "putting:" val) 
     (async/>!! buffer val)))) 
    (while true  ; start receiving loop in this thread 
    (println "received:" 
     (async/<!! buffer)))) 

с выходом:

*clojure-version* => {:major 1, :minor 8, :incremental 0, :qualifier nil} 
java.version => 1.8.0_111 
putting: 4 
received: 4 
putting: 1 
received: 1 
putting: 0 
received: 0 
putting: 0 
received: 0 
putting: 0 
received: 0 
putting: 1 
received: 1 
putting: 1 
received: 1 
putting: 0 
received: 0 
putting: 0 
received: 0 
putting: 0 
received: 0 
putting: 3 
received: 3 
putting: 0 
received: 0 
putting: 1 
received: 1 
putting: 4 
received: 4 
putting: 3 
received: 3 
putting: 2 
received: 2 
putting: 4 
received: 4 

каждые 3 секунды. Пожалуйста, смотрите также:

http://www.braveclojure.com/core-async/

http://clojure.com/blog/2013/06/28/clojure-core-async-channels.html

http://clojure.github.io/core.async/

https://github.com/clojure/core.async/blob/master/examples/walkthrough.clj

https://www.infoq.com/presentations/clojure-core-async

https://youtu.be/enwIIGzhahw

Answer 2

Вместо использования преобразователя используйте «обычную» карту (ваш ручной список не разбит):

(doseq [t (map inc (infinite-updates))] (println t))