Skip to content

Commit a2793e4

Browse files
author
Philipp Hausmann
committed
Wrap callbacks in newtype to make wrong usage harder
1 parent 3a3291c commit a2793e4

File tree

7 files changed

+36
-27
lines changed

7 files changed

+36
-27
lines changed

‎src/Kafka/Callbacks.hs

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,13 @@ module Kafka.Callbacks
22
( errorCallback
33
, logCallback
44
, statsCallback
5+
, Callback
56
)
67
where
78

89
import Data.ByteString (ByteString)
910
import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb)
10-
import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf)
11+
import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf, Callback(..))
1112
import Kafka.Types (KafkaError(..), KafkaLogLevel(..))
1213

1314
-- | Add a callback for errors.
@@ -20,10 +21,10 @@ import Kafka.Types (KafkaError(..), KafkaLogLevel(..))
2021
-- >
2122
-- > myErrorCallback :: 'KafkaError' -> String -> IO ()
2223
-- > myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message
23-
errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO ()
24-
errorCallback callback k =
24+
errorCallback :: (KafkaError -> String -> IO ()) -> Callback
25+
errorCallback callback =
2526
let realCb _ err = callback (KafkaResponseError err)
26-
in rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb
27+
in Callback $ \k -> rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb
2728

2829
-- | Add a callback for logs.
2930
--
@@ -35,10 +36,10 @@ errorCallback callback k =
3536
-- >
3637
-- > myLogCallback :: 'KafkaLogLevel' -> String -> String -> IO ()
3738
-- > myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message
38-
logCallback :: HasKafkaConf k => (KafkaLogLevel -> String -> String -> IO ()) -> k -> IO ()
39-
logCallback callback k =
39+
logCallback :: (KafkaLogLevel -> String -> String -> IO ()) -> Callback
40+
logCallback callback =
4041
let realCb _ = callback . toEnum
41-
in rdKafkaConfSetLogCb (getRdKafkaConf k) realCb
42+
in Callback $ \k -> rdKafkaConfSetLogCb (getRdKafkaConf k) realCb
4243

4344
-- | Add a callback for stats. The passed ByteString contains an UTF-8 encoded JSON document and can e.g. be parsed using Data.Aeson.decodeStrict. For more information about the content of the JSON document see <https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md>.
4445
--
@@ -50,7 +51,7 @@ logCallback callback k =
5051
-- >
5152
-- > myStatsCallback :: String -> IO ()
5253
-- > myStatsCallback stats = print $ show stats
53-
statsCallback :: HasKafkaConf k => (ByteString -> IO ()) -> k -> IO ()
54-
statsCallback callback k =
54+
statsCallback :: (ByteString -> IO ()) -> Callback
55+
statsCallback callback =
5556
let realCb _ = callback
56-
in rdKafkaConfSetStatsCb (getRdKafkaConf k) realCb
57+
in Callback $ \k -> rdKafkaConfSetStatsCb (getRdKafkaConf k) realCb

‎src/Kafka/Consumer.hs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ import Foreign hiding (void)
8585
import Kafka.Consumer.Convert (fromMessagePtr, fromNativeTopicPartitionList'', offsetCommitToBool, offsetToInt64, toMap, toNativeTopicPartitionList, toNativeTopicPartitionList', toNativeTopicPartitionListNoDispose, topicPartitionFromMessageForCommit)
8686
import Kafka.Consumer.Types (KafkaConsumer (..))
8787
import Kafka.Internal.RdKafka (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), newRdKafkaT, newRdKafkaTopicPartitionListT, newRdKafkaTopicT, rdKafkaAssign, rdKafkaAssignment, rdKafkaCommit, rdKafkaCommitted, rdKafkaConfSetDefaultTopicConf, rdKafkaConsumeBatchQueue, rdKafkaConsumeQueue, rdKafkaConsumerClose, rdKafkaConsumerPoll, rdKafkaOffsetsStore, rdKafkaPausePartitions, rdKafkaPollSetConsumer, rdKafkaPosition, rdKafkaQueueDestroy, rdKafkaQueueNew, rdKafkaResumePartitions, rdKafkaSeek, rdKafkaSetLogLevel, rdKafkaSubscribe, rdKafkaSubscription, rdKafkaTopicConfDup, rdKafkaTopicPartitionListAdd)
88-
import Kafka.Internal.Setup (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf)
88+
import Kafka.Internal.Setup (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf, Callback(..))
8989
import Kafka.Internal.Shared (kafkaErrorToMaybe, maybeToLeft, rdKafkaErrorToEither)
9090

