hw-kafka-client 2.4.3 → 2.4.4
raw patch · 12 files changed
+147/−65 lines, 12 filesdep −temporary
Dependencies removed: temporary
Files
- example/ProducerExample.hs +1/−1
- hw-kafka-client.cabal +2/−4
- src/Kafka/Consumer.hs +19/−12
- src/Kafka/Consumer/ConsumerProperties.hs +7/−2
- src/Kafka/Consumer/Convert.hs +0/−18
- src/Kafka/Consumer/Subscription.hs +15/−9
- src/Kafka/Internal/RdKafka.chs +4/−1
- src/Kafka/Internal/Shared.hs +27/−0
- src/Kafka/Producer/Callbacks.hs +38/−15
- src/Kafka/Producer/ProducerProperties.hs +7/−2
- src/Kafka/Producer/Types.hs +23/−0
- src/Kafka/Types.hs +4/−1
example/ProducerExample.hs view
@@ -13,7 +13,7 @@ producerProps :: ProducerProperties producerProps = brokersList [BrokerAddress "localhost:9092"] <> sendTimeout (Timeout 10000)- <> setCallback (deliveryErrorsCallback print)+ <> setCallback (deliveryCallback print) <> logLevel KafkaLogDebug -- Topic to send messages to
hw-kafka-client.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 8d76dba4ea94855b431d568543c5cd5d2f86a3bab474cd1151a9f936673891fc+-- hash: ae89ef8878b399834f3550efb6ad9b04c27fad1d06d9d22fecfc872a4f0d23c4 name: hw-kafka-client-version: 2.4.3+version: 2.4.4 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka C library. .@@ -49,7 +49,6 @@ , bifunctors , bytestring , containers- , temporary , transformers , unix build-tools:@@ -94,7 +93,6 @@ , bytestring , containers , hw-kafka-client- , temporary , transformers , unix if !(flag(examples))
src/Kafka/Consumer.hs view
@@ -8,7 +8,7 @@ , committed, position, seek , pollMessage, pollConsumerEvents , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets-, storeOffsetMessage+, storeOffsets, storeOffsetMessage , closeConsumer -- ReExport Types , KafkaConsumer@@ -107,12 +107,20 @@ -- | Stores message's offset locally for the message's partition. storeOffsetMessage :: MonadIO m- => KafkaConsumer- -> ConsumerRecord k v- -> m (Maybe KafkaError)+ => KafkaConsumer+ -> ConsumerRecord k v+ -> m (Maybe KafkaError) storeOffsetMessage k m = liftIO $ toNativeTopicPartitionList [topicPartitionFromMessageForCommit m] >>= commitOffsetsStore k +-- | Stores offsets locally+storeOffsets :: MonadIO m+ => KafkaConsumer+ -> [TopicPartition]+ -> m (Maybe KafkaError)+storeOffsets k ps =+ liftIO $ toNativeTopicPartitionList ps >>= commitOffsetsStore k+ -- | Commit offsets for all currently assigned partitions. commitAllOffsets :: MonadIO m => OffsetCommit@@ -171,12 +179,11 @@ where seekAll = runExceptT $ do tr <- traverse (ExceptT . topicPair) tps- _ <- traverse (\(kt, p, o) -> ExceptT (rdSeek kt p o)) tr+ void $ traverse (\(kt, p, o) -> ExceptT (rdSeek kt p o)) tr return () - rdSeek kt (PartitionId p) o = do- res <- rdKafkaSeek kt (fromIntegral p) (offsetToInt64 o) timeout- return $ rdKafkaErrorToEither res+ rdSeek kt (PartitionId p) o =+ rdKafkaErrorToEither <$> rdKafkaSeek kt (fromIntegral p) (offsetToInt64 o) timeout topicPair tp = do let (TopicName tn) = tpTopicName tp@@ -228,7 +235,7 @@ CToken.cancel ct mbq <- readIORef qr void $ traverse rdKafkaQueueDestroy mbq- (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaConsumerClose k+ kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaConsumerClose k ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf@@ -256,11 +263,11 @@ commitOffsets :: OffsetCommit -> KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError) commitOffsets o (KafkaConsumer (Kafka k) _) pl =- (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaCommit k pl (offsetCommitToBool o)+ kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaCommit k pl (offsetCommitToBool o) commitOffsetsStore :: KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError) commitOffsetsStore (KafkaConsumer (Kafka k) _) pl =- (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaOffsetsStore k pl+ kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaOffsetsStore k pl setConsumerLogLevel :: KafkaConsumer -> KafkaLogLevel -> IO () setConsumerLogLevel (KafkaConsumer (Kafka k) _) level =@@ -268,7 +275,7 @@ redirectCallbacksPoll :: KafkaConsumer -> IO (Maybe KafkaError) redirectCallbacksPoll (KafkaConsumer (Kafka k) _) =- (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaPollSetConsumer k+ kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaPollSetConsumer k runConsumerLoop :: KafkaConsumer -> CancellationToken -> Maybe Timeout -> IO () runConsumerLoop k ct timeout =
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -9,6 +9,7 @@ import qualified Data.List as L import Data.Map (Map) import qualified Data.Map as M+import Data.Semigroup as Sem import Kafka.Consumer.Types import Kafka.Internal.Setup import Kafka.Types@@ -22,6 +23,11 @@ , cpCallbacks :: [KafkaConf -> IO ()] } +instance Sem.Semigroup ConsumerProperties where+ (ConsumerProperties m1 ll1 cb1) <> (ConsumerProperties m2 ll2 cb2) =+ ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)+ {-# INLINE (<>) #-}+ -- | /Right biased/ so we prefer newer properties over older ones. instance Monoid ConsumerProperties where mempty = ConsumerProperties@@ -30,8 +36,7 @@ , cpCallbacks = [] } {-# INLINE mempty #-}- mappend (ConsumerProperties m1 ll1 cb1) (ConsumerProperties m2 ll2 cb2) =- ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)+ mappend = (Sem.<>) {-# INLINE mappend #-} brokersList :: [BrokerAddress] -> ConsumerProperties
src/Kafka/Consumer/Convert.hs view
@@ -132,24 +132,6 @@ , crValue = payload } - readTopic msg = newForeignPtr_ (topic'RdKafkaMessageT msg) >>= rdKafkaTopicName- readPayload = readBS len'RdKafkaMessageT payload'RdKafkaMessageT- readKey = readBS keyLen'RdKafkaMessageT key'RdKafkaMessageT- readTimestamp msg =- alloca $ \p -> do- typeP <- newForeignPtr_ p- ts <- fromIntegral <$> rdKafkaMessageTimestamp msg typeP- tsType <- peek p- return $ case tsType of- RdKafkaTimestampCreateTime -> CreateTime (Millis ts)- RdKafkaTimestampLogAppendTime -> LogAppendTime (Millis ts)- RdKafkaTimestampNotAvailable -> NoTimestamp-- readBS flen fdata s = if fdata s == nullPtr- then return Nothing- else Just <$> word8PtrToBS (flen s) (fdata s)-- offsetCommitToBool :: OffsetCommit -> Bool offsetCommitToBool OffsetCommit = False offsetCommitToBool OffsetCommitAsync = True
src/Kafka/Consumer/Subscription.hs view
@@ -1,21 +1,27 @@ module Kafka.Consumer.Subscription where ----import Data.Map (Map)-import qualified Data.Map as M-import qualified Data.List as L-import Kafka.Types-import Kafka.Consumer.Types+import qualified Data.List as L+import Data.Map (Map)+import qualified Data.Map as M+import Data.Semigroup as Sem+import Kafka.Consumer.Types+import Kafka.Types data Subscription = Subscription [TopicName] (Map String String) -instance Monoid Subscription where- mempty = Subscription [] M.empty- mappend (Subscription ts1 m1) (Subscription ts2 m2) =+instance Sem.Semigroup Subscription where+ (Subscription ts1 m1) <> (Subscription ts2 m2) = let ts' = L.nub $ L.union ts1 ts2 ps' = M.union m1 m2 in Subscription ts' ps'+ {-# INLINE (<>) #-}++instance Monoid Subscription where+ mempty = Subscription [] M.empty+ {-# INLINE mempty #-}+ mappend = (Sem.<>)+ {-# INLINE mappend #-} topics :: [TopicName] -> Subscription topics ts = Subscription (L.nub ts) M.empty
src/Kafka/Internal/RdKafka.chs view
@@ -26,7 +26,7 @@ {#enum rd_kafka_type_t as ^ {underscoreToCase} deriving (Show, Eq) #} {#enum rd_kafka_conf_res_t as ^ {underscoreToCase} deriving (Show, Eq) #}-{#enum rd_kafka_resp_err_t as ^ {underscoreToCase} deriving (Show, Eq) #}+{#enum rd_kafka_resp_err_t as ^ {underscoreToCase} deriving (Show, Eq, Bounded) #} {#enum rd_kafka_timestamp_type_t as ^ {underscoreToCase} deriving (Show, Eq) #} type RdKafkaMsgFlag = Int@@ -47,6 +47,9 @@ {} -> `String' #} {#fun pure rd_kafka_err2str as ^+ {enumToCInt `RdKafkaRespErrT'} -> `String' #}++{#fun pure rd_kafka_err2name as ^ {enumToCInt `RdKafkaRespErrT'} -> `String' #} {#fun pure rd_kafka_errno2err as ^
@@ -6,7 +6,9 @@ import Control.Monad (void, when) import qualified Data.ByteString as BS import qualified Data.ByteString.Internal as BSI+import Foreign hiding (void) import Foreign.C.Error+import Kafka.Consumer.Types import Kafka.Internal.CancellationToken as CToken import Kafka.Internal.RdKafka import Kafka.Internal.Setup@@ -70,3 +72,28 @@ maybeToLeft :: Maybe a -> Either a () maybeToLeft = maybe (Right ()) Left {-# INLINE maybeToLeft #-}++readPayload :: RdKafkaMessageT -> IO (Maybe BS.ByteString)+readPayload = readBS len'RdKafkaMessageT payload'RdKafkaMessageT++readTopic :: RdKafkaMessageT -> IO String+readTopic msg = newForeignPtr_ (topic'RdKafkaMessageT msg) >>= rdKafkaTopicName++readKey :: RdKafkaMessageT -> IO (Maybe BSI.ByteString)+readKey = readBS keyLen'RdKafkaMessageT key'RdKafkaMessageT++readTimestamp :: RdKafkaMessageTPtr -> IO Timestamp+readTimestamp msg =+ alloca $ \p -> do+ typeP <- newForeignPtr_ p+ ts <- fromIntegral <$> rdKafkaMessageTimestamp msg typeP+ tsType <- peek p+ return $ case tsType of+ RdKafkaTimestampCreateTime -> CreateTime (Millis ts)+ RdKafkaTimestampLogAppendTime -> LogAppendTime (Millis ts)+ RdKafkaTimestampNotAvailable -> NoTimestamp++readBS :: (t -> Int) -> (t -> Ptr Word8) -> t -> IO (Maybe BS.ByteString)+readBS flen fdata s = if fdata s == nullPtr+ then return Nothing+ else Just <$> word8PtrToBS (flen s) (fdata s)
src/Kafka/Producer/Callbacks.hs view
@@ -1,28 +1,51 @@ module Kafka.Producer.Callbacks-( deliveryErrorsCallback+( deliveryCallback , module X ) where -import Foreign-import Foreign.C.Error-import Kafka.Callbacks as X-import Kafka.Internal.RdKafka-import Kafka.Internal.Setup-import Kafka.Internal.Shared-import Kafka.Types+import Foreign+import Foreign.C.Error+import Kafka.Callbacks as X+import Kafka.Consumer.Types+import Kafka.Internal.RdKafka+import Kafka.Internal.Setup+import Kafka.Internal.Shared+import Kafka.Producer.Types+import Kafka.Types --- | Sets the callback for delivery errors.--- The callback is only called in case of errors.-deliveryErrorsCallback :: (KafkaError -> IO ()) -> KafkaConf -> IO ()-deliveryErrorsCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb+-- | Sets the callback for delivery reports.+deliveryCallback :: (DeliveryReport -> IO ()) -> KafkaConf -> IO ()+deliveryCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb where realCb :: t -> Ptr RdKafkaMessageT -> IO () realCb _ mptr = if mptr == nullPtr- then getErrno >>= (callback . kafkaRespErr)+ then getErrno >>= (callback . NoMessageError . kafkaRespErr) else do s <- peek mptr if err'RdKafkaMessageT s /= RdKafkaRespErrNoError- then (callback . KafkaResponseError $ err'RdKafkaMessageT s)- else pure ()+ then mkErrorReport s >>= callback+ else mkSuccessReport s >>= callback++mkErrorReport :: RdKafkaMessageT -> IO DeliveryReport+mkErrorReport msg = do+ prodRec <- mkProdRec msg+ pure $ DeliveryFailure prodRec (KafkaResponseError (err'RdKafkaMessageT msg))++mkSuccessReport :: RdKafkaMessageT -> IO DeliveryReport+mkSuccessReport msg = do+ prodRec <- mkProdRec msg+ pure $ DeliverySuccess prodRec (Offset $ offset'RdKafkaMessageT msg)++mkProdRec :: RdKafkaMessageT -> IO ProducerRecord+mkProdRec msg = do+ topic <- readTopic msg+ key <- readKey msg+ payload <- readPayload msg+ pure ProducerRecord+ { prTopic = TopicName topic+ , prPartition = SpecifiedPartition (partition'RdKafkaMessageT msg)+ , prKey = key+ , prValue = payload+ }
src/Kafka/Producer/ProducerProperties.hs view
@@ -8,6 +8,7 @@ import qualified Data.List as L import Data.Map (Map) import qualified Data.Map as M+import Data.Semigroup as Sem import Kafka.Internal.Setup import Kafka.Producer.Callbacks import Kafka.Types@@ -20,6 +21,11 @@ , ppCallbacks :: [KafkaConf -> IO ()] } +instance Sem.Semigroup ProducerProperties where+ (ProducerProperties k1 t1 ll1 cb1) <> (ProducerProperties k2 t2 ll2 cb2) =+ ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)+ {-# INLINE (<>) #-}+ -- | /Right biased/ so we prefer newer properties over older ones. instance Monoid ProducerProperties where mempty = ProducerProperties@@ -29,8 +35,7 @@ , ppCallbacks = [] } {-# INLINE mempty #-}- mappend (ProducerProperties k1 t1 ll1 cb1) (ProducerProperties k2 t2 ll2 cb2) =- ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)+ mappend = (Sem.<>) {-# INLINE mappend #-} brokersList :: [BrokerAddress] -> ProducerProperties
src/Kafka/Producer/Types.hs view
@@ -7,6 +7,7 @@ import Data.Typeable import Kafka.Internal.Setup import Kafka.Types+import Kafka.Consumer.Types -- | Main pointer to Kafka object, which contains our brokers data KafkaProducer = KafkaProducer@@ -39,3 +40,25 @@ SpecifiedPartition {-# UNPACK #-} !Int -- the partition number of the topic | UnassignedPartition deriving (Show, Eq, Ord, Typeable)++data DeliveryReport+ = DeliverySuccess ProducerRecord Offset+ | DeliveryFailure ProducerRecord KafkaError+ | NoMessageError KafkaError+ deriving (Show, Eq)++-- -- | Represents the report of a successfully delivered message.+-- data ProducerSuccess = ProducerSuccess+-- { psTopic :: !TopicName -- ^ Kafka topic this message was received from+-- , psPartition :: !PartitionId -- ^ Kafka partition this message was received from+-- , psOffset :: !Offset -- ^ Offset within the 'crPartition' Kafka partition+-- , psKey :: !(Maybe BS.ByteString)+-- , psValue :: !(Maybe BS.ByteString)+-- }+-- deriving (Eq, Show, Read, Typeable)++-- -- | Represents the failure to deliver a message.+-- data ProducerError = ProducerError+-- { peValue :: !(Maybe BS.ByteString) -- ^ The message that failed.+-- , peError :: KafkaError -- ^ The reason for the failure.+-- } deriving (Eq, Show)
src/Kafka/Types.hs view
@@ -68,7 +68,10 @@ | KafkaBadConfiguration deriving (Eq, Show, Typeable) -instance Exception KafkaError+instance Exception KafkaError where+ displayException (KafkaResponseError err) =+ "[" ++ rdKafkaErr2name err ++ "] " ++ rdKafkaErr2str err+ displayException err = show err data KafkaDebug = DebugGeneric