6868acb..main · rowtype-yoga/purescript-kafkajs

git.iko.soy/rowtype-yoga/purescript-kafkajs

7 files changed+78−12

unifiedsplit

packages.dhall+1−2 src/Kafka/Consumer.purs+11−2 src/Kafka/Kafka.js+7−1 src/Kafka/Kafka.purs+23 src/Kafka/Transaction.js+3−1 src/Kafka/Transaction.purs+32−5 src/Kafka/Types.purs+1−1

packages.dhall+1−2

@@ -1,5 +1,4 @@

1−

21let upstream =

3− https://github.com/purescript/package-sets/releases/download/psc-0.14.0-20210302/packages.dhall sha256:20cc5b89cf15433623ad6f250f112bf7a6bd82b5972363ecff4abf1febb02c50

2+ https://github.com/purescript/package-sets/releases/download/psc-0.14.0-20210313/packages.dhall sha256:ba6368b31902aad206851fec930e89465440ebf5a1fe0391f8be396e2d2f1d87

54in upstream

src/Kafka/Consumer.purs+11−2

@@ -1,5 +1,6 @@

11module Kafka.Consumer

22 ( Consumer

3+ , GroupId(..)

34 , ConsumerConfig

45 , SubscriptionConfig

56 , makeConsumer

@@ -26,6 +27,7 @@ import Kafka.Kafka (Kafka)

2627import Data.Function.Uncurried (Fn2, Fn3, Fn7, mkFn7, runFn2, runFn3)

2728import Effect (Effect)

2829import Effect.Aff (Aff)

30+import Data.Show (class Show)

2931import Kafka.Types

3032import Data.Nullable (Nullable, toMaybe)

3133import Node.Buffer (Buffer)

@@ -36,12 +38,19 @@ import Effect.Exception (error)

3638import Data.Int (fromString)

3739import Control.Applicative (pure)

3840import Data.Either (Either, either, note)

39−import Data.Newtype (un)

41+import Data.Newtype (un, class Newtype)

4042

4143foreign import data Consumer :: Type

4244

45+newtype GroupId

46+ = GroupId String

47+

48+instance ntGroupId :: Newtype GroupId String

49+

50+derive newtype instance showGroupId :: Show GroupId

51+

4352type ConsumerConfig

44− = { groupId :: String

53+ = { groupId :: GroupId

4554 , readUncommitted :: Boolean

4655 , autoCommit :: Boolean

4756 }

src/Kafka/Kafka.js+7−1

@@ -1,3 +1,9 @@

1−const { Kafka } = require('kafkajs')

1+const { Kafka, logLevel } = require('kafkajs')

33exports.makeClientImpl = params => new Kafka(params)

5+exports.internalLogNothing = logLevel.NOTHING

6+exports.internalLogDebug = logLevel.DEBUG

7+exports.internalLogInfo = logLevel.INFO

8+exports.internalLogWarn = logLevel.WARN

9+exports.internalLogError = logLevel.ERROR

src/Kafka/Kafka.purs+23

@@ -3,6 +3,7 @@ module Kafka.Kafka

33 , Kafka

44 , KafkaConfig

55 , makeClient

6+ , LogLevel(..)

67 ) where

89import Prelude

@@ -11,6 +12,18 @@ import Data.Nullable (Nullable, toNullable)

1112

1213foreign import data Kafka :: Type

1314

15+

16+data LogLevel = LogNothing | LogDebug | LogInfo | LogWarn | LogError

17+

18+foreign import data InternalLogLevel :: Type

19+

20+foreign import internalLogNothing :: InternalLogLevel

21+

22+foreign import internalLogDebug :: InternalLogLevel

23+foreign import internalLogInfo:: InternalLogLevel

24+foreign import internalLogWarn :: InternalLogLevel

25+foreign import internalLogError :: InternalLogLevel

26+

1427type SaslConfig

1528 = { mechanism :: String

1629 , username :: String

@@ -22,6 +35,7 @@ type InternalKafkaConfig

2235 , brokers :: Array String

2336 , ssl :: Boolean

2437 , sasl :: Nullable SaslConfig

38+ , logLevel :: Nullable InternalLogLevel

2539 }

2640

2741type KafkaConfig

@@ -29,6 +43,7 @@ type KafkaConfig

2943 , brokers :: Array String

3044 , ssl :: Boolean

3145 , sasl :: Maybe SaslConfig

46+ , logLevel :: Maybe LogLevel

3247 }

3348

