Back
dorun (clj)
(source)function
(dorun coll)
(dorun n coll)
When lazy sequences are produced via functions that have side
effects, any effects other than those needed to produce the first
element in the seq do not occur until the seq is consumed. dorun can
be used to force any effects. Walks through the successive nexts of
the seq, does not retain the head and returns nil.
Examples
typedclojure/typedclojure
(ns ^:no-doc typed.ann.clojure
"Type annotations for the base Clojure distribution."
#?(:cljs (:require-macros [typed.ann-macros.clojure :as macros]))
(:require [clojure.core :as cc]
[typed.clojure :as t]
#?(:clj [typed.ann-macros.clojure :as macros])
#?(:clj typed.ann.clojure.jvm) ;; jvm annotations
#?(:clj clojure.core.typed))
#?(:clj
(:import (clojure.lang PersistentHashSet PersistentList
APersistentMap #_IPersistentCollection
#_ITransientSet
IRef)
(java.util Comparator Collection))))
cc/doall (t/All [[c :< (t/U nil t/AnySeqable)]]
[(t/? t/AnyInteger) c :-> c])
cc/dorun [(t/? t/AnyInteger) t/AnySeqable :-> nil]
cc/iterate (t/All [x]
[[x :-> x] x :-> (t/NonEmptyASeq x)])
cc/memoize (t/All [x y :..]
[[y :.. y :-> x] :-> [y :.. y :-> x]])
apa512/clj-rethinkdb
(ns rethinkdb.core-test
(:require [clojure.java.io :as io]
[byte-streams :as bs]
[clj-time.core :as t]
[clj-time.coerce :as c]
[clojure.test :refer :all]
[clojure.core.async :as async :refer [go go-loop <! take! <!! close!]]
[manifold.stream :as s]
[rethinkdb.query :as r]
[rethinkdb.test-utils :as utils])
(:import (clojure.lang ExceptionInfo)
(java.util UUID Arrays)
(ch.qos.logback.classic Logger Level)
(manifold.stream.default Stream)
(org.slf4j LoggerFactory)))
(deftest binary
(testing "Encoding and decoding is thread safe"
(dorun (pmap
#(with-open [conn (r/connect :db test-db)]
(let [file (io/file (io/resource "pikachu.png"))
file-bytes (bs/to-byte-array file)]
(-> (r/table "pokedex")
(r/insert {:national_no %
:name "Pikachu"
:image file-bytes})
(r/run conn))
(let [resp (-> (r/table "pokedex") (r/get %) (r/run conn) :image)]
(is (Arrays/equals ^bytes resp ^bytes file-bytes)))))
(range 20)))))
reborg/parallel
(require '[parallel.core :as p])
(require '[criterium.core :refer [bench quick-bench]])
(require '[clojure.core.reducers :as r])
(time (dorun (frequencies bigger-data)))
;; 4320.984379 ms
(time (dorun (p/frequencies bigger-data)))
;; 1980.512017 ms
muoncore/photon
(ns photon.current.projections
(:require [muon-clojure.core :as cl]
[photon.muon :as muon]
[photon.config :as conf]
[photon.api :as api]
[clojure.tools.logging :as log]
[clojure.core.async :refer [<!!]]
[com.stuartsierra.component :as component]
[photon.current.common :refer :all])
(:use midje.sweet))
(let [uuid (java.util.UUID/randomUUID)
ms (new-server uuid)
s-name (str "photon-integration-test-" uuid)
url-req (str "request://" s-name)
url-str (str "stream://" s-name)
m (cl/muon-client :local (str "client-" uuid)
"client" "test")
_ (Thread/sleep 5000)
spn (cl/with-muon m
(cl/request! (str url-req "/projection")
{:projection-name "idontexist"}))
sp (cl/with-muon m
(cl/request! (str url-req "/projection")
{:projection-name "__streams__"}))]
(fact "There is a default projection loaded with photon"
sp => truthy)
(fact "But there are no 'imaginary' projections"
spn => falsey)
(log/info (with-out-str (clojure.pprint/pprint sp)))
(fact "The streams projection indicates 2 event processed"
(:processed sp) => 2)
(post-one-event m s-name)
(Thread/sleep 2000)
(let [new-sp (cl/with-muon m
(cl/request! (str url-req "/projection")
{:projection-name "__streams__"}))]
(log/info (with-out-str (clojure.pprint/pprint new-sp)))
(fact "Now there are three events processed"
(:processed new-sp) => 3))
#_(let [sn (cl/with-muon m
(cl/subscribe! (str url-str "/projection/imaginary")
{:from 0 :stream-type :hot}))]
(fact "There is no stream for a non-existing projection"
(class (<!! sn)) => io.muoncore.exception.MuonException))
(let [s (cl/with-muon m
(cl/subscribe! (str url-str "/projection/__streams__")
{:from 0 :stream-type :hot}))]
(log/info "Subscribing")
(Thread/sleep 5000)
(post-one-event m s-name)
(log/info "Posted")
(fact "Four events in projection" (:processed (<!! s)) => 4)
(log/info "To next subscription")
(let [s2 (cl/with-muon m
(cl/subscribe! (str url-str "/projection/__streams__")
{:from 0 :stream-type :hot
:stream-name "__all__"}))]
(log/info "Subscribing")
(Thread/sleep 5000)
(post-one-event m s-name)
(log/info "Posted")
(let [val2 (<!! s2)]
(fact "Two streams receive the same result..."
(:processed (<!! s)) => (:processed val2))
(fact "... and that result is 5.0" (:processed val2) => 5))))
(cl/with-muon m (cl/request! (str url-req "/projections")
{:projection-name "dummy-proj"
:stream-name "dummy"
:language "clojure"
:reduction "(fn [a b] (inc a))"
:initial-value "0"}))
(cl/with-muon m (cl/request! (str url-req "/projections")
{:projection-name "chatter-proj"
:stream-name "chatter"
:language "clojure"
:reduction "(fn [a b] (inc a))"
:initial-value "0"}))
(let [res (cl/with-muon m
(cl/request! (str url-req "/projection-keys") {}))
sd (cl/with-muon m
(cl/subscribe! (str url-str "/projection/dummy-proj")
{:from 0 :stream-type :hot
:stream-name "__all__"}))
sc (cl/with-muon m
(cl/subscribe! (str url-str "/projection/chatter-proj")
{:from 0 :stream-type :hot
:stream-name "__all__"}))]
(Thread/sleep 5000)
(fact "There is now a chatter count projection"
(contains? (into #{} (:projection-keys res)) "chatter-proj")
=> true)
(fact "There is now a dummy count projection"
(contains? (into #{} (:projection-keys res)) "dummy-proj")
=> true)
(post-one-event m s-name)
(let [val (time-limited 3000 (<!! sc))]
(fact "There are 4 events processed in chatter-proj"
(:current-value val) => 4)
(dorun (take 1000 (repeatedly #(post-one-event m s-name))))
(Thread/sleep 5000)
(let [res (cl/with-muon m
(cl/request! (str url-req "/projection")
{:projection-name "chatter-proj"}))]
(fact (:current-value res) => 1004))
(let [res (cl/with-muon m
(cl/request! (str url-req "/projection")
{:projection-name "chatter-proj"
:query-key "current-value"}))]
(fact res => 1004)))
(fact "The dummy-proj stream has generated nothing as of yet"
(time-limited 3000 (<!! sd)) => (throws Exception))
logaan/promise-stream
(ns promise-stream.pstream-test
(:use-macros [promise-stream.macros :only [for-pstream]])
(:use [jayq.util :only [log]]
[promise-stream.pstream :only
[closed-pstream open-pstream append! close! reduce* map* mapd* concat*
with-open-pstream mapcat* count* resolves-within?
pairwise-traverse zip* promise fmap filter* rests* reductions* doall*
dorun*]])
(:require [jayq.core :as jq]
[promise-stream.pcell :as pc]
[clojure.core.reducers :as r]))
(assert (nil? (dorun* (r/map (pc/dapply inc) (closed-pstream 1 2 3 4)))))