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

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

on main help sign in

packages.dhall+1−2

since splitsnapshots

unifiedsplit

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

1−

2let upstream =

1let 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

5in upstream

4in upstream

src/Kafka/Consumer.purs+11−2

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

1module Kafka.Consumer

2 ( Consumer

3+ , GroupId(..)

3 , ConsumerConfig

4 , ConsumerConfig

4 , SubscriptionConfig

5 , SubscriptionConfig

5 , makeConsumer

6 , makeConsumer

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

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

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

27import Effect (Effect)

28import Effect (Effect)

28import Effect.Aff (Aff)

29import Effect.Aff (Aff)

30+import Data.Show (class Show)

29import Kafka.Types

31import Kafka.Types

30import Data.Nullable (Nullable, toMaybe)

32import Data.Nullable (Nullable, toMaybe)

31import Node.Buffer (Buffer)

33import Node.Buffer (Buffer)

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

36import Data.Int (fromString)

38import Data.Int (fromString)

37import Control.Applicative (pure)

39import Control.Applicative (pure)

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

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

39−import Data.Newtype (un)

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

41foreign import data Consumer :: Type

43foreign import data Consumer :: Type

45+newtype GroupId

46+ = GroupId String

47+

48+instance ntGroupId :: Newtype GroupId String

49+

50+derive newtype instance showGroupId :: Show GroupId

51+

43type ConsumerConfig

52type ConsumerConfig

44− = { groupId :: String

53+ = { groupId :: GroupId

45 , readUncommitted :: Boolean

54 , readUncommitted :: Boolean

46 , autoCommit :: Boolean

55 , autoCommit :: Boolean

47 }

56 }

src/Kafka/Kafka.js+7−1

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

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

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

3exports.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

3 , Kafka

4 , KafkaConfig

5 , makeClient

6+ , LogLevel(..)

6 ) where

7 ) where

8import Prelude

9import Prelude

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

12foreign import data Kafka :: Type

13foreign import data Kafka :: Type

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+

14type SaslConfig

27type SaslConfig

15 = { mechanism :: String

28 = { mechanism :: String

16 , username :: String

29 , username :: String

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

22 , brokers :: Array String

35 , brokers :: Array String

23 , ssl :: Boolean

36 , ssl :: Boolean

24 , sasl :: Nullable SaslConfig

37 , sasl :: Nullable SaslConfig

38+ , logLevel :: Nullable InternalLogLevel

25 }

39 }

27type KafkaConfig

41type KafkaConfig

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

29 , brokers :: Array String

43 , brokers :: Array String

30 , ssl :: Boolean

44 , ssl :: Boolean

31 , sasl :: Maybe SaslConfig

45 , sasl :: Maybe SaslConfig

46+ , logLevel :: Maybe LogLevel

32 }

47 }

34foreign import makeClientImpl :: InternalKafkaConfig -> Kafka

49foreign import makeClientImpl :: InternalKafkaConfig -> Kafka

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

36makeClient :: KafkaConfig -> Kafka

51makeClient :: KafkaConfig -> Kafka

37makeClient = toInternal >>> makeClientImpl

52makeClient = toInternal >>> makeClientImpl

38 where

53 where

54+

55+ toInternalLogLevel LogNothing = internalLogNothing

56+ toInternalLogLevel LogDebug = internalLogDebug

57+ toInternalLogLevel LogInfo = internalLogDebug

58+ toInternalLogLevel LogWarn = internalLogWarn

59+ toInternalLogLevel LogError = internalLogError

60+

39 toInternal config =

61 toInternal config =

40 { clientId: config.clientId

62 { clientId: config.clientId

41 , brokers: config.brokers

63 , brokers: config.brokers

42 , ssl: config.ssl

64 , ssl: config.ssl

43 , sasl: toNullable config.sasl

65 , sasl: toNullable config.sasl

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

44 }

67 }

src/Kafka/Transaction.js+3−1

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

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

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

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

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

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

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

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

src/Kafka/Transaction.purs+32−5

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

4 , commit

5 , abort

6 , send

7+ , sendOffsets

8+ , SendOffsets

9+ , TopicCommit

10+ , PartitionCommit

7 ) where

11 ) where

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

10import Control.Promise (Promise, toAffE)

13import Control.Promise (Promise, toAffE)

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

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

12import Effect (Effect)

15import Effect (Effect)

13import Effect.Aff (Aff)

16import Effect.Aff (Aff)

17+import Kafka.Consumer (GroupId)

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

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

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

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

16foreign import data Transaction :: Type

22foreign import data Transaction :: Type

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

20transaction :: Producer -> Aff Transaction

26transaction :: Producer -> Aff Transaction

21transaction = transactionImpl >>> toAffE

27transaction = transactionImpl >>> toAffE

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+

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

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

25commit :: Transaction -> Aff Unit

56commit :: Transaction -> Aff Unit

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

30abort :: Transaction -> Aff Unit

61abort :: Transaction -> Aff Unit

31abort = abortImpl >>> toAffE

62abort = abortImpl >>> toAffE

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

27instance ntTopic :: Newtype Topic String

29−derive newtype instance showTopic :: Show Topic

29+derive newtype instance showTopic :: Show Topic