0a3ff1c · rowtype-yoga/purescript-kafkajs

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

Cleanup types. Remove internal module. Add exports section.

commit0a3ff1c9ab251192e3424795b0bc11531ca175e1 authorhivemind-js <jan.schulte@hivemindtechnologies.com> dateMon 8 Mar 2021, 14:36 (6 years ago) parentd82159d

6 files changed+116−68

compare with main

unifiedsplit

example/transactional-producer-consumer/src/Kafka/Example/Producer.purs+2−1 src/Kafka/Consumer.purs+29−13 src/Kafka/Internal/Internal.purs−30 src/Kafka/Producer.purs+53−8 src/Kafka/Transaction.purs+11−8 src/Kafka/Types.purs+21−8

example/transactional-producer-consumer/src/Kafka/Example/Producer.purs+2−1

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

src/Kafka/Consumer.purs+29−13

@@ -1,25 +1,41 @@

1−module Kafka.Consumer where

2−

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

1+module Kafka.Consumer

2+ ( Consumer

3+ , ConsumerConfig

4+ , SubscriptionConfig

5+ , makeConsumer

6+ , connect

7+ , subscribe

8+ , ConsumerMessage

9+ , ConsumerBatch

10+ , Heartbeat

11+ , ResolveOffset

12+ , CommitOffsetsIfNecessary

13+ , OffsetInfo

14+ , toOffsetInfo

15+ , UncommittedOffsets

16+ , IsStale

17+ , IsRunning

18+ , EachBatch

19+ , eachBatch

20+ , disconnect

21+ ) where

22+

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

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

525import Kafka.Kafka (Kafka)

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

727import Effect (Effect)

8−import Effect.Class (liftEffect)

928import Effect.Aff (Aff)

10−import Kafka.Internal.Internal (InternalOutputMessage)

1129import Kafka.Types

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

30+import Data.Nullable (Nullable, toMaybe)

1331import Node.Buffer (Buffer)

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

32+import Data.Maybe (Maybe, maybe)

1533import 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)

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

35+import Effect.Exception (error)

2036import Data.Int (fromString)

2137import Control.Applicative (pure)

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

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

2339import Data.Newtype (un)

2440

2541foreign import data Consumer :: Type

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

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

266282

267283disconnect :: Consumer -> Aff Unit

268−disconnect = disconnectImpl >>> toAffE

284+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/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.purs+11−8

@@ -1,14 +1,17 @@

1−module Kafka.Transaction where

2−

1+module Kafka.Transaction

2+ ( Transaction

3+ , transaction

4+ , commit

5+ , abort

6+ , send

7+ ) where

49import Prelude (Unit, (#), (>>>))

510import Control.Promise (Promise, toAffE)

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

712import Effect (Effect)

813import Effect.Aff (Aff)

9−import Kafka.Producer (Producer)

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

11−import Kafka.Kafka (Payload, RecordMetadata)

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

1215

1316foreign import data Transaction :: Type

1417

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

2730abort :: Transaction -> Aff Unit

2831abort = abortImpl >>> toAffE

2932

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

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

3134

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

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

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

36+send t pl = runFn2 sendImpl t (toInternalProducerBatch 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