716bc32...main · rowtype-yoga/purescript-kafkajs

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

16 files changed+271−122

since splitsnapshots

unifiedsplit

.github/workflows/ci.yml+32 bower.json+27 example/README.md+2−2 example/transactional-producer-consumer/src/Kafka/Example/Consumer.purs+7−14 example/transactional-producer-consumer/src/Kafka/Example/Producer.purs+4−3 packages.dhall+1−2 spago.dhall+1−5 src/Kafka/Consumer.purs+40−15 src/Kafka/Internal/Internal.purs−30 src/Kafka/Kafka.js+7−1 src/Kafka/Kafka.purs+23−23 src/Kafka/Producer.purs+53−8 src/Kafka/Transaction.js+3−1 src/Kafka/Transaction.purs+40−10 src/Kafka/Types.purs+21−8 test/Test/Main.purs+10

.github/workflows/ci.ymlnew file+32

@@ -0,0 +1,32 @@

1+name: CI

3+on:

4+ push:

5+ branches: [main]

6+ pull_request:

7+ branches: [main]

9+jobs:

10+ build:

11+ runs-on: ubuntu-latest

12+ steps:

13+ - uses: actions/checkout@v2

14+

15+ - uses: purescript-contrib/setup-purescript@main

16+

17+ - name: Cache PureScript dependencies

18+ uses: actions/cache@v2

19+ # This cache uses the .dhall files to know when it should reinstall

20+ # and rebuild packages. It caches both the installed packages from

21+ # the `.spago` directory and compilation artifacts from the `output`

22+ # directory. When restored the compiler will rebuild any files that

23+ # have changed. If you do not want to cache compiled output, remove

24+ # the `output` path.

25+ with:

26+ key: ${{ runner.os }}-spago-${{ hashFiles('**/*.dhall') }}

27+ path: |

28+ .spago

29+ output

30+ - run: spago build

31+

32+ - run: spago test --no-install

bower.jsonnew file+27

@@ -0,0 +1,27 @@

1+{

2+ "name": "purescript-kafkajs",

3+ "license": [

4+ "MIT"

5+ ],

6+ "repository": {

7+ "type": "git",

8+ "url": "https://github.com/HivemindTechnologies/purescript-kafkajs.git"

9+ },

10+ "ignore": [

11+ "**/.*",

12+ "node_modules",

13+ "bower_components",

14+ "output"

15+ ],

16+ "dependencies": {

17+ "purescript-aff-promise": "^v2.1.0",

18+ "purescript-console": "^v5.0.0",

19+ "purescript-debug": "^v4.0.1",

20+ "purescript-effect": "^v3.0.0",

21+ "purescript-maybe": "^v5.0.0",

22+ "purescript-node-buffer": "^v7.0.0",

23+ "purescript-nullable": "^v5.0.0",

24+ "purescript-psci-support": "^v5.0.0",

25+ "purescript-spec": "^v5.0.0"

26+ }

27+}

example/README.md+2−2

@@ -9,8 +9,8 @@ docker-compose up -d

99## Transactional producer consumer

1010

1111Install everything

12−```bash

13−npm install

12+```bash

13+cd transactional-producer-consumer && npm install

