@@ -71,6 +71,23 @@ foreign import connectImpl :: Producer -> Effect (Promise Unit)
7171connect :: Producer -> Aff Unit
7272connect = connectImpl >>> toAffE
7373
74+foreign import data Transaction :: Type
75+
76+foreign import transactionImpl :: Producer -> Effect (Promise Transaction)
77+
78+transaction :: Producer -> Aff Transaction
79+transaction = transactionImpl >>> toAffE
80+
81+foreign import commitImpl :: Transaction -> Effect (Promise Unit)
82+
83+commit :: Transaction -> Aff Unit
84+commit = commitImpl >>> toAffE
85+
86+foreign import abortImpl :: Transaction -> Effect (Promise Unit)
87+
88+abort :: Transaction -> Aff Unit
89+abort = abortImpl >>> toAffE
90+
7491type Message
7592 = { key :: Maybe String
7693 , value :: String
@@ -97,14 +114,20 @@ foreign import data RecordMetadata :: Type
97114
98115foreign import sendImpl :: Fn2 Producer InternalPayload (Effect (Promise (Array RecordMetadata)))
99116
117+foreign import sendTImpl :: Fn2 Transaction InternalPayload (Effect (Promise (Array RecordMetadata)))
118+
119+convertMessage :: Message -> InternalMessage
120+convertMessage { key, value, partition } = { key: toNullable key, value: value, partition: toNullable partition }
121+
122+convert :: Payload -> InternalPayload
123+convert { topic, messages } = { topic, messages: messages <#> convertMessage }
124+
125+
100126send :: Producer -> Payload -> Aff (Array RecordMetadata)
101127send p pl = runFn2 sendImpl p (convert pl) # toAffE
102− where
103− convertMessage :: Message -> InternalMessage
104− convertMessage { key, value, partition } = { key: toNullable key, value: value, partition: toNullable partition }
105128
106− convert :: Payload -> InternalPayload
107− convert { topic, messages } = { topic, messages: messages <#> convertMessage }
129+sendT :: Transaction -> Payload -> Aff (Array RecordMetadata)
130+sendT t pl = runFn2 sendTImpl t (convert pl) # toAffE
108131
109132foreign import disconnectImpl :: Producer -> Effect (Promise Unit)
110133