prpr.stream.cross — employeerepublic/promisespromises 2.5.0.1

advance-stream-buffers clj

(advance-stream-buffers key-compare-fn
                        skey-streams
                        skey-streambufs
                        next-skey-set)

given a map {skey stream}, a map {skey [[hk hvs] [nk nk-buf]]}, and a set next-skey-set of skeys to update the stream-buffers for, async fetch next values for the streams with the given keys, returning an updated map {skey [[nk nvs] [nnk nnk-buf]]}

buffer-values clj

(buffer-values key-compare-fn stream-key stream [sort-key buf-v])

given a stream and a [k buf] structure, retrieve values from the stream matching key k into buf, until a value not matching k is encountered. returns Deferred<[[k updated-buf] [nk nk-buf]]>

close-all-streams clj

(close-all-streams & streams)

coerce-linked-map clj

(coerce-linked-map m)

coerce a map to a linked/map with reliable insertion-based ordering

cross-streams clj

(cross-streams {id :id
                key-compare-fn :key-compare-fn
                selector-fn :selector-fn
                finish-merge-fn :finish-merge-fn
                product-sort-fn :product-sort-fn
                skey-streams :skey-streams
                :as args})

join values from some sorted streams onto a new sorted stream

given keyed streams in {skey stream} values with the same key (as determined by -key) will be take from the head of each stream, and the values from each stream with the lowest key value (keys being compared with key-compare-fn) will be passed to the selector-fn which selects one or more of them, as input to a cartesian product of the selected values from each stream. the output tuples of the cartesian product are then merged into a stream of output values and the stream of output values is passed to product-sort-fn to be sorted into the correct result order

different selector-fns and ISortedStream impls can give different behaviours, such as sort-merge or join

event-source->sorted-stream clj

(event-source->sorted-stream default-key-fn stream)

if stream is already a SortedStream, return it unchanged. if it's an IEventSource create a SortedStream with the supplied default-key-fn

full-outer-join-streams clj

(full-outer-join-streams {key-compare-fn :key-compare-fn
                          default-key-fn :default-key-fn
                          finish-merge-fn :finish-merge-fn
                          product-sort-fn :product-sort-fn
                          skey-streams :skey-streams
                          :or {key-compare-fn compare default-key-fn identity}
                          :as args})

full-outer-joins records from multiple streams, which must already be sorted by key. 1-1, 1-many, many-1 and many-many joins are all supported. all elements from all streams for each join key will be buffered in memory

head-values-cartesian-product-merge clj

(head-values-cartesian-product-merge init-output-value
                                     merge-fn
                                     finish-merge-fn
                                     product-sort-fn
                                     skey-streams
                                     skey-values)

given a map skey-values {skey [val+]} of vectors of records for the same key from different streams, first produce a cartesian product and then merge each product value to a single value, finishing off each merge with finish-merge-fn and then sorting the merged values with product-sort-fn

init-stream-buffers clj

(init-stream-buffers key-compare-fn skey-streams)

given a map {skey stream} async fetch values from each stream returning a map {skey [[head-key head-values] [next-key first-next-value]]}. this sets up the skey-streambufs structure for next-output-values

inner-join-streams clj

(inner-join-streams {key-compare-fn :key-compare-fn
                     default-key-fn :default-key-fn
                     finish-merge-fn :finish-merge-fn
                     product-sort-fn :product-sort-fn
                     skey-streams :skey-streams
                     :or {key-compare-fn compare default-key-fn identity}
                     :as args})

ISortedStream clj protocol

defines how to handle values on a sorted stream

-close! clj

(-close! s)

close the stream

-key clj

(-key _ v)

extracts the sort-key from the value - the head-values with the lowest sort-key will be given to the selector-fn to select which values will compose the next output-value

-stream clj

(-stream this)

returns the manifold stream

-take! clj

(-take! _)

(-take! _ default-val)

take a value from the stream

merge-stream-objects clj

(merge-stream-objects init-output-value merge-fn skey-streams skey-vals)

given a seq of [skey val] objects, merge the vals using the -merge operation defined on the stream from skey-streams {skey stream}

min-key-skey-values clj

(min-key-skey-values key-compare-fn skey-streams skey-streambufs)

returns a map {skey [val+]} with entries for each stream which has a head buffer with vals matching the minimum value -key across all streams. uses key-compare-fn to compare value -keys

min-key-val clj

(min-key-val key-compare-fn ks)

uses the comparator to find the minimum key value from ks

n-left-join-streams clj

(n-left-join-streams {key-compare-fn :key-compare-fn
                      default-key-fn :default-key-fn
                      finish-merge-fn :finish-merge-fn
                      product-sort-fn :product-sort-fn
                      skey-streams :skey-streams
                      n :n
                      :or {key-compare-fn compare default-key-fn identity n 1}
                      :as args})

next-output-values clj

(next-output-values {key-compare-fn :key-compare-fn
                     selector-fn :selector-fn
                     finish-merge-fn :finish-merge-fn
                     product-sort-fn :product-sort-fn
                     skey-streams :skey-streams
                     skey-streambufs :skey-streambufs
                     :as args})

given skey-streams and the current head values from those streams in skey-streambufs, offers the {skey head-value}s with the lowest (according to key-compare-fn) key values to the select-skey-values which uses the selector-fn to choose the same-key head-values from one or more stream

then a cartesian product of selected values for each stream are merged to give the output-values and the streambufs are updated accordingly

• returns [output-value next-skey-streambufs] or [::drained]

select-all clj

(select-all skey-head-values)

selector-fn which takes all offered elements

select-first clj

(select-first skey-head-values)

selector-fn which takes the first element from the offered set of elements

select-skey-values clj

(select-skey-values selector-fn skey-values)

uses the selector-fn to select 1 or more values from an {skey [values...]} map. errors if there are values to select and the selector-fn doesn't do something sensible

set-select-all clj

(set-select-all skey-head-values)

set-streams-difference clj

(set-streams-difference {key-compare-fn :key-compare-fn
                         default-key-fn :default-key-fn
                         finish-merge-fn :finish-merge-fn
                         product-sort-fn :product-sort-fn
                         skey-streams :skey-streams
                         :or {key-compare-fn compare default-key-fn identity}
                         :as args})

set-streams-intersect clj

(set-streams-intersect {key-compare-fn :key-compare-fn
                        default-key-fn :default-key-fn
                        finish-merge-fn :finish-merge-fn
                        product-sort-fn :product-sort-fn
                        skey-streams :skey-streams
                        :or {key-compare-fn compare default-key-fn identity}
                        :as args})

set-streams-union clj

(set-streams-union {key-compare-fn :key-compare-fn
                    default-key-fn :default-key-fn
                    finish-merge-fn :finish-merge-fn
                    product-sort-fn :product-sort-fn
                    skey-streams :skey-streams
                    :or {key-compare-fn compare default-key-fn identity}
                    :as args})

sort-merge-streams clj

(sort-merge-streams {key-compare-fn :key-compare-fn
                     default-key-fn :default-key-fn
                     skey-streams :skey-streams
                     :or {key-compare-fn compare default-key-fn identity}
                     :as args})

the merge phase of a merge-sort, with streams.

given some streams each with content sorted by a key (extracted by key-fn), produces a new stream with all elements from the input streams, also sorted by the same key

for elements with the same key, skey-streams order is used

sorted-stream clj

(sorted-stream key-fn stream)

create an ISortedStream from a manifold stream with a supplied key-fn. if the stream is already a SortedStream, update its key-fn

warn-full clj

(warn-full e msg)

warn with an error, tag and value