9191
import Kafka.Consumer.ConsumerProperties as X
@@ -327,7 +327,7 @@ closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr statusVar)) = liftIO $
327327
newConsumerConf :: ConsumerProperties -> IO KafkaConf
328328
newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do
329329
conf <- kafkaConf (KafkaProps m)
330-
forM_ cbs (\setCb -> setCb conf)
330+
forM_ cbs (\(Callback setCb) -> setCb conf)
331331
return conf
332332

333333
-- | Subscribes to a given list of topics.

‎src/Kafka/Consumer/Callbacks.hs

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -14,16 +14,17 @@ import Kafka.Callbacks as X
1414
import Kafka.Consumer.Convert (fromNativeTopicPartitionList', fromNativeTopicPartitionList'')
1515
import Kafka.Consumer.Types (KafkaConsumer (..), RebalanceEvent (..), TopicPartition (..))
1616
import Kafka.Internal.RdKafka
17-
import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue)
17+
import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue, Callback (..))
1818
import Kafka.Types (KafkaError (..), PartitionId (..), TopicName (..))
1919

2020
import qualified Data.Text as Text
2121

2222
-- | Sets a callback that is called when rebalance is needed.
23-
rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> KafkaConf -> IO ()
24-
rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb conf realCb
23+
rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> Callback
24+
rebalanceCallback callback =
25+
Callback $ \kc@(KafkaConf con _ _) -> rdKafkaConfSetRebalanceCb con (realCb kc)
2526
where
26-
realCb k err pl = do
27+
realCb kc k err pl = do
2728
k' <- newForeignPtr_ k
2829
pls <- newForeignPtr_ pl
2930
setRebalanceCallback callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls
@@ -36,10 +37,11 @@ rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb c
3637
-- If no partitions had valid offsets to commit this callback will be called
3738
-- with 'KafkaResponseError' 'RdKafkaRespErrNoOffset' which is not to be considered
3839
-- an error.
39-
offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO ()
40-
offsetCommitCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetOffsetCommitCb conf realCb
40+
offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> Callback
41+
offsetCommitCallback callback =
42+
Callback $ \kc@(KafkaConf conf _ _) -> rdKafkaConfSetOffsetCommitCb conf (realCb kc)
4143
where
42-
realCb k err pl = do
44+
realCb kc k err pl = do
4345
k' <- newForeignPtr_ k
4446
pls <- fromNativeTopicPartitionList' pl
4547
callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls

‎src/Kafka/Consumer/ConsumerProperties.hs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ import Data.Semigroup as Sem
3434
import Data.Text (Text)
3535
import qualified Data.Text as Text
3636
import Kafka.Consumer.Types (ConsumerGroupId (..))
37-
import Kafka.Internal.Setup (KafkaConf (..))
37+
import Kafka.Internal.Setup (KafkaConf (..), Callback(..))
3838
import Kafka.Types (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), Millis (..), kafkaCompressionCodecToText, kafkaDebugToText)
3939

4040
import Kafka.Consumer.Callbacks as X
@@ -53,7 +53,7 @@ data CallbackPollMode =
5353
data ConsumerProperties = ConsumerProperties
5454
{ cpProps :: Map Text Text
5555
, cpLogLevel :: Maybe KafkaLogLevel
56-
, cpCallbacks :: [KafkaConf -> IO ()]
56+
, cpCallbacks :: [Callback]
5757
, cpCallbackPollMode :: CallbackPollMode
5858
}
5959

@@ -117,7 +117,7 @@ clientId (ClientId cid) =
117117
-- * 'errorCallback'
118118
-- * 'logCallback'
119119
-- * 'statsCallback'
120-
setCallback :: (KafkaConf -> IO ()) -> ConsumerProperties
120+
setCallback :: Callback -> ConsumerProperties
121121
setCallback cb = mempty { cpCallbacks = [cb] }
122122

