@@ -17,15 +17,14 @@ import Effect.Class (liftEffect)
1717import Effect.Class.Console (logShow)
1818import Node.Buffer (toString)
1919import Node.Encoding (Encoding(..))
20+import Data.Functor (void)
2021
2122printAndResolve :: ResolveOffset -> ConsumerMessage -> Aff Unit
2223printAndResolve resolveOffset { value, offset } = do
23− value <- liftEffect $ toString UTF8 value
24− _ <- logShow $ "Got message: " <> value
24+ v <- liftEffect $ toString UTF8 value
25+ logShow $ "Got message: " <> v
2526 liftEffect $ resolveOffset offset
2627
27−
28−
2928main :: Effect (Promise Unit)
3029main =
3130 fromAff
@@ -33,20 +32,14 @@ main =
3332 let
3433 kafka = makeClient { clientId: "transactional-consumer", brokers: [ "localhost:9092" ], ssl: false, sasl: Nothing }
3534
36− consumerConfig = { groupId: "transactional-group", readUncommitted: false, autoCommit: false }
37−
38− consumer = makeConsumer kafka consumerConfig
35+ consumerConfig = { groupId: "transactional-group", readUncommitted: false, autoCommit: false }
3936
40− connect consumer
37+ consumer = makeConsumer kafka consumerConfig
38+ connect consumer
4139 subscribe consumer { topic: "transactions" }
4240 let
4341 handler :: EachBatch
4442 handler { highWatermark, messages } resolveOffset heartBeat commitIfNecessary uncommittedOffsets isRunning isStale = do
45−
46− _ <- traverse (printAndResolve resolveOffset) messages :: Aff (Array Unit)
43+ void $ traverse (printAndResolve resolveOffset) messages
4744 heartBeat
48−
4945 eachBatch consumer false handler
50−
51−
52−