protocols.clj

(ns jackdaw.streams.protocols
  "Kafka streams protocols."
  {:license "BSD 3-Clause License <https://github.com/FundingCircle/jackdaw/blob/master/LICENSE>"}
  (:refer-clojure :exclude [count map merge reduce group-by filter peek])
  (:import org.apache.kafka.streams.KafkaStreams
           org.apache.kafka.streams.StreamsBuilder))

(set! *warn-on-reflection* true)

(defprotocol IStreamsBuilder
  (kstream
    [topology-builder topic-config]
    [topology-builder topic-config topic-pattern]
    "Creates a KStream that will consume messages from the specified topic.")

  (kstreams
    [topology-builder topic-configs]
    "Creates a KStream that will consume messages from the specified topics.")

  (ktable
    [topology-builder topic-config]
    [topology-builder topic-config store-name]
    "Creates a KTable that will consist of data from the specified topic.")

  (global-ktable
    [topology-builder topic-config]
    "Creates a GlobalKTable that will consist of data from the specified
    topic.")

  (source-topics
    [topology-builder]
    "Gets the names of source topics for the topology.")

  (streams-builder*
    [streams-builder]
    "Returns the underlying KStreamBuilder."))

(defprotocol IKStreamBase
  "Methods common to KStream and KTable."
  (left-join
    [kstream-or-ktable ktable value-joiner-fn]
    [kstream-or-ktable ktable value-joiner-fn this-topic-config other-topic-config]
    [kstream-or-ktable ktable value-joiner-fn this-topic-config other-topic-config join-name]
    "Creates a KStream from the result of calling `value-joiner-fn` with
    each element in the KStream and the value in the KTable with the same
    key.")

  (filter
    [kstream-or-ktable predicate-fn]
    "Creates a KStream that consists of all elements that satisfy a
    predicate.")

  (filter-not
    [kstream-or-ktable predicate-fn]
    "Creates a KStream that consists of all elements that do not satisfy a
    predicate.")

  (group-by
    [ktable-or-ktable key-value-mapper-fn]
    [ktable-or-ktable key-value-mapper-fn topic-config]
    "Groups the records of this KStream/KTable using the key-value-mapper-fn.")

  (peek
    [kstream-or-ktable peek-fn]
    "Performs `peek-fn` on each element of the input stream.")

  (map-values
    [kstream-or-ktable value-mapper-fn]
    "Creates a KStream that is the result of calling `value-mapper-fn` on each
    element of the input stream.")

  (write-as-text!
    [kstream-or-ktable file-path]
    [kstream-or-ktable file-path topic-config]
    "Writes the elements of a stream to a file at the given path."))

(defprotocol IKStreamA
  "A KStream is an abstraction of a stream of key-value pairs."
  (branch
    [kstream predicate-fns]
    [kstream name predicate-fns]
    "Returns a list of KStreams, one for each of the `predicate-fns`
    provided.")

  (flat-map
    [kstream key-value-mapper-fn]
    "Creates a KStream that will consist of the concatenation of messages
    returned by calling `key-value-mapper-fn` on each key/value pair in the
    input stream.")

  (flat-map-values
    [kstream value-mapper-fn]
    "Creates a KStream that will consist of the concatenation of the values
    returned by calling `value-mapper-fn` on each value in the input stream.")

  (for-each!
    [kstream foreach-fn]
    "Performs an action on each element of KStream.")

  (print!
    [kstream]
    "Prints the elements of the stream to *out*.")

  (through
    [kstream topic-config]
    "Materializes a stream to a topic, and returns a new KStream that will
    consume messages from the topic. Messages in the new topic will be partitioned
    based on the output of the optional partition function that represents StreamPartitioner class")

  (to!
    [kstream topic-config]
    "Materializes a stream to a topic.")

  (group-by-key
    [kstream]
    [kstream topic-config]
    "Groups records with the same key into a KGroupedStream."))

(defprotocol IKStreamB
  (join-windowed
    [kstream other-kstream value-joiner-fn windows]
    [kstream other-kstream value-joiner-fn windows this-topic-config other-topic-config]
    [kstream other-kstream value-joiner-fn windows this-topic-config other-topic-config join-name]
    "Combines the values of two streams that share the same key using a
    windowed inner join.")

  (left-join-windowed
    [kstream other-kstream value-joiner-fn windows]
    [kstream other-kstream value-joiner-fn windows this-topic-config other-topic-config]
    [kstream other-kstream value-joiner-fn windows this-topic-config other-topic-config join-name]
    "Combines the values of two streams that share the same key using a
    windowed left join.")

  (map
    [kstream key-value-mapper-fn]
    "Creates a KStream that consists of the result of applying
    `key-value-mapper-fn` to each key/value pair in the input stream.")

  (merge
    [kstream other]
    "Creates a KStream that has the records from both streams.")

  (outer-join-windowed
    [kstream other-kstream value-joiner-fn windows]
    [kstream other-kstream value-joiner-fn windows this-topic-config other-topic-config]
    [kstream other-kstream value-joiner-fn windows this-topic-config other-topic-config join-name]
    "Combines the values of two streams that share the same key using a
    windowed outer join.")

  (process!
    [kstream processor-fn state-store-names]
    "Applies `processor-fn` to each item in the input stream.")

  (select-key
    [kstream select-key-value-mapper-fn]
    "Create a new key from the current key and value.

    `select-key-value-mapper-fn` should be a function that takes a key-value
    pair, and returns the value of the new key. Here is example multiplies each
    key by 10:

    ```(fn [[k v]] (* 10 k))```")

  (transform
    [kstream transformer-supplier-fn]
    [kstream transformer-supplier-fn state-store-names]
    "Creates a KStream that consists of the results of applying the transformer
    to each key/value in the input stream.")

  (transform-values
    [kstream value-transformer-supplier-fn]
    [kstream value-transformer-supplier-fn state-store-names]
    "Creates a KStream that consists of the results of applying the transformer
    to each value in the input stream.")

  (join-global
    [kstream global-ktable kv-mapper joiner])

  (left-join-global
    [kstream global-ktable kv-mapper joiner]
    [kstream global-ktable kv-mapper joiner join-name])

  (kstream*
    [kstream]
    "Returns the underlying KStream object."))

(defprotocol IKTable
  "A Ktable is an abstraction of a changlog stream."
  (join
    [ktable other-ktable value-joiner-fn]
    "Combines the values of the two KTables that share the same key using an
    inner join.")

  (outer-join
    [ktable other-ktable value-joiner-fn]
    "Combines the values of two KTables that share the same key using an outer
    join.")

  (to-kstream
    [ktable]
    [ktable key-value-mapper-fn]
    "Converts a KTable to a KStream.")

  (suppress
    [ktable {:keys [max-records max-bytes until-time-limit-ms]}]
    "Suppress some updates from this changelog stream.
    You can either specify `max-records` or `max-bytes`. If an empty map is
    passed, the suppress will be unbounded. If `until-time-limit-ms` is set,
    this will override the `TimeWindow` interval. Note that when relying on the
    configured `TimeWindow` the default `grace` period is `24h - window-size`.")

  (ktable*
    [ktable]
    "Returns the underlying KTable object."))

(defprotocol IKGroupedBase
  "Methods shared between `IKGroupedTable` and `IKGroupedStream`."
  (aggregate
    [kgrouped initializer-fn adder-fn subtractor-or-merger-fn topic-config]
    [kgrouped initializer-fn aggregator-fn subtractor-fn-or-topic-config]
    [kgrouped initializer-fn aggregator-fn]
    "Aggregates values by key into a new KTable.")

  (count
    [kgrouped]
    [kgrouped topic-config]
    "Counts the number of records by key into a new KTable.")

  (reduce
    [kgrouped adder-fn subtractor-fn topic-config]
    [kgrouped reducer-fn subtractor-fn-or-topic-config]
    [kgrouped reducer-fn]
    "Combines values of a stream by key into a new KTable."))

(defprotocol IKGroupedTable
  "KGroupedTable is an abstraction of a grouped changelog stream."
  (kgroupedtable*
    [kgroupedtable]
    "Returns the underlying KGroupedTable object."))

(defprotocol IKGroupedStream
  "KGroupedStream is an abstraction of a grouped stream."
  (windowed-by-time [kgroupedstream window])

  (windowed-by-session [kgroupedstream window])

  (kgroupedstream*
    [kgroupedstream]
    "Returns the underlying KGroupedStream object."))

(defprotocol ITimeWindowedKStream
  "ITimeWindowedKStream is an abstraction of a time windowed stream."

  (time-windowed-kstream*
    [ktime-windowed-kstream]
    "Returns the underlying TimeWindowedKStream object."))

(defprotocol ISessionWindowedKStream
  "ISessionWindowedKStream is an abstraction of a session windowed stream."

  (session-windowed-kstream*
    [ksession-windowed-kstream]
    "Returns the underlying SessionWindowedKStream object."))

(defprotocol IGlobalKTable
  (global-ktable*
    [globalktable]
    "Returns the underlying GlobalKTable"))