123123
-- | Set the logging level.

‎src/Kafka/Internal/Setup.hs

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ module Kafka.Internal.Setup
77
, HasKafka(..)
88
, HasKafkaConf(..)
99
, HasTopicConf(..)
10+
, Callback(..)
1011
, CallbackPollStatus(..)
1112
, getRdKafka
1213
, getRdKafkaConf
@@ -46,6 +47,11 @@ newtype TopicProps = TopicProps (Map Text Text) deriving (Show, Eq)
4647
newtype Kafka = Kafka RdKafkaTPtr deriving Show
4748
newtype TopicConf = TopicConf RdKafkaTopicConfTPtr deriving Show
4849

50+
-- | Callbacks allow retrieving various information like error occurences, statistics
51+
-- and log messages.
52+
-- See `Kafka.Consumer.setCallback` (Consumer) and `Kafka.Producer.setCallback` (Producer) for more details.
53+
newtype Callback = Callback (KafkaConf -> IO ())
54+
4955
data CallbackPollStatus = CallbackPollEnabled | CallbackPollDisabled deriving (Show, Eq)
5056

5157
data KafkaConf = KafkaConf

‎src/Kafka/Producer.hs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ import Foreign.Ptr (Ptr, nullPtr, plusPtr)
8282
import Foreign.Storable (Storable (..))
8383
import Foreign.StablePtr (newStablePtr, castStablePtrToPtr)
8484
import Kafka.Internal.RdKafka (RdKafkaMessageT (..), RdKafkaRespErrT (..), RdKafkaTypeT (..), destroyUnmanagedRdKafkaTopic, newRdKafkaT, newUnmanagedRdKafkaTopicT, rdKafkaOutqLen, rdKafkaProduce, rdKafkaProduceBatch, rdKafkaSetLogLevel)
85-
import Kafka.Internal.Setup (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf)
85+
import Kafka.Internal.Setup (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf, Callback(..))
8686
import Kafka.Internal.Shared (pollEvents)
8787
import Kafka.Producer.Convert (copyMsgFlags, handleProduceErr', producePartitionCInt, producePartitionInt)
8888
import Kafka.Producer.Types (KafkaProducer (..), ImmediateError(..))
@@ -120,7 +120,7 @@ newProducer pps = liftIO $ do
120120
deliveryCallback (const mempty) kc
121121

122122
-- set callbacks
123-
forM_ (ppCallbacks pps) (\setCb -> setCb kc)
123+
forM_ (ppCallbacks pps) (\(Callback setCb) -> setCb kc)
124124

125125
mbKafka <- newRdKafkaT RdKafkaProducer kc'
126126
case mbKafka of

‎src/Kafka/Producer/ProducerProperties.hs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import Control.Monad (MonadPlus(mplus))
2727
import Data.Map (Map)
2828
import qualified Data.Map as M
2929
import Data.Semigroup as Sem
30-
import Kafka.Internal.Setup (KafkaConf(..))
30+
import Kafka.Internal.Setup (KafkaConf(..), Callback(..))
3131
import Kafka.Types (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText, Millis(..))
3232

3333
import Kafka.Producer.Callbacks
@@ -37,7 +37,7 @@ data ProducerProperties = ProducerProperties
3737
{ ppKafkaProps :: Map Text Text
3838
, ppTopicProps :: Map Text Text
3939
, ppLogLevel :: Maybe KafkaLogLevel
40-
, ppCallbacks :: [KafkaConf -> IO ()]
40+
, ppCallbacks :: [Callback]
4141
}
4242

4343
instance Sem.Semigroup ProducerProperties where
@@ -70,7 +70,7 @@ brokersList bs =
7070
-- * 'errorCallback'
7171
-- * 'logCallback'
7272
-- * 'statsCallback'
73-
setCallback :: (KafkaConf -> IO ()) -> ProducerProperties
73+
setCallback :: Callback -> ProducerProperties
7474
setCallback cb = mempty { ppCallbacks = [cb] }
7575

7676
-- | Sets the logging level.

0 commit comments

Comments
 (0)