git.iko.soy/rowtype-yoga/purescript-kafkajs
on main help sign in
rowtype-yoga/purescript-kafkajs/src/Kafka/Transaction.purs
63 lines 1.7 kB history raw
1module Kafka.Transaction
2 ( Transaction
3 , transaction
4 , commit
5 , abort
6 , send
7 , sendOffsets
8 , SendOffsets
9 , TopicCommit
10 , PartitionCommit
11 ) where
13import Control.Promise (Promise, toAffE)
14import Data.Function.Uncurried (Fn2, runFn2)
15import Effect (Effect)
16import Effect.Aff (Aff)
17import Kafka.Consumer (GroupId)
18import Kafka.Producer (InternalProducerBatch, Producer, ProducerBatch, RecordMetadata, toInternalProducerBatch)
19import Kafka.Types (Offset, Partition, Topic)
20import Prelude (Unit, (#), (>>>))
22foreign import data Transaction :: Type
24foreign import transactionImpl :: Producer -> Effect (Promise Transaction)
26transaction :: Producer -> Aff Transaction
27transaction = transactionImpl >>> toAffE
29type PartitionCommit = {
30 partition:: Partition,
31 offset :: Offset
32 }
34type TopicCommit = {
35 topic :: Topic,
36 partitions :: Array PartitionCommit
39type SendOffsets = {
40 consumerGroupId :: GroupId,
41 topics :: Array TopicCommit
44foreign import sendOffsetsImpl :: Fn2 Transaction SendOffsets (Effect (Promise Unit))
46sendOffsets :: Transaction -> SendOffsets -> Aff Unit
47sendOffsets t so = runFn2 sendOffsetsImpl t so # toAffE
49foreign import sendImpl :: Fn2 Transaction InternalProducerBatch (Effect (Promise (Array RecordMetadata)))
51send :: Transaction -> ProducerBatch -> Aff (Array RecordMetadata)
52send t pl = runFn2 sendImpl t (toInternalProducerBatch pl) # toAffE
54foreign import commitImpl :: Transaction -> Effect (Promise Unit)
56commit :: Transaction -> Aff Unit
57commit = commitImpl >>> toAffE
59foreign import abortImpl :: Transaction -> Effect (Promise Unit)
61abort :: Transaction -> Aff Unit
62abort = abortImpl >>> toAffE