@@ -52,7 +52,14 @@ type KafkaConfig
5252foreign import data Producer :: Type
5353
5454type ProducerConfig
55− = {}
55+ = { idempotent :: Maybe Boolean
56+ , transactionalId :: Maybe String
57+ }
58+
59+type InternalProducerConfig
60+ = { idempotent :: Nullable Boolean
61+ , transactionalId :: Nullable String
62+ }
5663
5764foreign import makeClientImpl :: InternalKafkaConfig -> Kafka
5865
@@ -66,17 +73,22 @@ makeClient = toInternal >>> makeClientImpl
6673 , sasl: toNullable config.sasl
6774 }
6875
69−foreign import makeProducerImpl :: Fn2 Kafka ProducerConfig Producer
76+foreign import makeProducerImpl :: Fn2 Kafka InternalProducerConfig Producer
7077
7178makeProducer :: Kafka -> ProducerConfig -> Producer
72−makeProducer = runFn2 makeProducerImpl
79+makeProducer k pc = runFn2 makeProducerImpl k ipc
80+ where
81+ ipc =
82+ { idempotent: toNullable pc.idempotent
83+ , transactionalId: toNullable pc.transactionalId
84+ }
7385
7486foreign import connectImpl :: Producer -> Effect (Promise Unit)
7587
7688connect :: Producer -> Aff Unit
7789connect = connectImpl >>> toAffE
7890
79−foreign import data Transaction :: Type
91+foreign import data Transaction :: Type
8092
8193foreign import transactionImpl :: Producer -> Effect (Promise Transaction)
8294
@@ -85,12 +97,12 @@ transaction = transactionImpl >>> toAffE
8597
8698foreign import commitImpl :: Transaction -> Effect (Promise Unit)
8799
88−commit :: Transaction -> Aff Unit
100+commit :: Transaction -> Aff Unit
89101commit = commitImpl >>> toAffE
90102
91103foreign import abortImpl :: Transaction -> Effect (Promise Unit)
92104
93−abort :: Transaction -> Aff Unit
105+abort :: Transaction -> Aff Unit
94106abort = abortImpl >>> toAffE
95107
96108type Message
@@ -127,7 +139,6 @@ convertMessage { key, value, partition } = { key: toNullable key, value: value,
127139convert :: Payload -> InternalPayload
128140convert { topic, messages } = { topic, messages: messages <#> convertMessage }
129141
130−
131142send :: Producer -> Payload -> Aff (Array RecordMetadata)
132143send p pl = runFn2 sendImpl p (convert pl) # toAffE
133144