git.iko.soy/rowtype-yoga/purescript-kafkajs
on main help sign in
rowtype-yoga/purescript-kafkajs/src/Kafka/Consumer.purs
293 lines 7.2 kB history raw
1module Kafka.Consumer
2 ( Consumer
3 , GroupId(..)
4 , ConsumerConfig
5 , SubscriptionConfig
6 , makeConsumer
7 , connect
8 , subscribe
9 , ConsumerMessage
10 , ConsumerBatch
11 , Heartbeat
12 , ResolveOffset
13 , CommitOffsetsIfNecessary
14 , OffsetInfo
15 , toOffsetInfo
16 , UncommittedOffsets
17 , IsStale
18 , IsRunning
19 , EachBatch
20 , eachBatch
21 , disconnect
22 ) where
24import Prelude (Unit, bind, (#), ($), (>>>))
25import Control.Promise (Promise, toAffE, fromAff)
26import Kafka.Kafka (Kafka)
27import Data.Function.Uncurried (Fn2, Fn3, Fn7, mkFn7, runFn2, runFn3)
28import Effect (Effect)
29import Effect.Aff (Aff)
30import Data.Show (class Show)
31import Kafka.Types
32import Data.Nullable (Nullable, toMaybe)
33import Node.Buffer (Buffer)
34import Data.Maybe (Maybe, maybe)
35import Data.Traversable (traverse)
36import Control.Monad.Error.Class (throwError)
37import Effect.Exception (error)
38import Data.Int (fromString)
39import Control.Applicative (pure)
40import Data.Either (Either, either, note)
41import Data.Newtype (un, class Newtype)
43foreign import data Consumer :: Type
45newtype GroupId
46 = GroupId String
48instance ntGroupId :: Newtype GroupId String
50derive newtype instance showGroupId :: Show GroupId
52type ConsumerConfig
53 = { groupId :: GroupId
54 , readUncommitted :: Boolean
55 , autoCommit :: Boolean
56 }
58type SubscriptionConfig
59 = { topic :: String }
61foreign import makeConsumerImpl :: Fn2 Kafka ConsumerConfig Consumer
63makeConsumer :: Kafka -> ConsumerConfig -> Consumer
64makeConsumer = runFn2 makeConsumerImpl
66foreign import connectImpl :: Consumer -> Effect (Promise Unit)
68connect :: Consumer -> Aff Unit
69connect = connectImpl >>> toAffE
71foreign import subscribeImpl :: Fn2 Consumer SubscriptionConfig (Effect (Promise Unit))
73subscribe :: Consumer -> SubscriptionConfig -> Aff Unit
74subscribe consumer subscriptionConfig = runFn2 subscribeImpl consumer subscriptionConfig # toAffE
76type InternalConsumerMessage
77 = { key :: Nullable Buffer
78 , value :: Buffer
79 , offset :: String
80 }
82type InternalConsumerBatch
83 = { fetchedOffset :: String
84 , highWatermark :: String
85 , messages :: Array InternalConsumerMessage
86 , partition :: Int
87 , topic :: String
88 }
90type ConsumerMessage
91 = { key :: Maybe Buffer
92 , value :: Buffer
93 , offset :: Offset
94 }
96type ConsumerBatch
97 = { fetchedOffset :: Offset
98 , highWatermark :: Offset
99 , messages :: Array ConsumerMessage
100 , partition :: Partition
101 , topic :: Topic
102 }
104type InternalHeartbeat
105 = Effect (Promise Unit)
107type Heartbeat
108 = Aff Unit
110type InternalResolveOffset
111 = Int -> Effect Unit
113type ResolveOffset
114 = Offset -> Effect Unit
116type InternalCommitOffsetsIfNecessary
117 = Effect (Promise Unit)
119type CommitOffsetsIfNecessary
120 = Aff Unit
122type InternalOffsetInfo
123 = { topics ::
124 Array
125 { partitions ::
126 Array
127 { offset :: String
128 , partition :: String
129 }
130 , topic :: String
131 }
132 }
134type InternalUncommittedOffsets
135 = Effect InternalOffsetInfo
137type OffsetInfo
138 = { topics ::
139 Array
140 { partitions ::
141 Array
142 { offset :: Offset
143 , partition :: Partition
144 }
145 , topic :: Topic
146 }
147 }
149raiseMaybe :: forall a. Maybe a -> Effect a
150raiseMaybe = maybe (throwError $ error "Maybe is Nothing") pure
152raiseEither :: forall a. Either String a -> Effect a
153raiseEither = either (error >>> throwError) pure
155toOffsetInfo :: InternalOffsetInfo -> Effect OffsetInfo
156toOffsetInfo { topics } = do
157 topicInfo <- traverse parseTopicInfo topics
158 pure { topics: topicInfo }
159 where
160 parsePartitionInfo ::
161 { offset :: String
162 , partition :: String
163 } ->
164 Effect
165 { offset :: Offset
166 , partition :: Partition
167 }
168 parsePartitionInfo { offset, partition } = do
169 o <- raiseMaybe $ fromString offset
170 p <- raiseMaybe $ fromString partition
171 pure { offset: Offset o, partition: Partition p }
173 parseTopicInfo ::
174 { partitions ::
175 Array
176 { offset :: String
177 , partition :: String
178 }
179 , topic :: String
180 } ->
181 Effect
182 { partitions ::
183 Array
184 { offset :: Offset
185 , partition :: Partition
186 }
187 , topic :: Topic
188 }
189 parseTopicInfo { partitions, topic } = do
190 partitionInfo <- traverse parsePartitionInfo partitions
191 pure { partitions: partitionInfo, topic: Topic topic }
193type UncommittedOffsets
194 = Effect OffsetInfo
196type InternalIsStale
197 = Effect Boolean
199type IsStale
200 = Effect Boolean
202type InternalIsRunning
203 = Effect Boolean
205type IsRunning
206 = Effect Boolean
208type EachBatchImpl
209 = Fn7
210 InternalConsumerBatch
211 InternalResolveOffset
212 InternalHeartbeat
213 InternalCommitOffsetsIfNecessary
214 InternalUncommittedOffsets
215 InternalIsRunning
216 InternalIsStale
217 (Effect (Promise Unit))
219type EachBatch
220 = ConsumerBatch ->
221 ResolveOffset ->
222 Heartbeat ->
223 CommitOffsetsIfNecessary ->
224 UncommittedOffsets ->
225 IsRunning ->
226 IsStale ->
227 (Aff Unit)
229toConsumerMessage :: InternalConsumerMessage -> Either String ConsumerMessage
230toConsumerMessage { key, value, offset } = do
231 o <- note "offset is not a number" $ fromString offset
232 pure
233 { key: toMaybe key
234 , value: value
235 , offset: Offset o
236 }
238toConsumerBatch :: InternalConsumerBatch -> Either String ConsumerBatch
239toConsumerBatch { fetchedOffset, highWatermark, messages, partition, topic } = do
240 fo <- note "fetchedOffset is not a number" $ fromString fetchedOffset
241 hwm <- note "highWatermark is not a number" $ fromString highWatermark
242 msgs <- traverse toConsumerMessage messages
243 pure
244 { fetchedOffset: Offset fo
245 , highWatermark: Offset hwm
246 , messages: msgs
247 , partition: Partition partition
248 , topic: Topic topic
249 }
251foreign import eachBatchImpl :: Fn3 Consumer Boolean EachBatchImpl (Effect (Promise Unit))
253eachBatch :: Consumer -> Boolean -> EachBatch -> Aff Unit
254eachBatch consumer eachBatchAutoResolve handler =
255 let
256 iResolveOffset :: (Int -> Effect Unit) -> (Offset -> Effect Unit)
257 iResolveOffset f = un Offset >>> f
259 iUncommitedOffsets :: Effect InternalOffsetInfo -> Effect OffsetInfo
260 iUncommitedOffsets ioiEff = do
261 ioi <- ioiEff
262 toOffsetInfo ioi
264 internalHandlerCurried ::
265 InternalConsumerBatch ->
266 InternalResolveOffset ->
267 InternalHeartbeat ->
268 InternalCommitOffsetsIfNecessary ->
269 InternalUncommittedOffsets ->
270 InternalIsRunning ->
271 InternalIsStale ->
272 (Effect (Promise Unit))
273 internalHandlerCurried icb iro ih icoin iuo iir iis = do
274 batch <- raiseEither (toConsumerBatch icb)
275 handler
276 batch
277 (iResolveOffset iro)
278 (toAffE ih)
279 (toAffE icoin)
280 (iUncommitedOffsets iuo)
281 (iir)
282 (iis)
283 # fromAff
285 internalHandler :: EachBatchImpl
286 internalHandler = mkFn7 internalHandlerCurried
287 in
288 runFn3 eachBatchImpl consumer eachBatchAutoResolve internalHandler # toAffE
290foreign import disconnectImpl :: Consumer -> Effect (Promise Unit)
292disconnect :: Consumer -> Aff Unit
293disconnect = disconnectImpl >>> toAffE