git.iko.soy/rowtype-yoga/purescript-kafkajs
on main help sign in
Cleanup types. Remove internal module. Add exports section.
0a3ff1chivemind-js
example/transactional-producer-consumer/src/Kafka/Example/Producer.purs+2−1
compare with main
unifiedsplit
@@ -14,6 +14,7 @@ import Effect.Class.Console (log)
14import Data.Time.Duration (Milliseconds(..))
14import Data.Time.Duration (Milliseconds(..))
15import Effect.Random (randomBool)
15import Effect.Random (randomBool)
16import Effect.Class (liftEffect)
16import Effect.Class (liftEffect)
17+import Kafka.Types (Topic(..))
17
18
18-- | Send 10 messages every 5 seconds and randomly abort or commit
19-- | Send 10 messages every 5 seconds and randomly abort or commit
19use :: Int -> Producer -> Aff Unit
20use :: Int -> Producer -> Aff Unit
@@ -22,7 +23,7 @@ use run producer = do
22 condition <- liftEffect $ randomBool
23 condition <- liftEffect $ randomBool
23 _ <-
24 _ <-
24 send trx
25 send trx
25− { topic: "transactions"
26+ { topic: Topic "transactions"
26 , messages:
27 , messages:
27 [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
28 [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
28 <#> \i ->
29 <#> \i ->
src/Kafka/Consumer.purs+29−13
@@ -1,25 +1,41 @@
1−module Kafka.Consumer where
1+module Kafka.Consumer
2−
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+
3−import Prelude (Unit, (>>>), (#), (<#>), bind, ($), (<>))
23+import Prelude (Unit, bind, (#), ($), (>>>))
4import Control.Promise (Promise, toAffE, fromAff)
24import Control.Promise (Promise, toAffE, fromAff)
5import Kafka.Kafka (Kafka)
25import Kafka.Kafka (Kafka)
6import Data.Function.Uncurried (Fn2, Fn3, Fn7, mkFn7, runFn2, runFn3)
26import Data.Function.Uncurried (Fn2, Fn3, Fn7, mkFn7, runFn2, runFn3)
7import Effect (Effect)
27import Effect (Effect)
8−import Effect.Class (liftEffect)
9import Effect.Aff (Aff)
28import Effect.Aff (Aff)
10−import Kafka.Internal.Internal (InternalOutputMessage)
11import Kafka.Types
29import Kafka.Types
12−import Data.Nullable (Nullable, toNullable, toMaybe)
30+import Data.Nullable (Nullable, toMaybe)
13import Node.Buffer (Buffer)
31import Node.Buffer (Buffer)
14−import Data.Maybe (Maybe(..), maybe)
32+import Data.Maybe (Maybe, maybe)
15import Data.Traversable (traverse)
33import Data.Traversable (traverse)
16−import Node.Buffer (toString)
17−import Node.Encoding (Encoding(..))
18−import Control.Monad.Error.Class (class MonadThrow, throwError)
34+import Control.Monad.Error.Class (throwError)
19−import Effect.Exception (error, Error)
35+import Effect.Exception (error)
20import Data.Int (fromString)
36import Data.Int (fromString)
21import Control.Applicative (pure)
37import Control.Applicative (pure)
22−import Data.Either (Either(..), note, either)
38+import Data.Either (Either, either, note)
23import Data.Newtype (un)
39import Data.Newtype (un)
24
40
25foreign import data Consumer :: Type
41foreign import data Consumer :: Type
@@ -265,4 +281,4 @@ eachBatch consumer eachBatchAutoResolve handler =
265foreign import disconnectImpl :: Consumer -> Effect (Promise Unit)
281foreign import disconnectImpl :: Consumer -> Effect (Promise Unit)
266
282
267disconnect :: Consumer -> Aff Unit
283disconnect :: 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
2
16
3−import Prelude (Unit, (#), (>>>))
17+import Prelude (Unit, (#), (>>>), (<#>))
4import Control.Promise (Promise, toAffE)
18import Control.Promise (Promise, toAffE)
5import Data.Function.Uncurried (Fn2, runFn2)
19import Data.Function.Uncurried (Fn2, runFn2)
6import Data.Maybe (Maybe)
20import Data.Maybe (Maybe)
7import Data.Nullable (Nullable, toNullable)
21import Data.Nullable (Nullable, toNullable)
8import Effect (Effect)
22import Effect (Effect)
9import Effect.Aff (Aff)
23import Effect.Aff (Aff)
10−import Kafka.Kafka (Kafka, Payload, RecordMetadata)
24+import Kafka.Kafka (Kafka)
11−import Kafka.Internal.Internal (InternalPayload, convert)
25+import Kafka.Types (Partition(..), Topic(..))
26+import Data.Newtype (un)
12
27
13foreign import data Producer :: Type
28foreign import data Producer :: Type
14
29
@@ -32,7 +47,7 @@ makeProducer k pc = runFn2 makeProducerImpl k ipc
32 ipc =
47 ipc =
33 { idempotent: toNullable pc.idempotent
48 { idempotent: toNullable pc.idempotent
34 , transactionalId: toNullable pc.transactionalId
49 , transactionalId: toNullable pc.transactionalId
35− , maxInFlightRequests : toNullable pc.maxInFlightRequests
50+ , maxInFlightRequests: toNullable pc.maxInFlightRequests
36 }
51 }
37
52
38foreign import connectImpl :: Producer -> Effect (Promise Unit)
53foreign import connectImpl :: Producer -> Effect (Promise Unit)
@@ -40,10 +55,40 @@ foreign import connectImpl :: Producer -> Effect (Promise Unit)
40connect :: Producer -> Aff Unit
55connect :: Producer -> Aff Unit
41connect = connectImpl >>> toAffE
56connect = connectImpl >>> toAffE
42
57
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+
43−foreign import sendImpl :: Fn2 Producer InternalPayload (Effect (Promise (Array RecordMetadata)))
88+foreign import sendImpl :: Fn2 Producer InternalProducerBatch (Effect (Promise (Array RecordMetadata)))
44
89
45−send :: Producer -> Payload -> Aff (Array RecordMetadata)
90+send :: Producer -> ProducerBatch -> Aff (Array RecordMetadata)
46−send p pl = runFn2 sendImpl p (convert pl) # toAffE
91+send p pl = runFn2 sendImpl p (toInternalProducerBatch pl) # toAffE
47
92
48foreign import disconnectImpl :: Producer -> Effect (Promise Unit)
93foreign import disconnectImpl :: Producer -> Effect (Promise Unit)
49
94
src/Kafka/Transaction.purs+11−8
@@ -1,14 +1,17 @@
1−module Kafka.Transaction where
1+module Kafka.Transaction
2−
2+ ( Transaction
3+ , transaction
4+ , commit
5+ , abort
6+ , send
7+ ) where
3
8
4import Prelude (Unit, (#), (>>>))
9import Prelude (Unit, (#), (>>>))
5import Control.Promise (Promise, toAffE)
10import Control.Promise (Promise, toAffE)
6import Data.Function.Uncurried (Fn2, runFn2)
11import Data.Function.Uncurried (Fn2, runFn2)
7import Effect (Effect)
12import Effect (Effect)
8import Effect.Aff (Aff)
13import Effect.Aff (Aff)
9−import Kafka.Producer (Producer)
14+import Kafka.Producer (InternalProducerBatch, Producer, ProducerBatch, RecordMetadata, toInternalProducerBatch)
10−import Kafka.Internal.Internal (InternalPayload, convert)
11−import Kafka.Kafka (Payload, RecordMetadata)
12
15
13foreign import data Transaction :: Type
16foreign import data Transaction :: Type
14
17
@@ -27,7 +30,7 @@ foreign import abortImpl :: Transaction -> Effect (Promise Unit)
27abort :: Transaction -> Aff Unit
30abort :: Transaction -> Aff Unit
28abort = abortImpl >>> toAffE
31abort = abortImpl >>> toAffE
29
32
30−foreign import sendImpl :: Fn2 Transaction InternalPayload (Effect (Promise (Array RecordMetadata)))
33+foreign import sendImpl :: Fn2 Transaction InternalProducerBatch (Effect (Promise (Array RecordMetadata)))
31
34
32−send :: Transaction -> Payload -> Aff (Array RecordMetadata)
35+send :: Transaction -> ProducerBatch -> Aff (Array RecordMetadata)
33−send t pl = runFn2 sendImpl t (convert pl) # toAffE
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
2
6
3import Data.Newtype (class Newtype)
7import Data.Newtype (class Newtype)
4import Data.Show (class Show)
8import Data.Show (class Show)
5
9
6−newtype Offset = Offset Int
10+newtype Offset
11+ = Offset Int
12+
7−instance ntOffset :: Newtype Offset Int
13+instance ntOffset :: Newtype Offset Int
14+
8−derive newtype instance showOffset :: Show Offset
15+derive newtype instance showOffset :: Show Offset
16+
17+newtype Partition
18+ = Partition Int
9
19
10−newtype Partition = Partition Int
11instance ntPartiton :: Newtype Partition Int
20instance ntPartiton :: Newtype Partition Int
12−derive newtype instance showPartition :: Show Partition
13
21
22+derive newtype instance showPartition :: Show Partition
23+
14−newtype Topic = Topic String
24+newtype Topic
25+ = Topic String
26+
15instance ntTopic :: Newtype Topic String
27instance ntTopic :: Newtype Topic String
28+
16−derive newtype instance showTopic :: Show Topic
29+derive newtype instance showTopic :: Show Topic