git.iko.soy/rowtype-yoga/purescript-kafkajs
on main help sign in
rowtype-yoga/purescript-kafkajs/src/Kafka/Consumer.js
33 lines 1.2 kB history raw
3exports.makeConsumerImpl = (kafka, consumerConfig) => kafka.consumer(consumerConfig)
5exports.connectImpl = (consumer) => () => consumer.connect()
7exports.subscribeImpl = (consumer, subscriptionConfig) => () => consumer.subscribe(subscriptionConfig)
9exports.eachBatchImpl = (consumer, eachBatchAutoResolve, handler) => () => {
10 return consumer.run({
11 eachBatchAutoResolve: eachBatchAutoResolve,
12 eachBatch: ({
13 batch,
14 resolveOffset,
15 heartbeat,
16 commitOffsetsIfNecessary,
17 uncommittedOffsets,
18 isRunning,
19 isStale }) => {
20 const resolveOffsetEff = offset => () => resolveOffset(offset)
21 const commitOffsetsIfNecessaryEff = () => commitOffsetsIfNecessary()
22 const uncommittedOffsetsEff = () => uncommittedOffsets()
23 const isRunningEff = () => isRunning()
24 const isStaleEff = () => isStale()
26 return handler(batch, resolveOffsetEff, heartbeat, commitOffsetsIfNecessaryEff, uncommittedOffsetsEff, isRunningEff, isStaleEff)()
27 }
29 })
32exports.disconnectImpl = (consumer) => () => consumer.disconnect()