git.iko.soy/rowtype-yoga/purescript-kafkajs
on main help sign in
Add transactional id
f4382dfhivemind-js
src/Kafka.purs+18−7
compare with main
unifiedsplit
@@ -52,7 +52,14 @@ type KafkaConfig
52foreign import data Producer :: Type
52foreign import data Producer :: Type
53
53
54type ProducerConfig
54type ProducerConfig
55+ = { idempotent :: Maybe Boolean
56+ , transactionalId :: Maybe String
57+ }
58+
59+type InternalProducerConfig
60+ = { idempotent :: Nullable Boolean
61+ , transactionalId :: Nullable String
55− = {}
62+ }
56
63
57foreign import makeClientImpl :: InternalKafkaConfig -> Kafka
64foreign import makeClientImpl :: InternalKafkaConfig -> Kafka
58
65
@@ -66,17 +73,22 @@ makeClient = toInternal >>> makeClientImpl
66 , sasl: toNullable config.sasl
73 , sasl: toNullable config.sasl
67 }
74 }
68
75
69−foreign import makeProducerImpl :: Fn2 Kafka ProducerConfig Producer
76+foreign import makeProducerImpl :: Fn2 Kafka InternalProducerConfig Producer
70
77
71makeProducer :: Kafka -> ProducerConfig -> Producer
78makeProducer :: 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+ }
73
85
74foreign import connectImpl :: Producer -> Effect (Promise Unit)
86foreign import connectImpl :: Producer -> Effect (Promise Unit)
75
87
76connect :: Producer -> Aff Unit
88connect :: Producer -> Aff Unit
77connect = connectImpl >>> toAffE
89connect = connectImpl >>> toAffE
78
90
79−foreign import data Transaction :: Type
91+foreign import data Transaction :: Type
80
92
81foreign import transactionImpl :: Producer -> Effect (Promise Transaction)
93foreign import transactionImpl :: Producer -> Effect (Promise Transaction)
82
94
@@ -85,12 +97,12 @@ transaction = transactionImpl >>> toAffE
85
97
86foreign import commitImpl :: Transaction -> Effect (Promise Unit)
98foreign import commitImpl :: Transaction -> Effect (Promise Unit)
87
99
88−commit :: Transaction -> Aff Unit
100+commit :: Transaction -> Aff Unit
89commit = commitImpl >>> toAffE
101commit = commitImpl >>> toAffE
90
102
91foreign import abortImpl :: Transaction -> Effect (Promise Unit)
103foreign import abortImpl :: Transaction -> Effect (Promise Unit)
92
104
93−abort :: Transaction -> Aff Unit
105+abort :: Transaction -> Aff Unit
94abort = abortImpl >>> toAffE
106abort = abortImpl >>> toAffE
95
107
96type Message
108type Message
@@ -127,7 +139,6 @@ convertMessage { key, value, partition } = { key: toNullable key, value: value,
127convert :: Payload -> InternalPayload
139convert :: Payload -> InternalPayload
128convert { topic, messages } = { topic, messages: messages <#> convertMessage }
140convert { topic, messages } = { topic, messages: messages <#> convertMessage }
129
141
130−
131send :: Producer -> Payload -> Aff (Array RecordMetadata)
142send :: Producer -> Payload -> Aff (Array RecordMetadata)
132send p pl = runFn2 sendImpl p (convert pl) # toAffE
143send p pl = runFn2 sendImpl p (convert pl) # toAffE
133
144