Remove no longer used pipe implementation
This commit is contained in:
parent
c606237b1c
commit
ce5cd3eaa2
2 changed files with 1 additions and 20 deletions
|
@ -1,7 +1,7 @@
|
|||
(ns clj-kafka.consumer.zk
|
||||
(:import [kafka.consumer ConsumerConfig Consumer KafkaStream]
|
||||
[kafka.javaapi.consumer ConsumerConnector])
|
||||
(:use [clj-kafka.core :only (as-properties to-clojure with-resource pipe)])
|
||||
(:use [clj-kafka.core :only (as-properties to-clojure with-resource)])
|
||||
(:require [zookeeper :as zk]))
|
||||
|
||||
(defn consumer
|
||||
|
|
|
@ -61,22 +61,3 @@
|
|||
TopicMetadataResponse
|
||||
(to-clojure [x]
|
||||
(map to-clojure (.topicsMetadata x))))
|
||||
|
||||
(defn pipe
|
||||
"Returns a vector containing a sequence that will read from the
|
||||
queue, and a function that inserts items into the queue.
|
||||
|
||||
Source: http://clj-me.cgrand.net/2010/04/02/pipe-dreams-are-not-necessarily-made-of-promises/"
|
||||
([] (pipe 100))
|
||||
([size]
|
||||
(let [q (java.util.concurrent.LinkedBlockingQueue. ^int size)
|
||||
EOQ (Object.)
|
||||
NIL (Object.)
|
||||
s (fn queue-seq [] (lazy-seq
|
||||
(let [x (.take q)]
|
||||
(when-not (= EOQ x)
|
||||
(cons (when-not (= NIL x) x)
|
||||
(queue-seq))))))]
|
||||
[(s) (fn queue-put
|
||||
([] (.put q EOQ))
|
||||
([x] (.put q (or x NIL))))])))
|
||||
|
|
Loading…
Reference in a new issue