1414```

1515

1616Run a producer in a terminal.

example/transactional-producer-consumer/src/Kafka/Example/Consumer.purs+7−14

@@ -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−

example/transactional-producer-consumer/src/Kafka/Example/Producer.purs+4−3

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

11module Kafka.Example.Producer where

3−import Prelude (Unit, (>>>), (#), (<#>), bind, ($), (<>), pure, unit, (*>), (+))

3+import Prelude (Unit, (>>>), (#), (<#>), bind, ($), (<>), pure, unit, (*>), (+), discard)

44import Control.Promise (Promise, toAffE, fromAff)

55import Effect (Effect)

66import Kafka.Producer (connect, disconnect, Producer, makeProducer, ProducerConfig)

@@ -14,6 +14,7 @@ import Effect.Class.Console (log)

1414import Data.Time.Duration (Milliseconds(..))

1515import Effect.Random (randomBool)

1616import Effect.Class (liftEffect)

17+import Kafka.Types (Topic(..))

1718

1819-- | Send 10 messages every 5 seconds and randomly abort or commit

1920use :: Int -> Producer -> Aff Unit

@@ -22,7 +23,7 @@ use run producer = do

2223 condition <- liftEffect $ randomBool

2324 _ <-

2425 send trx

25− { topic: "transactions"

26+ { topic: Topic "transactions"

2627 , messages:

2728 [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]

2829 <#> \i ->

@@ -52,5 +53,5 @@ main =

5253

5354 release :: Producer -> Aff Unit

5455 release = disconnect

55− _ <- bracket (acquire kafka producerConfig) release (use 0)

56+ bracket (acquire kafka producerConfig) release (use 0)

5657 pure unit

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

spago.dhall+1−5

@@ -1,8 +1,4 @@

1−{-

2−Welcome to a Spago project!

3−You can edit this file as you like.

4−-}

5−{ name = "purescript-kafkajs"

1+{ name = "kafkajs"

62, dependencies =

73 [ "aff-promise"

84 , "console"

src/Kafka/Consumer.purs+40−15

@@ -1,31 +1,56 @@

1−module Kafka.Consumer where

2−

3−import Prelude (Unit, (>>>), (#), (<#>), bind, ($), (<>))

1+module 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

23+

24+import Prelude (Unit, bind, (#), ($), (>>>))

425import Control.Promise (Promise, toAffE, fromAff)

526import Kafka.Kafka (Kafka)

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

728import Effect (Effect)

8−import Effect.Class (liftEffect)

929import Effect.Aff (Aff)

10−import Kafka.Internal.Internal (InternalOutputMessage)

30+import Data.Show (class Show)

1131import Kafka.Types

12−import Data.Nullable (Nullable, toNullable, toMaybe)

32+import Data.Nullable (Nullable, toMaybe)

1333import Node.Buffer (Buffer)

14−import Data.Maybe (Maybe(..), maybe)

34+import Data.Maybe (Maybe, maybe)

1535import Data.Traversable (traverse)

16−import Node.Buffer (toString)

17−import Node.Encoding (Encoding(..))

18−import Control.Monad.Error.Class (class MonadThrow, throwError)

19−import Effect.Exception (error, Error)

36+import Control.Monad.Error.Class (throwError)

37+import Effect.Exception (error)

2038import Data.Int (fromString)

2139import Control.Applicative (pure)

22−import Data.Either (Either(..), note, either)

23−import Data.Newtype (un)

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

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

2442

2543foreign import data Consumer :: Type

2644

45+newtype GroupId

46+ = GroupId String

47+

48+instance ntGroupId :: Newtype GroupId String

49+

50+derive newtype instance showGroupId :: Show GroupId

51+

2752type ConsumerConfig

28− = { groupId :: String

53+ = { groupId :: GroupId

2954 , readUncommitted :: Boolean

3055 , autoCommit :: Boolean

3156 }

@@ -265,4 +290,4 @@ eachBatch consumer eachBatchAutoResolve handler =

265290foreign import disconnectImpl :: Consumer -> Effect (Promise Unit)

266291

267292disconnect :: Consumer -> Aff Unit

268−disconnect = disconnectImpl >>> toAffE

293+disconnect = disconnectImpl >>> toAffE

src/Kafka/Internal/Internal.pursdeleted−30

@@ -1,30 +0,0 @@

1−module Kafka.Internal.Internal where

2−

3−import Prelude

4−import Kafka.Kafka (Payload, Message)

5−import Data.Nullable (Nullable, toNullable)

6−import Node.Buffer (Buffer)

7−

8−type InternalMessage

9− = { key :: Nullable String

10− , value :: String

11− , partition :: Nullable Int

12− }

13−

14−type InternalOutputMessage

15− = { key :: Nullable String

16− , value :: Buffer

17− , partition :: Int

18− , offset :: Int

19− }

20−

21−type InternalPayload

22− = { topic :: String

23− , messages :: Array (InternalMessage)

24− }

25−

26−convertMessage :: Message -> InternalMessage

27−convertMessage { key, value, partition } = { key: toNullable key, value: value, partition: toNullable partition }

28−

29−convert :: Payload -> InternalPayload

30−convert { topic, messages } = { topic, messages: messages <#> convertMessage }

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−23

@@ -1,11 +1,9 @@

11module Kafka.Kafka

22 ( SaslConfig

33 , Kafka

4− , RecordMetadata

54 , KafkaConfig

65 , makeClient

7− , Message

8− , Payload

6+ , LogLevel(..)

97 ) where

108

119import Prelude

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

1412

1513foreign import data Kafka :: Type

1614

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+

1727type SaslConfig

1828 = { mechanism :: String

1929 , username :: String

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

2535 , brokers :: Array String

2636 , ssl :: Boolean

2737 , sasl :: Nullable SaslConfig

38+ , logLevel :: Nullable InternalLogLevel

2839 }

2940

3041type KafkaConfig

@@ -32,36 +43,25 @@ type KafkaConfig

3243 , brokers :: Array String

3344 , ssl :: Boolean

3445 , sasl :: Maybe SaslConfig

46+ , logLevel :: Maybe LogLevel

3547 }

3648

37−

38−

3949foreign import makeClientImpl :: InternalKafkaConfig -> Kafka

4050

4151makeClient :: KafkaConfig -> Kafka

4252makeClient = toInternal >>> makeClientImpl

4353 where

54+

55+ toInternalLogLevel LogNothing = internalLogNothing

56+ toInternalLogLevel LogDebug = internalLogDebug

57+ toInternalLogLevel LogInfo = internalLogDebug

58+ toInternalLogLevel LogWarn = internalLogWarn

59+ toInternalLogLevel LogError = internalLogError

60+

4461 toInternal config =

4562 { clientId: config.clientId

4663 , brokers: config.brokers

4764 , ssl: config.ssl

4865 , sasl: toNullable config.sasl

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

4967 }

50−

51−

52−type Message

53− = { key :: Maybe String

54− , value :: String

55− , partition :: Maybe Int

56− }

57−

58−type Payload

59− = { topic :: String

60− , messages :: Array (Message)

61− }

62−

63−

64−foreign import data RecordMetadata :: Type

65−

66−

67−

src/Kafka/Producer.purs+53−8

@@ -1,14 +1,29 @@

1−module Kafka.Producer where

1+module Kafka.Producer

2+ ( Producer

3+ , ProducerConfig

4+ , makeProducer

5+ , connect

6+ , ProducerMessage

7+ , ProducerBatch

8+ , InternalProducerMessage

9+ , InternalProducerBatch

10+ , toInternalProducerMessage

11+ , toInternalProducerBatch

12+ , RecordMetadata

13+ , send

14+ , disconnect

15+ ) where

216

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

17+import Prelude (Unit, (#), (>>>), (<#>))

418import Control.Promise (Promise, toAffE)

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

620import Data.Maybe (Maybe)

721import Data.Nullable (Nullable, toNullable)

822import Effect (Effect)

923import Effect.Aff (Aff)

10−import Kafka.Kafka (Kafka, Payload, RecordMetadata)

11−import Kafka.Internal.Internal (InternalPayload, convert)

24+import Kafka.Kafka (Kafka)

25+import Kafka.Types (Partition(..), Topic(..))

26+import Data.Newtype (un)

1227

1328foreign import data Producer :: Type

1429

@@ -32,7 +47,7 @@ makeProducer k pc = runFn2 makeProducerImpl k ipc

3247 ipc =

3348 { idempotent: toNullable pc.idempotent

3449 , transactionalId: toNullable pc.transactionalId

35− , maxInFlightRequests : toNullable pc.maxInFlightRequests

50+ , maxInFlightRequests: toNullable pc.maxInFlightRequests

3651 }

3752

3853foreign import connectImpl :: Producer -> Effect (Promise Unit)

@@ -40,10 +55,40 @@ foreign import connectImpl :: Producer -> Effect (Promise Unit)

4055connect :: Producer -> Aff Unit

4156connect = connectImpl >>> toAffE

4257

43−foreign import sendImpl :: Fn2 Producer InternalPayload (Effect (Promise (Array RecordMetadata)))

58+type ProducerMessage

59+ = { key :: Maybe String

60+ , value :: String

61+ , partition :: Maybe Partition

62+ }

63+

64+type ProducerBatch

65+ = { topic :: Topic

66+ , messages :: Array (ProducerMessage)

67+ }

68+

69+foreign import data RecordMetadata :: Type

70+

71+type InternalProducerMessage

72+ = { key :: Nullable String

73+ , value :: String

74+ , partition :: Nullable Int

75+ }

76+

77+type InternalProducerBatch

78+ = { topic :: String

79+ , messages :: Array (InternalProducerMessage)

80+ }

81+

82+toInternalProducerMessage :: ProducerMessage -> InternalProducerMessage

83+toInternalProducerMessage { key, value, partition: partition } = { key: toNullable key, value: value, partition: partition <#> un Partition # toNullable }

84+

85+toInternalProducerBatch :: ProducerBatch -> InternalProducerBatch

86+toInternalProducerBatch { topic: Topic topic, messages } = { topic, messages: messages <#> toInternalProducerMessage }

87+

88+foreign import sendImpl :: Fn2 Producer InternalProducerBatch (Effect (Promise (Array RecordMetadata)))

4489

45−send :: Producer -> Payload -> Aff (Array RecordMetadata)

46−send p pl = runFn2 sendImpl p (convert pl) # toAffE

90+send :: Producer -> ProducerBatch -> Aff (Array RecordMetadata)

91+send p pl = runFn2 sendImpl p (toInternalProducerBatch pl) # toAffE

4792

4893foreign import disconnectImpl :: Producer -> Effect (Promise Unit)

4994

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+40−10

@@ -1,14 +1,23 @@

1−module Kafka.Transaction where

1+module Kafka.Transaction

2+ ( Transaction

3+ , transaction

4+ , commit

5+ , abort

6+ , send

7+ , sendOffsets

8+ , SendOffsets

9+ , TopicCommit

10+ , PartitionCommit

11+ ) where

212

3−

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

513import Control.Promise (Promise, toAffE)

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

715import Effect (Effect)

816import Effect.Aff (Aff)

9−import Kafka.Producer (Producer)

10−import Kafka.Internal.Internal (InternalPayload, convert)

11−import Kafka.Kafka (Payload, RecordMetadata)

17+import Kafka.Consumer (GroupId)

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

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

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

1221

1322foreign import data Transaction :: Type

1423

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

1726transaction :: Producer -> Aff Transaction

1827transaction = transactionImpl >>> toAffE

1928

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+

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

2155

2256commit :: Transaction -> Aff Unit

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

2761abort :: Transaction -> Aff Unit

2862abort = abortImpl >>> toAffE

2963

30−foreign import sendImpl :: Fn2 Transaction InternalPayload (Effect (Promise (Array RecordMetadata)))

31−

32−send :: Transaction -> Payload -> Aff (Array RecordMetadata)

33−send t pl = runFn2 sendImpl t (convert pl) # toAffE

src/Kafka/Types.purs+21−8

@@ -1,16 +1,29 @@

1−module Kafka.Types where

1+module Kafka.Types

2+ ( Offset(..)

3+ , Partition(..)

4+ , Topic(..)

5+ ) where

37import Data.Newtype (class Newtype)

48import Data.Show (class Show)

6−newtype Offset = Offset Int

7−instance ntOffset :: Newtype Offset Int

8−derive newtype instance showOffset :: Show Offset

10+newtype Offset

11+ = Offset Int

12+

13+instance ntOffset :: Newtype Offset Int

14+

15+derive newtype instance showOffset :: Show Offset

16+

17+newtype Partition

18+ = Partition Int

919

10−newtype Partition = Partition Int

1120instance ntPartiton :: Newtype Partition Int

12−derive newtype instance showPartition :: Show Partition

1321

14−newtype Topic = Topic String

22+derive newtype instance showPartition :: Show Partition

23+

24+newtype Topic

25+ = Topic String

26+

1527instance ntTopic :: Newtype Topic String

16−derive newtype instance showTopic :: Show Topic

28+

29+derive newtype instance showTopic :: Show Topic

test/Test/Main.pursnew file+10

@@ -0,0 +1,10 @@

1+module Test.Main where

3+import Data.Unit (Unit)

4+import Effect (Effect)

5+import Effect.Class.Console (log)

7+main :: Effect Unit

8+main =

9+ do

10+ log "Add some tests"