ottla.postgresql — com.joshuadavey/ottla 0.4.0

brin-index-template clj

btree-key-index-template clj

column-types clj

commit-modes clj

commit-offset! clj

(commit-offset! {:keys [conn schema]} {:keys [topic group]} cursor)

connect-config clj

(connect-config config)

create-topic clj

(create-topic {:keys [conn schema]}
              topic
              &
              {:keys [key-type value-type index-key?]
               :as opts
               :or {key-type :bytea value-type :bytea index-key? false}})

default-subscription-group clj

delete-topic clj

(delete-topic {:keys [conn schema]} topic)

deserializer-xf clj

(deserializer-xf config topic {:keys [deserialize-key deserialize-value]})

ensure-schema clj

(ensure-schema {:keys [conn schema]})

ensure-subscription clj

(ensure-subscription {:keys [conn schema]} {:keys [topic group]})

ensure-topic clj

(ensure-topic {:keys [conn] :as config}
              topic-name
              &
              {:keys [key-type value-type]
               :as opts
               :or {key-type :bytea value-type :bytea}})

Fetch or create a topic. Returns the existing topic if it already exists with the same key-type and value-type. Throws if the topic exists but was created with different column types. Note: opts apply only at creation time and cannot be used to change an existing topic.

fetch-records! clj

(fetch-records! config selection)

fetch-records* clj

(fetch-records* {:keys [conn schema] :as config}
                {:keys [topic group commit-mode] :as selection})

fetch-topic clj

(fetch-topic config topic-name)

insert-records clj

(insert-records {:keys [conn conn-map schema] :as cfg}
                topic
                records
                &
                {:keys [serialize-key serialize-value] :as opts})

legal-identifier? clj

(legal-identifier? s)

list-subscriptions clj

(list-subscriptions {:keys [conn schema]} & [{:keys [topics]}])

list-topics clj

(list-topics {:keys [conn schema]})

maybe-fetch-topic clj

(maybe-fetch-topic {:keys [conn schema]} topic-name)

normalize-selection clj

(normalize-selection selection)

normalize-topic-name clj

(normalize-topic-name topic)

reset-offset! clj

(reset-offset! {:keys [conn schema]} {:keys [topic group]} cursor)

selection-defaults clj

sql-entity clj

(sql-entity x)

topic-subscriptions clj

(topic-subscriptions {:keys [conn] :as config})

Returns all topics with their consumer group subscriptions nested under each topic. Unlike list-subscriptions, every topic appears in the result even if it has no subscribers.

Each entry is a map with:

• :topic the topic name
• :subscriptions a vector of subscription maps (empty when no subscribers)

Each subscription map contains the same keys as list-subscriptions (except :topic, which is on the outer map):

• :group consumer group id
• :offset last committed eid for this group
• :topic-eid highest eid available in the topic
• :lag number of unread records (topic-eid - offset)
• :updated-at java.time.Instant of the last offset commit (nil if never committed)
• :timestamp java.time.Instant of the last consumed record (nil if offset is 0)
• :topic-timestamp java.time.Instant of the latest record in the topic (nil if empty)
• :timestamp-lag java.time.Duration between subscription and topic timestamps (nil if either is nil)
• :processing-delay java.time.Duration from publish time to consumer commit for the most recently consumed record (nil if either :updated-at or :timestamp is nil)

topic-table-name clj

(topic-table-name topic)

trigger-function-name clj

(trigger-function-name schema)

trigger-function-template clj

trigger-template clj

trim-topic clj

(trim-topic {:keys [conn schema] :as config}
            topic
            &
            {:keys [before-eid before-timestamp all? ignore-subscriptions?]
             :or {ignore-subscriptions? false}})