3449foreign import makeClientImpl :: InternalKafkaConfig -> Kafka

@@ -36,9 +51,17 @@ foreign import makeClientImpl :: InternalKafkaConfig -> Kafka

3651makeClient :: KafkaConfig -> Kafka

3752makeClient = toInternal >>> makeClientImpl

3853 where

54+

55+ toInternalLogLevel LogNothing = internalLogNothing

56+ toInternalLogLevel LogDebug = internalLogDebug

57+ toInternalLogLevel LogInfo = internalLogDebug

58+ toInternalLogLevel LogWarn = internalLogWarn

59+ toInternalLogLevel LogError = internalLogError

60+

3961 toInternal config =

4062 { clientId: config.clientId

4163 , brokers: config.brokers

4264 , ssl: config.ssl

4365 , sasl: toNullable config.sasl

66+ , logLevel : config.logLevel <#> toInternalLogLevel # toNullable

4467 }

src/Kafka/Transaction.js+3−1

@@ -1,8 +1,10 @@

3+exports.transactionImpl = (producer) => () => producer.transaction()

35exports.sendImpl = (transaction, data) => () => transaction.send(data)

5−exports.transactionImpl = (producer) => () => producer.transaction()

7+exports.sendOffsetsImpl = (transaction, sendOffsets) => () => transaction.sendOffsets(sendOffsets)

79exports.commitImpl = (transaction) => () => transaction.commit()

810

src/Kafka/Transaction.purs+32−5

@@ -4,14 +4,20 @@ module Kafka.Transaction

44 , commit

55 , abort

66 , send

7+ , sendOffsets

8+ , SendOffsets

9+ , TopicCommit

10+ , PartitionCommit

711 ) where

812

9−import Prelude (Unit, (#), (>>>))

1013import Control.Promise (Promise, toAffE)

1114import Data.Function.Uncurried (Fn2, runFn2)

1215import Effect (Effect)

1316import Effect.Aff (Aff)

17+import Kafka.Consumer (GroupId)

1418import Kafka.Producer (InternalProducerBatch, Producer, ProducerBatch, RecordMetadata, toInternalProducerBatch)

19+import Kafka.Types (Offset, Partition, Topic)

20+import Prelude (Unit, (#), (>>>))

1521

1622foreign import data Transaction :: Type

1723

@@ -20,6 +26,31 @@ foreign import transactionImpl :: Producer -> Effect (Promise Transaction)

2026transaction :: Producer -> Aff Transaction

2127transaction = transactionImpl >>> toAffE

2228

29+type PartitionCommit = {

30+ partition:: Partition,

31+ offset :: Offset

32+ }

33+

34+type TopicCommit = {

35+ topic :: Topic,

36+ partitions :: Array PartitionCommit

37+}

38+

39+type SendOffsets = {

40+ consumerGroupId :: GroupId,

41+ topics :: Array TopicCommit

42+}

43+

44+foreign import sendOffsetsImpl :: Fn2 Transaction SendOffsets (Effect (Promise Unit))

45+

46+sendOffsets :: Transaction -> SendOffsets -> Aff Unit

47+sendOffsets t so = runFn2 sendOffsetsImpl t so # toAffE

48+

49+foreign import sendImpl :: Fn2 Transaction InternalProducerBatch (Effect (Promise (Array RecordMetadata)))

50+

51+send :: Transaction -> ProducerBatch -> Aff (Array RecordMetadata)

52+send t pl = runFn2 sendImpl t (toInternalProducerBatch pl) # toAffE

53+

2354foreign import commitImpl :: Transaction -> Effect (Promise Unit)

2455

2556commit :: Transaction -> Aff Unit

@@ -30,7 +61,3 @@ foreign import abortImpl :: Transaction -> Effect (Promise Unit)

3061abort :: Transaction -> Aff Unit

3162abort = abortImpl >>> toAffE

3263

33−foreign import sendImpl :: Fn2 Transaction InternalProducerBatch (Effect (Promise (Array RecordMetadata)))

34−

35−send :: Transaction -> ProducerBatch -> Aff (Array RecordMetadata)

36−send t pl = runFn2 sendImpl t (toInternalProducerBatch pl) # toAffE

src/Kafka/Types.purs+1−1

@@ -26,4 +26,4 @@ newtype Topic

2626

2727instance ntTopic :: Newtype Topic String

2828

29−derive newtype instance showTopic :: Show Topic

29+derive newtype instance showTopic :: Show Topic