diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ To run an example please compile with the `examples` flag. ```Haskell +import Control.Exception (bracket) import Data.Monoid ((<>)) import Kafka import Kafka.Consumer @@ -52,8 +53,14 @@ -- Running an example runConsumerExample :: IO () runConsumerExample = do - res <- runConsumer consumerProps consumerSub processMessages + res <- bracket mkConsumer clConsumer runHandler print res + where + mkConsumer = newConsumer consumerProps consumerSub + clConsumer (Left err) = return (Left err) + clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc + runHandler (Left err) = return (Left err) + runHandler (Right kc) = processMessages kc ------------------------------------------------------------------- processMessages :: KafkaConsumer -> IO (Either KafkaError ()) @@ -111,8 +118,10 @@ ### Example ```Haskell +{-# LANGUAGE OverloadedStrings #-} +import Control.Exception (bracket) import Control.Monad (forM_) -import Kafka +import Data.ByteString (ByteString) import Kafka.Producer -- Global producer properties @@ -126,9 +135,14 @@ -- Run an example runProducerExample :: IO () -runProducerExample = do - res <- runProducer producerProps sendMessages - print res +runProducerExample = + bracket mkProducer clProducer runHandler >>= print + where + mkProducer = newProducer producerProps + clProducer (Left _) = return () + clProducer (Right prod) = closeProducer prod + runHandler (Left err) = return $ Left err + runHandler (Right prod) = sendMessages prod sendMessages :: KafkaProducer -> IO (Either KafkaError ()) sendMessages prod = do diff --git a/example/ConsumerExample.hs b/example/ConsumerExample.hs --- a/example/ConsumerExample.hs +++ b/example/ConsumerExample.hs @@ -1,11 +1,14 @@ +{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} module ConsumerExample where -import Control.Arrow ((&&&)) -import Data.Monoid ((<>)) +import Control.Arrow ((&&&)) +import Control.Exception (bracket) +import Data.Monoid ((<>)) import Kafka.Consumer +import Data.Text (Text) -- Global consumer properties consumerProps :: ConsumerProperties @@ -25,8 +28,14 @@ runConsumerExample :: IO () runConsumerExample = do print $ cpLogLevel consumerProps - res <- runConsumer consumerProps consumerSub processMessages + res <- bracket mkConsumer clConsumer runHandler print res + where + mkConsumer = newConsumer consumerProps consumerSub + clConsumer (Left err) = return (Left err) + clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc + runHandler (Left err) = return (Left err) + runHandler (Right kc) = processMessages kc ------------------------------------------------------------------- processMessages :: KafkaConsumer -> IO (Either KafkaError ()) diff --git a/example/ProducerExample.hs b/example/ProducerExample.hs --- a/example/ProducerExample.hs +++ b/example/ProducerExample.hs @@ -3,11 +3,13 @@ module ProducerExample where +import Control.Exception (bracket) import Control.Monad (forM_) import Data.ByteString (ByteString) import Data.ByteString.Char8 (pack) import Data.Monoid import Kafka.Producer +import Data.Text (Text) -- Global producer properties producerProps :: ProducerProperties @@ -30,9 +32,14 @@ -- Run an example runProducerExample :: IO () -runProducerExample = do - res <- runProducer producerProps sendMessages - print res +runProducerExample = + bracket mkProducer clProducer runHandler >>= print + where + mkProducer = newProducer producerProps + clProducer (Left _) = return () + clProducer (Right prod) = closeProducer prod + runHandler (Left err) = return $ Left err + runHandler (Right prod) = sendMessages prod sendMessages :: KafkaProducer -> IO (Either KafkaError ()) sendMessages prod = do diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -1,11 +1,5 @@ --- This file has been generated from package.yaml by hpack version 0.28.2. --- --- see: https://github.com/sol/hpack --- --- hash: c01d07e897b9cf5d0f6237227a00230ec841fe4a3a22766142dc015f4f703ba1 - name: hw-kafka-client -version: 2.5.0 +version: 2.6.0 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka C library. . @@ -40,7 +34,12 @@ library hs-source-dirs: src - ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints + ghc-options: + -Wall + -Wcompat + -Wincomplete-record-updates + -Wincomplete-uni-patterns + -Wredundant-constraints extra-libraries: rdkafka build-depends: @@ -48,6 +47,7 @@ , bifunctors , bytestring , containers + , text , transformers , unix build-tools: @@ -92,6 +92,7 @@ , bytestring , containers , hw-kafka-client + , text , transformers , unix if !(flag(examples)) @@ -117,6 +118,8 @@ , hspec , hw-kafka-client , monad-loops + , text + , transformers other-modules: Kafka.IntegrationSpec Kafka.TestEnv @@ -137,6 +140,7 @@ , either , hspec , hw-kafka-client + , text , monad-loops other-modules: Kafka.Consumer.ConsumerRecordMapSpec diff --git a/src/Kafka/Callbacks.hs b/src/Kafka/Callbacks.hs --- a/src/Kafka/Callbacks.hs +++ b/src/Kafka/Callbacks.hs @@ -5,9 +5,9 @@ ) where -import Kafka.Internal.RdKafka -import Kafka.Internal.Setup -import Kafka.Types +import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb) +import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf) +import Kafka.Types (KafkaError(..)) errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO () errorCallback callback k = diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections #-} module Kafka.Consumer ( module X @@ -17,25 +18,73 @@ ) where -import Control.Arrow +import Data.Set (Set) +import qualified Data.Set as Set +import qualified Data.Text as Text +import Control.Arrow ((&&&), left) import Control.Concurrent (forkIO, rtsSupportsBoundThreads) -import Control.Exception +import Control.Exception (bracket) import Control.Monad (forM_, void, when) -import Control.Monad.IO.Class -import Control.Monad.Trans.Except -import Data.Bifunctor +import Control.Monad.IO.Class (MonadIO(liftIO)) +import Control.Monad.Trans.Except (ExceptT(ExceptT), runExceptT) +import Data.Bifunctor (first, bimap) import qualified Data.ByteString as BS -import Data.IORef +import Data.IORef (writeIORef, readIORef) import qualified Data.Map as M import Data.Maybe (fromMaybe) import Data.Monoid ((<>)) import Foreign hiding (void) import Kafka.Consumer.Convert -import Kafka.Consumer.Types + (fromMessagePtr, toNativeTopicPartitionList, topicPartitionFromMessageForCommit + , toNativeTopicPartitionListNoDispose, toNativeTopicPartitionList, fromNativeTopicPartitionList'' + , toMap, offsetToInt64, toNativeTopicPartitionList', offsetCommitToBool + ) +import Kafka.Consumer.Types (KafkaConsumer(..)) import Kafka.Internal.CancellationToken as CToken import Kafka.Internal.RdKafka + ( RdKafkaRespErrT(..) + , RdKafkaTopicPartitionListTPtr + , RdKafkaTypeT(..) + , newRdKafkaT + , rdKafkaQueueNew + , rdKafkaConsumeQueue + , rdKafkaPollSetConsumer + , rdKafkaSetLogLevel + , rdKafkaOffsetsStore + , rdKafkaCommit + , rdKafkaConfSetDefaultTopicConf + , rdKafkaTopicConfDup + , rdKafkaSubscribe + , rdKafkaTopicPartitionListAdd + , newRdKafkaTopicPartitionListT + , rdKafkaConsumerClose + , rdKafkaQueueDestroy + , rdKafkaConsumerPoll + , rdKafkaPosition + , rdKafkaCommitted + , rdKafkaSeek + , rdKafkaResumePartitions + , rdKafkaPausePartitions + , rdKafkaSubscription + , rdKafkaAssignment + , rdKafkaConsumeBatchQueue + , newRdKafkaTopicT + ) import Kafka.Internal.Setup + ( Kafka(..) + , KafkaConf(..) + , TopicConf(..) + , KafkaProps(..) + , TopicProps(..) + , kafkaConf + , topicConf + , getRdKafka + ) import Kafka.Internal.Shared + ( kafkaErrorToMaybe + , maybeToLeft + , rdKafkaErrorToEither + ) import Kafka.Consumer.ConsumerProperties as X import Kafka.Consumer.Subscription as X @@ -67,7 +116,7 @@ newConsumer props (Subscription ts tp) = liftIO $ do let cp = setCallback (rebalanceCallback (\_ _ -> return ())) <> props kc@(KafkaConf kc' qref ct) <- newConsumerConf cp - tp' <- topicConf (TopicProps $ M.toList tp) + tp' <- topicConf (TopicProps tp) _ <- setDefaultTopicConf kc tp' rdk <- newRdKafkaT RdKafkaConsumer kc' case rdk of @@ -179,14 +228,14 @@ pausePartitions :: MonadIO m => KafkaConsumer -> [(TopicName, PartitionId)] -> m KafkaError pausePartitions (KafkaConsumer (Kafka k) _) ps = liftIO $ do pl <- newRdKafkaTopicPartitionListT (length ps) - mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl topicName partitionId) ps + mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl (Text.unpack topicName) partitionId) ps KafkaResponseError <$> rdKafkaPausePartitions k pl -- | Resumes specified partitions on the current consumer. resumePartitions :: MonadIO m => KafkaConsumer -> [(TopicName, PartitionId)] -> m KafkaError resumePartitions (KafkaConsumer (Kafka k) _) ps = liftIO $ do pl <- newRdKafkaTopicPartitionListT (length ps) - mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl topicName partitionId) ps + mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl (Text.unpack topicName) partitionId) ps KafkaResponseError <$> rdKafkaResumePartitions k pl seek :: MonadIO m => KafkaConsumer -> Timeout -> [TopicPartition] -> m (Maybe KafkaError) @@ -203,8 +252,8 @@ topicPair tp = do let (TopicName tn) = tpTopicName tp - nt <- newRdKafkaTopicT k tn Nothing - return $ bimap KafkaError (,tpPartition tp, tpOffset tp) nt + nt <- newRdKafkaTopicT k (Text.unpack tn) Nothing + return $ bimap KafkaError (,tpPartition tp, tpOffset tp) (first Text.pack nt) -- | Retrieve committed offsets for topics+partitions. committed :: MonadIO m => KafkaConsumer -> Timeout -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition]) @@ -256,7 +305,7 @@ ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do - conf <- kafkaConf (KafkaProps $ M.toList m) + conf <- kafkaConf (KafkaProps m) forM_ cbs (\setCb -> setCb conf) return conf @@ -266,10 +315,10 @@ -- any topic name in the topics list that is prefixed with @^@ will -- be regex-matched to the full list of topics in the cluster and matching -- topics will be added to the subscription list. -subscribe :: KafkaConsumer -> [TopicName] -> IO (Maybe KafkaError) +subscribe :: KafkaConsumer -> Set TopicName -> IO (Maybe KafkaError) subscribe (KafkaConsumer (Kafka k) _) ts = do pl <- newRdKafkaTopicPartitionListT (length ts) - mapM_ (\(TopicName t) -> rdKafkaTopicPartitionListAdd pl t (-1)) ts + mapM_ (\(TopicName t) -> rdKafkaTopicPartitionListAdd pl (Text.unpack t) (-1)) (Set.toList ts) res <- KafkaResponseError <$> rdKafkaSubscribe k pl return $ kafkaErrorToMaybe res diff --git a/src/Kafka/Consumer/Callbacks.hs b/src/Kafka/Consumer/Callbacks.hs --- a/src/Kafka/Consumer/Callbacks.hs +++ b/src/Kafka/Consumer/Callbacks.hs @@ -6,19 +6,20 @@ ) where -import Control.Arrow ((&&&)) -import Control.Monad (forM_, void) -import Data.Monoid ((<>)) -import Foreign hiding (void) -import Kafka.Callbacks as X -import Kafka.Consumer.Convert -import Kafka.Consumer.Types -import Kafka.Internal.RdKafka -import Kafka.Internal.Setup -import Kafka.Internal.Shared -import Kafka.Types - -import Control.Concurrent (threadDelay) +import Control.Arrow ((&&&)) +import Control.Concurrent (threadDelay) +import Control.Monad (forM_, void) +import Data.Monoid ((<>)) +import qualified Data.Text as Text +import Foreign.ForeignPtr (newForeignPtr_) +import Foreign.Ptr (nullPtr) +import Kafka.Callbacks as X +import Kafka.Consumer.Convert (fromNativeTopicPartitionList', fromNativeTopicPartitionList'', toNativeTopicPartitionList) +import Kafka.Consumer.Types (KafkaConsumer(..), TopicPartition(..), RebalanceEvent(..)) +import Kafka.Internal.RdKafka +import Kafka.Internal.Setup (Kafka(..), KafkaConf(..), HasKafka(..), HasKafkaConf(..), getRdMsgQueue) +import Kafka.Internal.Shared (kafkaErrorToMaybe) +import Kafka.Types (KafkaError(..), PartitionId(..), TopicName(..)) -- | Sets a callback that is called when rebalance is needed. -- @@ -62,7 +63,7 @@ ------------------------------------------------------------------------------- redirectPartitionQueue :: Kafka -> TopicName -> PartitionId -> RdKafkaQueueTPtr -> IO () redirectPartitionQueue (Kafka k) (TopicName t) (PartitionId p) q = do - mpq <- rdKafkaQueueGetPartition k t p + mpq <- rdKafkaQueueGetPartition k (Text.unpack t) p case mpq of Nothing -> return () Just pq -> rdKafkaQueueForward pq q diff --git a/src/Kafka/Consumer/ConsumerProperties.hs b/src/Kafka/Consumer/ConsumerProperties.hs --- a/src/Kafka/Consumer/ConsumerProperties.hs +++ b/src/Kafka/Consumer/ConsumerProperties.hs @@ -1,24 +1,39 @@ +{-# LANGUAGE OverloadedStrings #-} + module Kafka.Consumer.ConsumerProperties -( module Kafka.Consumer.ConsumerProperties +( ConsumerProperties(..) +, brokersList +, noAutoCommit +, noAutoOffsetStore +, groupId +, clientId +, setCallback +, logLevel +, compression +, suppressDisconnectLogs +, extraProps +, extraProp +, debugOptions +, queuedMaxMessagesKBytes , module X ) where --- -import Control.Monad -import qualified Data.List as L +import Control.Monad (MonadPlus(mplus)) 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 +import Data.Text (Text) +import qualified Data.Text as Text +import Kafka.Consumer.Types (ConsumerGroupId(..)) +import Kafka.Internal.Setup (KafkaConf(..)) +import Kafka.Types (KafkaDebug(..), KafkaCompressionCodec(..), KafkaLogLevel(..), ClientId(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText) import Kafka.Consumer.Callbacks as X -- | Properties to create 'KafkaConsumer'. data ConsumerProperties = ConsumerProperties - { cpProps :: Map String String + { cpProps :: Map Text Text , cpLogLevel :: Maybe KafkaLogLevel , cpCallbacks :: [KafkaConf -> IO ()] } @@ -41,7 +56,7 @@ brokersList :: [BrokerAddress] -> ConsumerProperties brokersList bs = - let bs' = L.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) + let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) in extraProps $ M.fromList [("bootstrap.servers", bs')] -- | Disables auto commit for the consumer @@ -74,7 +89,7 @@ -- | Sets the compression codec for the consumer. compression :: KafkaCompressionCodec -> ConsumerProperties compression c = - extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToString c) + extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) -- | Suppresses consumer disconnects logs. -- @@ -86,13 +101,13 @@ -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found -extraProps :: Map String String -> ConsumerProperties +extraProps :: Map Text Text -> ConsumerProperties extraProps m = mempty { cpProps = m } {-# INLINE extraProps #-} -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found -extraProp :: String -> String -> ConsumerProperties +extraProp :: Text -> Text -> ConsumerProperties extraProp k v = mempty { cpProps = M.singleton k v } {-# INLINE extraProp #-} @@ -101,10 +116,10 @@ debugOptions :: [KafkaDebug] -> ConsumerProperties debugOptions [] = extraProps M.empty debugOptions d = - let points = L.intercalate "," (kafkaDebugToString <$> d) + let points = Text.intercalate "," (kafkaDebugToText <$> d) in extraProps $ M.fromList [("debug", points)] queuedMaxMessagesKBytes :: Int -> ConsumerProperties queuedMaxMessagesKBytes kBytes = - extraProp "queued.max.messages.kbytes" (show kBytes) + extraProp "queued.max.messages.kbytes" (Text.pack $ show kBytes) {-# INLINE queuedMaxMessagesKBytes #-} diff --git a/src/Kafka/Consumer/Convert.hs b/src/Kafka/Consumer/Convert.hs --- a/src/Kafka/Consumer/Convert.hs +++ b/src/Kafka/Consumer/Convert.hs @@ -1,18 +1,48 @@ module Kafka.Consumer.Convert - +( offsetSyncToInt +, offsetToInt64 +, int64ToOffset +, fromNativeTopicPartitionList'' +, fromNativeTopicPartitionList' +, fromNativeTopicPartitionList +, toNativeTopicPartitionList +, toNativeTopicPartitionListNoDispose +, toNativeTopicPartitionList' +, topicPartitionFromMessage +, topicPartitionFromMessageForCommit +, toMap +, fromMessagePtr +, offsetCommitToBool +) where -import Control.Monad +import Control.Monad ((>=>)) import qualified Data.ByteString as BS +import Data.Int (Int64) import Data.Map.Strict (Map, fromListWith) import qualified Data.Set as S -import Foreign -import Foreign.C.Error -import Foreign.C.String -import Kafka.Consumer.Types +import qualified Data.Text as Text +import Foreign.Ptr (Ptr, nullPtr) +import Foreign.ForeignPtr (withForeignPtr) +import Foreign.Storable (Storable(..)) +import Foreign.C.Error (getErrno) +import Kafka.Consumer.Types (ConsumerRecord(..), TopicPartition(..), Offset(..), OffsetCommit(..), PartitionOffset(..), OffsetStoreSync(..)) import Kafka.Internal.RdKafka -import Kafka.Internal.Shared -import Kafka.Types + ( RdKafkaRespErrT(..) + , RdKafkaMessageT(..) + , RdKafkaTopicPartitionListTPtr + , RdKafkaTopicPartitionListT(..) + , RdKafkaMessageTPtr + , RdKafkaTopicPartitionT(..) + , rdKafkaTopicPartitionListAdd + , newRdKafkaTopicPartitionListT + , rdKafkaMessageDestroy + , rdKafkaTopicPartitionListSetOffset + , rdKafkaTopicPartitionListNew + , peekCText + ) +import Kafka.Internal.Shared (kafkaRespErr, readTopic, readKey, readPayload, readTimestamp) +import Kafka.Types (KafkaError(..), PartitionId(..), TopicName(..)) -- | Converts offsets sync policy to integer (the way Kafka understands it): -- @@ -65,7 +95,7 @@ where toPart :: RdKafkaTopicPartitionT -> IO TopicPartition toPart p = do - topic <- peekCString $ topic'RdKafkaTopicPartitionT p + topic <- peekCText $ topic'RdKafkaTopicPartitionT p return TopicPartition { tpTopicName = TopicName topic, tpPartition = PartitionId $ partition'RdKafkaTopicPartitionT p, @@ -79,8 +109,9 @@ let TopicName tn = tpTopicName p (PartitionId tp) = tpPartition p to = offsetToInt64 $ tpOffset p - _ <- rdKafkaTopicPartitionListAdd pl tn tp - rdKafkaTopicPartitionListSetOffset pl tn tp to) ps + tnS = Text.unpack tn + _ <- rdKafkaTopicPartitionListAdd pl tnS tp + rdKafkaTopicPartitionListSetOffset pl tnS tp to) ps return pl toNativeTopicPartitionListNoDispose :: [TopicPartition] -> IO RdKafkaTopicPartitionListTPtr @@ -90,15 +121,16 @@ let TopicName tn = tpTopicName p (PartitionId tp) = tpPartition p to = offsetToInt64 $ tpOffset p - _ <- rdKafkaTopicPartitionListAdd pl tn tp - rdKafkaTopicPartitionListSetOffset pl tn tp to) ps + tnS = Text.unpack tn + _ <- rdKafkaTopicPartitionListAdd pl tnS tp + rdKafkaTopicPartitionListSetOffset pl tnS tp to) ps return pl toNativeTopicPartitionList' :: [(TopicName, PartitionId)] -> IO RdKafkaTopicPartitionListTPtr toNativeTopicPartitionList' tps = do let utps = S.toList . S.fromList $ tps pl <- newRdKafkaTopicPartitionListT (length utps) - mapM_ (\(TopicName t, PartitionId p) -> rdKafkaTopicPartitionListAdd pl t p) utps + mapM_ (\(TopicName t, PartitionId p) -> rdKafkaTopicPartitionListAdd pl (Text.unpack t) p) utps return pl topicPartitionFromMessage :: ConsumerRecord k v -> TopicPartition diff --git a/src/Kafka/Consumer/Subscription.hs b/src/Kafka/Consumer/Subscription.hs --- a/src/Kafka/Consumer/Subscription.hs +++ b/src/Kafka/Consumer/Subscription.hs @@ -1,44 +1,55 @@ +{-# LANGUAGE OverloadedStrings #-} + module Kafka.Consumer.Subscription +( Subscription(..) +, topics +, offsetReset +, autoCommit +, extraSubscriptionProps +) where -import qualified Data.List as L +import qualified Data.Text as Text +import Data.Text (Text) import Data.Map (Map) import qualified Data.Map as M import Data.Semigroup as Sem -import Kafka.Consumer.Types -import Kafka.Types +import Kafka.Consumer.Types (OffsetReset(..)) +import Kafka.Types (TopicName(..), Millis(..)) +import Data.Set (Set) +import qualified Data.Set as Set -data Subscription = Subscription [TopicName] (Map String String) +data Subscription = Subscription (Set TopicName) (Map Text Text) instance Sem.Semigroup Subscription where (Subscription ts1 m1) <> (Subscription ts2 m2) = - let ts' = L.nub $ L.union ts1 ts2 + let ts' = Set.union ts1 ts2 ps' = M.union m1 m2 in Subscription ts' ps' {-# INLINE (<>) #-} instance Monoid Subscription where - mempty = Subscription [] M.empty + mempty = Subscription Set.empty M.empty {-# INLINE mempty #-} mappend = (Sem.<>) {-# INLINE mappend #-} topics :: [TopicName] -> Subscription -topics ts = Subscription (L.nub ts) M.empty +topics ts = Subscription (Set.fromList ts) M.empty offsetReset :: OffsetReset -> Subscription offsetReset o = let o' = case o of Earliest -> "earliest" Latest -> "latest" - in Subscription [] (M.fromList [("auto.offset.reset", o')]) + in Subscription (Set.empty) (M.fromList [("auto.offset.reset", o')]) autoCommit :: Millis -> Subscription -autoCommit (Millis ms) = Subscription [] $ +autoCommit (Millis ms) = Subscription (Set.empty) $ M.fromList [ ("enable.auto.commit", "true") - , ("auto.commit.interval.ms", show ms) + , ("auto.commit.interval.ms", Text.pack $ show ms) ] -extraSubscriptionProps :: Map String String -> Subscription -extraSubscriptionProps = Subscription [] +extraSubscriptionProps :: Map Text Text -> Subscription +extraSubscriptionProps = Subscription (Set.empty) diff --git a/src/Kafka/Consumer/Types.hs b/src/Kafka/Consumer/Types.hs --- a/src/Kafka/Consumer/Types.hs +++ b/src/Kafka/Consumer/Types.hs @@ -1,15 +1,42 @@ {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} module Kafka.Consumer.Types +( KafkaConsumer(..) +, ConsumerGroupId(..) +, Offset(..) +, OffsetReset(..) +, RebalanceEvent(..) +, PartitionOffset(..) +, SubscribedPartitions(..) +, Timestamp(..) +, OffsetCommit(..) +, OffsetStoreSync(..) +, OffsetStoreMethod(..) +, TopicPartition(..) +, ConsumerRecord(..) +, crMapKey +, crMapValue +, crMapKV +-- why are these here? +-- * Deprecated +, sequenceFirst +, traverseFirst +, traverseFirstM +, traverseM +, bitraverseM +) where -import Data.Bifoldable -import Data.Bifunctor -import Data.Bitraversable -import Data.Int -import Data.Typeable -import Kafka.Internal.Setup -import Kafka.Types +import Data.Bifoldable (Bifoldable (..)) +import Data.Bifunctor (Bifunctor (..)) +import Data.Bitraversable (Bitraversable (..), bimapM, bisequenceA) +import Data.Int (Int64) +import Data.Text (Text) +import Data.Typeable (Typeable) +import GHC.Generics (Generic) +import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..)) +import Kafka.Types (Millis (..), PartitionId (..), TopicName (..)) data KafkaConsumer = KafkaConsumer { kcKafkaPtr :: !Kafka @@ -24,9 +51,9 @@ getKafkaConf = kcKafkaConf {-# INLINE getKafkaConf #-} -newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: String} deriving (Show, Ord, Eq) -newtype Offset = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read) -data OffsetReset = Earliest | Latest deriving (Show, Eq) +newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: Text } deriving (Show, Ord, Eq, Generic) +newtype Offset = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read, Generic) +data OffsetReset = Earliest | Latest deriving (Show, Eq, Generic) -- | A set of events which happen during the rebalancing process data RebalanceEvent = @@ -38,7 +65,7 @@ | RebalanceBeforeRevoke [(TopicName, PartitionId)] -- | Happens after the rejection is confirmed | RebalanceRevoke [(TopicName, PartitionId)] - deriving (Eq, Show) + deriving (Eq, Show, Generic) data PartitionOffset = PartitionOffsetBeginning @@ -46,24 +73,24 @@ | PartitionOffset Int64 | PartitionOffsetStored | PartitionOffsetInvalid - deriving (Eq, Show) + deriving (Eq, Show, Generic) data SubscribedPartitions = SubscribedPartitions [PartitionId] | SubscribedPartitionsAll - deriving (Show, Eq) + deriving (Show, Eq, Generic) data Timestamp = CreateTime !Millis | LogAppendTime !Millis | NoTimestamp - deriving (Show, Eq, Read) + deriving (Show, Eq, Read, Generic) -- | Offsets commit mode data OffsetCommit = OffsetCommit -- ^ Forces consumer to block until the broker offsets commit is done | OffsetCommitAsync -- ^ Offsets will be committed in a non-blocking way - deriving (Show, Eq) + deriving (Show, Eq, Generic) -- | Indicates how offsets are to be synced to disk @@ -71,17 +98,20 @@ OffsetSyncDisable -- ^ Do not sync offsets (in Kafka: -1) | OffsetSyncImmediate -- ^ Sync immediately after each offset commit (in Kafka: 0) | OffsetSyncInterval Int -- ^ Sync after specified interval in millis + deriving (Show, Eq, Generic) -- | Indicates the method of storing the offsets data OffsetStoreMethod = OffsetStoreBroker -- ^ Offsets are stored in Kafka broker (preferred) | OffsetStoreFile FilePath OffsetStoreSync -- ^ Offsets are stored in a file (and synced to disk according to the sync policy) + deriving (Show, Eq, Generic) -- | Kafka topic partition structure data TopicPartition = TopicPartition { tpTopicName :: TopicName , tpPartition :: PartitionId - , tpOffset :: PartitionOffset } deriving (Show, Eq) + , tpOffset :: PartitionOffset + } deriving (Show, Eq, Generic) -- | Represents a /received/ message from Kafka (i.e. used in a consumer) data ConsumerRecord k v = ConsumerRecord @@ -92,7 +122,7 @@ , crKey :: !k , crValue :: !v } - deriving (Eq, Show, Read, Typeable) + deriving (Eq, Show, Read, Typeable, Generic) instance Bifunctor ConsumerRecord where bimap f g (ConsumerRecord t p o ts k v) = ConsumerRecord t p o ts (f k) (g v) @@ -130,10 +160,12 @@ crMapKV = bimap {-# INLINE crMapKV #-} +{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use 'bitraverse id pure'" #-} sequenceFirst :: (Bitraversable t, Applicative f) => t (f k) v -> f (t k v) sequenceFirst = bitraverse id pure {-# INLINE sequenceFirst #-} +{-# DEPRECATED traverseFirst "Isn't concern of this library. Use 'bitraverse f pure'" #-} traverseFirst :: (Bitraversable t, Applicative f) => (k -> f k') -> t k v @@ -141,6 +173,7 @@ traverseFirst f = bitraverse f pure {-# INLINE traverseFirst #-} +{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use 'bitraverse id pure <$> bitraverse f pure r'" #-} traverseFirstM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> t k v @@ -148,6 +181,7 @@ traverseFirstM f r = bitraverse id pure <$> bitraverse f pure r {-# INLINE traverseFirstM #-} +{-# DEPRECATED traverseM "Isn't concern of this library. Use 'sequenceA <$> traverse f r'" #-} traverseM :: (Traversable t, Applicative f, Monad m) => (v -> m (f v')) -> t v @@ -155,6 +189,7 @@ traverseM f r = sequenceA <$> traverse f r {-# INLINE traverseM #-} +{-# DEPRECATED bitraverseM "Isn't concern of this library. Use 'bisequenceA <$> bimapM f g r'" #-} bitraverseM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) diff --git a/src/Kafka/Dump.hs b/src/Kafka/Dump.hs --- a/src/Kafka/Dump.hs +++ b/src/Kafka/Dump.hs @@ -6,17 +6,33 @@ ) where --- import Kafka.Internal.RdKafka + ( CSizePtr + , rdKafkaConfDumpFree + , peekCText + , rdKafkaConfDump + , rdKafkaTopicConfDump + , rdKafkaDump + , handleToCFile + , rdKafkaConfPropertiesShow + ) import Kafka.Internal.Setup + ( HasKafka(..) + , HasTopicConf(..) + , HasKafkaConf(..) + , getRdKafka + , getRdTopicConf + , getRdKafkaConf + ) -import Control.Monad -import Control.Monad.IO.Class +import Control.Monad ((<=<)) +import Control.Monad.IO.Class (MonadIO(liftIO)) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map -import Foreign -import Foreign.C.String -import System.IO +import Foreign (Ptr, alloca, Storable(peek, peekElemOff)) +import Foreign.C.String (CString) +import System.IO (Handle) +import Data.Text (Text) -- | Prints out all supported Kafka conf properties to a handle hPrintSupportedKafkaConf :: MonadIO m => Handle -> m () @@ -27,25 +43,25 @@ hPrintKafka h k = liftIO $ handleToCFile h "w" >>= \f -> rdKafkaDump f (getRdKafka k) -- | Returns a map of the current topic configuration -dumpTopicConf :: (MonadIO m, HasTopicConf t) => t -> m (Map String String) +dumpTopicConf :: (MonadIO m, HasTopicConf t) => t -> m (Map Text Text) dumpTopicConf t = liftIO $ parseDump (rdKafkaTopicConfDump (getRdTopicConf t)) -- | Returns a map of the current kafka configuration -dumpKafkaConf :: (MonadIO m, HasKafkaConf k) => k -> m (Map String String) +dumpKafkaConf :: (MonadIO m, HasKafkaConf k) => k -> m (Map Text Text) dumpKafkaConf k = liftIO $ parseDump (rdKafkaConfDump (getRdKafkaConf k)) -parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String) +parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map Text Text) parseDump cstr = alloca $ \sizeptr -> do strPtr <- cstr sizeptr size <- peek sizeptr - keysAndValues <- mapM (peekCString <=< peekElemOff strPtr) [0..(fromIntegral size - 1)] + keysAndValues <- mapM (peekCText <=< peekElemOff strPtr) [0..(fromIntegral size - 1)] let ret = Map.fromList $ listToTuple keysAndValues rdKafkaConfDumpFree strPtr size return ret -listToTuple :: [String] -> [(String, String)] -listToTuple [] = [] -listToTuple (k:v:t) = (k, v) : listToTuple t -listToTuple _ = error "list to tuple can only be called on even length lists" +listToTuple :: [Text] -> [(Text, Text)] +listToTuple [] = [] +listToTuple (k:v:ts) = (k, v) : listToTuple ts +listToTuple _ = error "list to tuple can only be called on even length lists" diff --git a/src/Kafka/Internal/CancellationToken.hs b/src/Kafka/Internal/CancellationToken.hs --- a/src/Kafka/Internal/CancellationToken.hs +++ b/src/Kafka/Internal/CancellationToken.hs @@ -1,7 +1,13 @@ module Kafka.Internal.CancellationToken +( CancellationStatus(..) +, CancellationToken(..) +, newCancellationToken +, status +, cancel +) where -import Data.IORef +import Data.IORef (IORef, newIORef, readIORef, atomicWriteIORef) data CancellationStatus = Cancelled | Running deriving (Show, Eq) newtype CancellationToken = CancellationToken (IORef CancellationStatus) diff --git a/src/Kafka/Internal/RdKafka.chs b/src/Kafka/Internal/RdKafka.chs --- a/src/Kafka/Internal/RdKafka.chs +++ b/src/Kafka/Internal/RdKafka.chs @@ -3,15 +3,22 @@ module Kafka.Internal.RdKafka where -import Control.Monad -import Data.Word -import Foreign -import Foreign.C.Error -import Foreign.C.String -import Foreign.C.Types -import System.IO -import System.Posix.IO -import System.Posix.Types +import Data.Text (Text) +import qualified Data.Text as Text +import Control.Monad (liftM) +import Data.Int (Int32, Int64) +import Data.Word (Word8) +import Foreign.Marshal.Alloc (alloca, allocaBytes) +import Foreign.Marshal.Array (peekArray, allocaArray) +import Foreign.Storable (Storable(..)) +import Foreign.Ptr (Ptr, FunPtr, castPtr, nullPtr) +import Foreign.ForeignPtr (FinalizerPtr, addForeignPtrFinalizer, withForeignPtr, newForeignPtr, newForeignPtr_) +import Foreign.C.Error (Errno(..), getErrno) +import Foreign.C.String (CString, newCString, withCAString, peekCAString, peekCAStringLen, peekCString) +import Foreign.C.Types (CFile, CInt(..), CSize, CChar) +import System.IO (Handle, stdin, stdout, stderr) +import System.Posix.IO (handleToFd) +import System.Posix.Types (Fd(..)) #include @@ -55,8 +62,13 @@ {#fun pure rd_kafka_errno2err as ^ {`Int'} -> `RdKafkaRespErrT' cIntToEnum #} +peekCAText :: CString -> IO Text +peekCAText cp = Text.pack <$> peekCAString cp -kafkaErrnoString :: IO (String) +peekCText :: CString -> IO Text +peekCText cp = Text.pack <$> peekCString cp + +kafkaErrnoString :: IO String kafkaErrnoString = do (Errno num) <- getErrno return $ rdKafkaErr2str $ rdKafkaErrno2err (fromIntegral num) @@ -818,13 +830,13 @@ foreign import ccall unsafe "rdkafka.h &rd_kafka_destroy" rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ()) -newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either String RdKafkaTPtr) +newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either Text RdKafkaTPtr) newRdKafkaT kafkaType confPtr = allocaBytes nErrorBytes $ \charPtr -> do duper <- rdKafkaConfDup confPtr ret <- rdKafkaNew kafkaType duper charPtr (fromIntegral nErrorBytes) withForeignPtr ret $ \realPtr -> do - if realPtr == nullPtr then peekCString charPtr >>= return . Left + if realPtr == nullPtr then peekCText charPtr >>= return . Left else do addForeignPtrFinalizer rdKafkaDestroy ret return $ Right ret diff --git a/src/Kafka/Internal/Setup.hs b/src/Kafka/Internal/Setup.hs --- a/src/Kafka/Internal/Setup.hs +++ b/src/Kafka/Internal/Setup.hs @@ -1,20 +1,46 @@ -module Kafka.Internal.Setup where +module Kafka.Internal.Setup +( KafkaProps(..) +, TopicProps(..) +, Kafka(..) +, KafkaConf(..) +, TopicConf(..) +, HasKafka(..) +, HasKafkaConf(..) +, HasTopicConf(..) +, getRdKafka +, getRdKafkaConf +, getRdMsgQueue +, getRdTopicConf +, newTopicConf +, newKafkaConf +, kafkaConf +, topicConf +, checkConfSetValue +, setKafkaConfValue +, setAllKafkaConfValues +, setTopicConfValue +, setAllTopicConfValues +) +where -import Kafka.Internal.RdKafka -import Kafka.Types +import Kafka.Internal.RdKafka (RdKafkaConfResT(..), CCharBufPointer, RdKafkaQueueTPtr, RdKafkaTPtr, RdKafkaConfTPtr, RdKafkaTopicConfTPtr, nErrorBytes, rdKafkaTopicConfSet, newRdKafkaTopicConfT, newRdKafkaConfT, rdKafkaConfSet) +import Kafka.Types (KafkaError(..)) -import Control.Exception -import Control.Monad -import Data.IORef -import Foreign -import Foreign.C.String -import Kafka.Internal.CancellationToken +import Control.Exception (throw) +import Data.IORef (IORef, newIORef, readIORef) +import Foreign.Marshal.Alloc (allocaBytes) +import Foreign.C.String (peekCString) +import Kafka.Internal.CancellationToken (CancellationToken(..), newCancellationToken) +import qualified Data.Text as Text +import Data.Map (Map) +import Data.Text (Text) +import qualified Data.Map as Map -- -- Configuration -- -newtype KafkaProps = KafkaProps [(String, String)] deriving (Show, Eq) -newtype TopicProps = TopicProps [(String, String)] deriving (Show, Eq) +newtype KafkaProps = KafkaProps (Map Text Text) deriving (Show, Eq) +newtype TopicProps = TopicProps (Map Text Text) deriving (Show, Eq) newtype Kafka = Kafka RdKafkaTPtr deriving Show data KafkaConf = KafkaConf RdKafkaConfTPtr (IORef (Maybe RdKafkaQueueTPtr)) CancellationToken newtype TopicConf = TopicConf RdKafkaTopicConfTPtr deriving Show @@ -80,25 +106,25 @@ RdKafkaConfOk -> return () RdKafkaConfInvalid -> do str <- peekCString charPtr - throw $ KafkaInvalidConfigurationValue str + throw $ KafkaInvalidConfigurationValue (Text.pack str) RdKafkaConfUnknown -> do str <- peekCString charPtr - throw $ KafkaUnknownConfigurationKey str + throw $ KafkaUnknownConfigurationKey (Text.pack str) -setKafkaConfValue :: KafkaConf -> String -> String -> IO () +setKafkaConfValue :: KafkaConf -> Text -> Text -> IO () setKafkaConfValue (KafkaConf confPtr _ _) key value = allocaBytes nErrorBytes $ \charPtr -> do - err <- rdKafkaConfSet confPtr key value charPtr (fromIntegral nErrorBytes) + err <- rdKafkaConfSet confPtr (Text.unpack key) (Text.unpack value) charPtr (fromIntegral nErrorBytes) checkConfSetValue err charPtr setAllKafkaConfValues :: KafkaConf -> KafkaProps -> IO () -setAllKafkaConfValues conf (KafkaProps props) = forM_ props $ uncurry (setKafkaConfValue conf) +setAllKafkaConfValues conf (KafkaProps props) = Map.foldMapWithKey (setKafkaConfValue conf) props --forM_ props $ uncurry (setKafkaConfValue conf) -setTopicConfValue :: TopicConf -> String -> String -> IO () +setTopicConfValue :: TopicConf -> Text -> Text -> IO () setTopicConfValue (TopicConf confPtr) key value = allocaBytes nErrorBytes $ \charPtr -> do - err <- rdKafkaTopicConfSet confPtr key value charPtr (fromIntegral nErrorBytes) + err <- rdKafkaTopicConfSet confPtr (Text.unpack key) (Text.unpack value) charPtr (fromIntegral nErrorBytes) checkConfSetValue err charPtr setAllTopicConfValues :: TopicConf -> TopicProps -> IO () -setAllTopicConfValues conf (TopicProps props) = forM_ props $ uncurry (setTopicConfValue conf) +setAllTopicConfValues conf (TopicProps props) = Map.foldMapWithKey (setTopicConfValue conf) props --forM_ props $ uncurry (setTopicConfValue conf) diff --git a/src/Kafka/Internal/Shared.hs b/src/Kafka/Internal/Shared.hs --- a/src/Kafka/Internal/Shared.hs +++ b/src/Kafka/Internal/Shared.hs @@ -1,18 +1,40 @@ module Kafka.Internal.Shared +( runEventLoop +, pollEvents +, word8PtrToBS +, kafkaRespErr +, throwOnError +, hasError +, rdKafkaErrorToEither +, kafkaErrorToEither +, kafkaErrorToMaybe +, maybeToLeft +, readPayload +, readTopic +, readKey +, readTimestamp +, readBS +) where +import Data.Text (Text) +import qualified Data.Text as Text import Control.Concurrent (forkIO, rtsSupportsBoundThreads) -import Control.Exception +import Control.Exception (throw) 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 Data.Word (Word8) +import Foreign.Ptr (Ptr, nullPtr) +import Foreign.Marshal.Alloc (alloca) +import Foreign.ForeignPtr (newForeignPtr_) +import Foreign.Storable (Storable(peek)) +import Foreign.C.Error (Errno(..)) +import Kafka.Consumer.Types (Timestamp(..)) import Kafka.Internal.CancellationToken as CToken -import Kafka.Internal.RdKafka -import Kafka.Internal.Setup -import Kafka.Types +import Kafka.Internal.RdKafka (RdKafkaTimestampTypeT(..), RdKafkaMessageTPtr, RdKafkaMessageT(..), RdKafkaRespErrT(..), Word8Ptr, rdKafkaPoll, rdKafkaErrno2err, rdKafkaTopicName, rdKafkaMessageTimestamp) +import Kafka.Internal.Setup (HasKafka(..), Kafka(..)) +import Kafka.Types (KafkaError(..), Timeout(..), Millis(..)) runEventLoop :: HasKafka a => a -> CancellationToken -> Maybe Timeout -> IO () runEventLoop k ct timeout = @@ -38,7 +60,7 @@ kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num) {-# INLINE kafkaRespErr #-} -throwOnError :: IO (Maybe String) -> IO () +throwOnError :: IO (Maybe Text) -> IO () throwOnError action = do m <- action case m of @@ -76,8 +98,8 @@ readPayload :: RdKafkaMessageT -> IO (Maybe BS.ByteString) readPayload = readBS len'RdKafkaMessageT payload'RdKafkaMessageT -readTopic :: RdKafkaMessageT -> IO String -readTopic msg = newForeignPtr_ (topic'RdKafkaMessageT msg) >>= rdKafkaTopicName +readTopic :: RdKafkaMessageT -> IO Text +readTopic msg = newForeignPtr_ (topic'RdKafkaMessageT msg) >>= (fmap Text.pack . rdKafkaTopicName) readKey :: RdKafkaMessageT -> IO (Maybe BSI.ByteString) readKey = readBS keyLen'RdKafkaMessageT key'RdKafkaMessageT diff --git a/src/Kafka/Metadata.hs b/src/Kafka/Metadata.hs --- a/src/Kafka/Metadata.hs +++ b/src/Kafka/Metadata.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} module Kafka.Metadata ( KafkaMetadata(..), BrokerMetadata(..), TopicMetadata(..), PartitionMetadata(..) , WatermarkOffsets(..) @@ -12,33 +14,36 @@ ) where -import Control.Arrow (left) -import Control.Exception (bracket) -import Control.Monad.IO.Class (MonadIO, liftIO) -import Data.Bifunctor -import Data.ByteString (ByteString, pack) -import Data.Monoid ((<>)) -import qualified Data.Set as S -import Foreign -import Foreign.C.String -import Kafka.Consumer.Convert -import Kafka.Consumer.Types -import Kafka.Internal.RdKafka -import Kafka.Internal.Setup -import Kafka.Internal.Shared -import Kafka.Types +import Control.Arrow (left) +import Control.Exception (bracket) +import Control.Monad.IO.Class (MonadIO (liftIO)) +import Data.Bifunctor (bimap) +import Data.ByteString (ByteString, pack) +import Data.Monoid ((<>)) +import Data.Text (Text) +import Foreign (Storable (peek), peekArray, withForeignPtr) +import GHC.Generics (Generic) +import Kafka.Consumer.Convert (fromNativeTopicPartitionList'', toNativeTopicPartitionList) +import Kafka.Consumer.Types (ConsumerGroupId (..), Offset (..), PartitionOffset (..), TopicPartition (..)) +import Kafka.Internal.RdKafka (RdKafkaGroupInfoT (..), RdKafkaGroupListT (..), RdKafkaGroupListTPtr, RdKafkaGroupMemberInfoT (..), RdKafkaMetadataBrokerT (..), RdKafkaMetadataPartitionT (..), RdKafkaMetadataT (..), RdKafkaMetadataTPtr, RdKafkaMetadataTopicT (..), RdKafkaRespErrT (..), RdKafkaTPtr, destroyUnmanagedRdKafkaTopic, newUnmanagedRdKafkaTopicT, peekCAText, rdKafkaListGroups, rdKafkaMetadata, rdKafkaOffsetsForTimes, rdKafkaQueryWatermarkOffsets) +import Kafka.Internal.Setup (HasKafka (..), Kafka (..)) +import Kafka.Internal.Shared (kafkaErrorToMaybe) +import Kafka.Types (BrokerId (..), ClientId (..), KafkaError (..), Millis (..), PartitionId (..), Timeout (..), TopicName (..)) +import qualified Data.Set as S +import qualified Data.Text as Text + data KafkaMetadata = KafkaMetadata { kmBrokers :: [BrokerMetadata] , kmTopics :: [TopicMetadata] , kmOrigBroker :: !BrokerId - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) data BrokerMetadata = BrokerMetadata { bmBrokerId :: !BrokerId - , bmBrokerHost :: !String + , bmBrokerHost :: !Text , bmBrokerPort :: !Int - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) data PartitionMetadata = PartitionMetadata { pmPartitionId :: !PartitionId @@ -46,39 +51,39 @@ , pmLeader :: !BrokerId , pmReplicas :: [BrokerId] , pmInSyncReplicas :: [BrokerId] - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) data TopicMetadata = TopicMetadata { tmTopicName :: !TopicName , tmPartitions :: [PartitionMetadata] , tmError :: Maybe KafkaError - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) data WatermarkOffsets = WatermarkOffsets { woTopicName :: !TopicName , woPartitionId :: !PartitionId , woLowWatermark :: !Offset , woHighWatermark :: !Offset - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) -newtype GroupMemberId = GroupMemberId String deriving (Show, Eq, Read, Ord) +newtype GroupMemberId = GroupMemberId Text deriving (Show, Eq, Read, Ord) data GroupMemberInfo = GroupMemberInfo { gmiMemberId :: !GroupMemberId , gmiClientId :: !ClientId - , gmiClientHost :: !String + , gmiClientHost :: !Text , gmiMetadata :: !ByteString , gmiAssignment :: !ByteString - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) -newtype GroupProtocolType = GroupProtocolType String deriving (Show, Eq, Read, Ord) -newtype GroupProtocol = GroupProtocol String deriving (Show, Eq, Read, Ord) +newtype GroupProtocolType = GroupProtocolType Text deriving (Show, Eq, Read, Ord, Generic) +newtype GroupProtocol = GroupProtocol Text deriving (Show, Eq, Read, Ord, Generic) data GroupState = GroupPreparingRebalance -- ^ Group is preparing to rebalance | GroupEmpty -- ^ Group has no more members, but lingers until all offsets have expired | GroupAwaitingSync -- ^ Group is awaiting state assignment from the leader | GroupStable -- ^ Group is stable | GroupDead -- ^ Group has no more members and its metadata is being removed - deriving (Show, Eq, Read, Ord) + deriving (Show, Eq, Read, Ord, Generic) data GroupInfo = GroupInfo { giGroup :: !ConsumerGroupId @@ -87,7 +92,7 @@ , giProtocolType :: !GroupProtocolType , giProtocol :: !GroupProtocol , giMembers :: [GroupMemberInfo] - } deriving (Show, Eq) + } deriving (Show, Eq, Generic) -- | Returns metadata for all topics in the cluster allTopicsMetadata :: (MonadIO m, HasKafka k) => k -> Timeout -> m (Either KafkaError KafkaMetadata) @@ -99,12 +104,12 @@ topicMetadata :: (MonadIO m, HasKafka k) => k -> Timeout -> TopicName -> m (Either KafkaError KafkaMetadata) topicMetadata k (Timeout timeout) (TopicName tn) = liftIO $ bracket mkTopic clTopic $ \mbt -> case mbt of - Left err -> return (Left $ KafkaError err) + Left err -> return (Left $ KafkaError (Text.pack err)) Right t -> do meta <- rdKafkaMetadata (getKafkaPtr k) False (Just t) timeout traverse fromKafkaMetadataPtr (left KafkaResponseError meta) where - mkTopic = newUnmanagedRdKafkaTopicT (getKafkaPtr k) tn Nothing + mkTopic = newUnmanagedRdKafkaTopicT (getKafkaPtr k) (Text.unpack tn) Nothing clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic -- | Query broker for low (oldest/beginning) and high (newest/end) offsets for a given topic. @@ -126,7 +131,7 @@ -- | Query broker for low (oldest/beginning) and high (newest/end) offsets for a specific partition partitionWatermarkOffsets :: (MonadIO m, HasKafka k) => k -> Timeout -> TopicName -> PartitionId -> m (Either KafkaError WatermarkOffsets) partitionWatermarkOffsets k (Timeout timeout) (TopicName t) (PartitionId p) = liftIO $ do - offs <- rdKafkaQueryWatermarkOffsets (getKafkaPtr k) t p timeout + offs <- rdKafkaQueryWatermarkOffsets (getKafkaPtr k) (Text.unpack t) p timeout return $ bimap KafkaResponseError toWatermark offs where toWatermark (l, h) = WatermarkOffsets (TopicName t) (PartitionId p) (Offset l) (Offset h) @@ -181,7 +186,7 @@ -- | Describe a given consumer group. consumerGroupInfo :: (MonadIO m, HasKafka k) => k -> Timeout -> ConsumerGroupId -> m (Either KafkaError [GroupInfo]) consumerGroupInfo k (Timeout timeout) (ConsumerGroupId gn) = liftIO $ do - res <- rdKafkaListGroups (getKafkaPtr k) (Just gn) timeout + res <- rdKafkaListGroups (getKafkaPtr k) (Just (Text.unpack gn)) timeout traverse fromGroupInfoListPtr (left KafkaResponseError res) ------------------------------------------------------------------------------- @@ -201,10 +206,10 @@ fromGroupInfoPtr gi = do --bmd <- peek (broker'RdKafkaGroupInfoT gi) -- >>= fromBrokerMetadataPtr --xxx <- fromBrokerMetadataPtr bmd - cid <- peekCAString $ group'RdKafkaGroupInfoT gi - stt <- peekCAString $ state'RdKafkaGroupInfoT gi - prt <- peekCAString $ protocolType'RdKafkaGroupInfoT gi - pr <- peekCAString $ protocol'RdKafkaGroupInfoT gi + cid <- peekCAText $ group'RdKafkaGroupInfoT gi + stt <- peekCAText $ state'RdKafkaGroupInfoT gi + prt <- peekCAText $ protocolType'RdKafkaGroupInfoT gi + pr <- peekCAText $ protocol'RdKafkaGroupInfoT gi mbs <- peekArray (memberCnt'RdKafkaGroupInfoT gi) (members'RdKafkaGroupInfoT gi) mbl <- mapM fromGroupMemberInfoPtr mbs return GroupInfo @@ -219,9 +224,9 @@ fromGroupMemberInfoPtr :: RdKafkaGroupMemberInfoT -> IO GroupMemberInfo fromGroupMemberInfoPtr mi = do - mid <- peekCAString $ memberId'RdKafkaGroupMemberInfoT mi - cid <- peekCAString $ clientId'RdKafkaGroupMemberInfoT mi - hst <- peekCAString $ clientHost'RdKafkaGroupMemberInfoT mi + mid <- peekCAText $ memberId'RdKafkaGroupMemberInfoT mi + cid <- peekCAText $ clientId'RdKafkaGroupMemberInfoT mi + hst <- peekCAText $ clientHost'RdKafkaGroupMemberInfoT mi mtd <- peekArray (memberMetadataSize'RdKafkaGroupMemberInfoT mi) (memberMetadata'RdKafkaGroupMemberInfoT mi) ass <- peekArray (memberAssignmentSize'RdKafkaGroupMemberInfoT mi) (memberAssignment'RdKafkaGroupMemberInfoT mi) return GroupMemberInfo @@ -234,7 +239,7 @@ fromTopicMetadataPtr :: RdKafkaMetadataTopicT -> IO TopicMetadata fromTopicMetadataPtr tm = do - tnm <- peekCAString (topic'RdKafkaMetadataTopicT tm) + tnm <- peekCAText (topic'RdKafkaMetadataTopicT tm) pts <- peekArray (partitionCnt'RdKafkaMetadataTopicT tm) (partitions'RdKafkaMetadataTopicT tm) pms <- mapM fromPartitionMetadataPtr pts return TopicMetadata @@ -258,7 +263,7 @@ fromBrokerMetadataPtr :: RdKafkaMetadataBrokerT -> IO BrokerMetadata fromBrokerMetadataPtr bm = do - host <- peekCAString (host'RdKafkaMetadataBrokerT bm) + host <- peekCAText (host'RdKafkaMetadataBrokerT bm) return BrokerMetadata { bmBrokerId = BrokerId (id'RdKafkaMetadataBrokerT bm) , bmBrokerHost = host @@ -280,11 +285,11 @@ , kmOrigBroker = BrokerId $ fromIntegral (origBrokerId'RdKafkaMetadataT km) } -groupStateFromKafkaString :: String -> GroupState +groupStateFromKafkaString :: Text -> GroupState groupStateFromKafkaString s = case s of "PreparingRebalance" -> GroupPreparingRebalance "AwaitingSync" -> GroupAwaitingSync "Stable" -> GroupStable "Dead" -> GroupDead "Empty" -> GroupEmpty - _ -> error $ "Unknown group state: " <> s + _ -> error $ "Unknown group state: " <> (Text.unpack s) diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -12,23 +12,36 @@ where import Control.Arrow ((&&&)) -import Control.Exception -import Control.Monad -import Control.Monad.IO.Class +import Control.Exception (bracket) +import Control.Monad (forM, forM_, (<=<)) +import Control.Monad.IO.Class (MonadIO(liftIO)) import qualified Data.ByteString as BS import qualified Data.ByteString.Internal as BSI import Data.Function (on) import Data.List (groupBy, sortBy) -import qualified Data.Map as M import Data.Ord (comparing) -import Foreign hiding (void) +import qualified Data.Text as Text +import Foreign.ForeignPtr (withForeignPtr, newForeignPtr_) +import Foreign.Marshal.Array (withArrayLen) +import Foreign.Ptr (Ptr, nullPtr, plusPtr) +import Foreign.Storable (Storable(..)) import Kafka.Internal.CancellationToken as CToken import Kafka.Internal.RdKafka -import Kafka.Internal.Setup -import Kafka.Internal.Shared -import Kafka.Producer.Callbacks -import Kafka.Producer.Convert -import Kafka.Producer.Types + ( RdKafkaMessageT(..) + , RdKafkaTypeT(..) + , RdKafkaRespErrT(..) + , newRdKafkaT + , newUnmanagedRdKafkaTopicT + , destroyUnmanagedRdKafkaTopic + , rdKafkaProduce + , rdKafkaSetLogLevel + , rdKafkaProduceBatch + , rdKafkaOutqLen + ) +import Kafka.Internal.Setup (KafkaConf(..), Kafka(..), TopicConf(..), kafkaConf, KafkaProps(..), topicConf, TopicProps(..)) +import Kafka.Internal.Shared (pollEvents) +import Kafka.Producer.Convert (copyMsgFlags, producePartitionCInt, producePartitionInt, handleProduceErr) +import Kafka.Producer.Types (KafkaProducer(..)) import Kafka.Producer.ProducerProperties as X import Kafka.Producer.Types as X hiding (KafkaProducer) @@ -56,8 +69,8 @@ -- A newly created producer must be closed with 'closeProducer' function. newProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError KafkaProducer) newProducer pps = liftIO $ do - kc@(KafkaConf kc' _ _) <- kafkaConf (KafkaProps $ M.toList (ppKafkaProps pps)) - tc <- topicConf (TopicProps $ M.toList (ppTopicProps pps)) + kc@(KafkaConf kc' _ _) <- kafkaConf (KafkaProps $ (ppKafkaProps pps)) + tc <- topicConf (TopicProps $ (ppTopicProps pps)) -- set callbacks forM_ (ppCallbacks pps) (\setCb -> setCb kc) @@ -81,11 +94,11 @@ pollEvents kp (Just $ Timeout 0) -- fire callbacks if any exist (handle delivery reports) bracket (mkTopic $ prTopic m) clTopic withTopic where - mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k tn (Just tc) + mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k (Text.unpack tn) (Just tc) clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic - withTopic (Left err) = return . Just . KafkaError $ err + withTopic (Left err) = return . Just . KafkaError $ Text.pack err withTopic (Right t) = withBS (prValue m) $ \payloadPtr payloadLength -> withBS (prKey m) $ \keyPtr keyLength -> @@ -112,14 +125,14 @@ mkSortKey = prTopic &&& prPartition mkBatches = groupBy ((==) `on` mkSortKey) . sortBy (comparing mkSortKey) - mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k tn (Just tc) + mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k (Text.unpack tn) (Just tc) clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic sendBatch [] = return [] sendBatch batch = bracket (mkTopic $ prTopic (head batch)) clTopic (withTopic batch) - withTopic ms (Left err) = return $ (, KafkaError err) <$> ms + withTopic ms (Left err) = return $ (, KafkaError (Text.pack err)) <$> ms withTopic ms (Right t) = do let (partInt, partCInt) = (producePartitionInt &&& producePartitionCInt) $ prPartition (head ms) withForeignPtr t $ \topicPtr -> do diff --git a/src/Kafka/Producer/Callbacks.hs b/src/Kafka/Producer/Callbacks.hs --- a/src/Kafka/Producer/Callbacks.hs +++ b/src/Kafka/Producer/Callbacks.hs @@ -4,15 +4,16 @@ ) where -import Foreign -import Foreign.C.Error +import Foreign.C.Error (getErrno) +import Foreign.Ptr (Ptr, nullPtr) +import Foreign.Storable (Storable(peek)) 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 +import Kafka.Consumer.Types (Offset(..)) +import Kafka.Internal.RdKafka (RdKafkaMessageT(..), RdKafkaRespErrT(..), rdKafkaConfSetDrMsgCb) +import Kafka.Internal.Setup (KafkaConf(..), getRdKafkaConf) +import Kafka.Internal.Shared (kafkaRespErr, readTopic, readKey, readPayload) +import Kafka.Producer.Types (ProducerRecord(..), DeliveryReport(..), ProducePartition(..)) +import Kafka.Types (KafkaError(..), TopicName(..)) -- | Sets the callback for delivery reports. deliveryCallback :: (DeliveryReport -> IO ()) -> KafkaConf -> IO () diff --git a/src/Kafka/Producer/Convert.hs b/src/Kafka/Producer/Convert.hs --- a/src/Kafka/Producer/Convert.hs +++ b/src/Kafka/Producer/Convert.hs @@ -1,12 +1,17 @@ module Kafka.Producer.Convert +( copyMsgFlags +, producePartitionInt +, producePartitionCInt +, handleProduceErr +) where -import Foreign.C.Error -import Foreign.C.Types -import Kafka.Internal.RdKafka -import Kafka.Internal.Shared -import Kafka.Types -import Kafka.Producer.Types +import Foreign.C.Error (getErrno) +import Foreign.C.Types (CInt) +import Kafka.Internal.RdKafka (rdKafkaMsgFlagCopy) +import Kafka.Internal.Shared (kafkaRespErr) +import Kafka.Types (KafkaError(..)) +import Kafka.Producer.Types (ProducePartition(..)) copyMsgFlags :: Int copyMsgFlags = rdKafkaMsgFlagCopy diff --git a/src/Kafka/Producer/ProducerProperties.hs b/src/Kafka/Producer/ProducerProperties.hs --- a/src/Kafka/Producer/ProducerProperties.hs +++ b/src/Kafka/Producer/ProducerProperties.hs @@ -1,22 +1,36 @@ +{-# LANGUAGE OverloadedStrings #-} + module Kafka.Producer.ProducerProperties -( module Kafka.Producer.ProducerProperties +( ProducerProperties(..) +, brokersList +, setCallback +, logLevel +, compression +, topicCompression +, sendTimeout +, extraProps +, suppressDisconnectLogs +, extraTopicProps +, debugOptions , module Kafka.Producer.Callbacks ) where -import Control.Monad -import qualified Data.List as L +import Data.Text (Text) +import qualified Data.Text as Text +import Control.Monad (MonadPlus(mplus)) import Data.Map (Map) import qualified Data.Map as M import Data.Semigroup as Sem -import Kafka.Internal.Setup +import Kafka.Internal.Setup (KafkaConf(..)) +import Kafka.Types (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText) + import Kafka.Producer.Callbacks -import Kafka.Types -- | Properties to create 'KafkaProducer'. data ProducerProperties = ProducerProperties - { ppKafkaProps :: Map String String - , ppTopicProps :: Map String String + { ppKafkaProps :: Map Text Text + , ppTopicProps :: Map Text Text , ppLogLevel :: Maybe KafkaLogLevel , ppCallbacks :: [KafkaConf -> IO ()] } @@ -40,7 +54,7 @@ brokersList :: [BrokerAddress] -> ProducerProperties brokersList bs = - let bs' = L.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) + let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) in extraProps $ M.fromList [("bootstrap.servers", bs')] setCallback :: (KafkaConf -> IO ()) -> ProducerProperties @@ -53,19 +67,19 @@ compression :: KafkaCompressionCodec -> ProducerProperties compression c = - extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToString c) + extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) topicCompression :: KafkaCompressionCodec -> ProducerProperties topicCompression c = - extraTopicProps $ M.singleton "compression.codec" (kafkaCompressionCodecToString c) + extraTopicProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) sendTimeout :: Timeout -> ProducerProperties sendTimeout (Timeout t) = - extraTopicProps $ M.singleton "message.timeout.ms" (show t) + extraTopicProps $ M.singleton "message.timeout.ms" (Text.pack $ show t) -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found -extraProps :: Map String String -> ProducerProperties +extraProps :: Map Text Text -> ProducerProperties extraProps m = mempty { ppKafkaProps = m } -- | Suppresses producer disconnects logs. @@ -78,7 +92,7 @@ -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found -extraTopicProps :: Map String String -> ProducerProperties +extraTopicProps :: Map Text Text -> ProducerProperties extraTopicProps m = mempty { ppTopicProps = m } -- | Sets debug features for the producer @@ -86,5 +100,5 @@ debugOptions :: [KafkaDebug] -> ProducerProperties debugOptions [] = extraProps M.empty debugOptions d = - let points = L.intercalate "," (kafkaDebugToString <$> d) + let points = Text.intercalate "," (kafkaDebugToText <$> d) in extraProps $ M.fromList [("debug", points)] diff --git a/src/Kafka/Producer/Types.hs b/src/Kafka/Producer/Types.hs --- a/src/Kafka/Producer/Types.hs +++ b/src/Kafka/Producer/Types.hs @@ -1,13 +1,19 @@ {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} module Kafka.Producer.Types - +( KafkaProducer(..) +, ProducerRecord(..) +, ProducePartition(..) +, DeliveryReport(..) +) where -import qualified Data.ByteString as BS -import Data.Typeable -import Kafka.Internal.Setup -import Kafka.Types -import Kafka.Consumer.Types +import Data.ByteString +import Data.Typeable (Typeable) +import GHC.Generics (Generic) +import Kafka.Consumer.Types (Offset (..)) +import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), HasTopicConf (..), Kafka (..), KafkaConf (..), TopicConf (..)) +import Kafka.Types (KafkaError (..), TopicName (..)) -- | Main pointer to Kafka object, which contains our brokers data KafkaProducer = KafkaProducer @@ -32,33 +38,17 @@ data ProducerRecord = ProducerRecord { prTopic :: !TopicName , prPartition :: !ProducePartition - , prKey :: Maybe BS.ByteString - , prValue :: Maybe BS.ByteString - } deriving (Eq, Show, Typeable) + , prKey :: Maybe ByteString + , prValue :: Maybe ByteString + } deriving (Eq, Show, Typeable, Generic) data ProducePartition = SpecifiedPartition {-# UNPACK #-} !Int -- the partition number of the topic | UnassignedPartition - deriving (Show, Eq, Ord, Typeable) + deriving (Show, Eq, Ord, Typeable, Generic) 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) + deriving (Show, Eq, Generic) diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs --- a/src/Kafka/Types.hs +++ b/src/Kafka/Types.hs @@ -1,20 +1,46 @@ {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} module Kafka.Types +( BrokerId(..) +, PartitionId(..) +, Millis(..) +, ClientId(..) +, BatchSize(..) +, TopicName(..) +, BrokerAddress(..) +, Timeout(..) +, KafkaLogLevel(..) +, KafkaError(..) +, KafkaDebug(..) +, KafkaCompressionCodec(..) +, TopicType(..) +, topicType +, kafkaDebugToText +, kafkaCompressionCodecToText +) where -import Control.Exception -import Data.Int -import Data.Typeable -import Kafka.Internal.RdKafka +import Control.Exception (Exception (..)) +import Data.Int (Int64) +import Data.Text (Text, isPrefixOf) +import Data.Typeable (Typeable) +import GHC.Generics (Generic) +import Kafka.Internal.RdKafka (RdKafkaRespErrT, rdKafkaErr2name, rdKafkaErr2str) -newtype BrokerId = BrokerId { unBrokerId :: Int} deriving (Show, Eq, Ord, Read) +newtype BrokerId = BrokerId { unBrokerId :: Int } deriving (Show, Eq, Ord, Read, Generic) -newtype PartitionId = PartitionId { unPartitionId :: Int} deriving (Show, Eq, Read, Ord, Enum) -newtype Millis = Millis { unMillis :: Int64 } deriving (Show, Read, Eq, Ord, Num) -newtype ClientId = ClientId { unClientId :: String} deriving (Show, Eq, Ord) -newtype BatchSize = BatchSize { unBatchSize :: Int } deriving (Show, Read, Eq, Ord, Num) +newtype PartitionId = PartitionId { unPartitionId :: Int } deriving (Show, Eq, Read, Ord, Enum, Generic) +newtype Millis = Millis { unMillis :: Int64 } deriving (Show, Read, Eq, Ord, Num, Generic) +newtype ClientId = ClientId { unClientId :: Text } deriving (Show, Eq, Ord, Generic) +newtype BatchSize = BatchSize { unBatchSize :: Int } deriving (Show, Read, Eq, Ord, Num, Generic) +data TopicType = + User -- ^ Normal topics that are created by user. + | System -- ^ Topics starting with "__" (__consumer_offsets, __confluent.support.metrics) are considered "system" topics + deriving (Show, Read, Eq, Ord, Generic) + -- | Topic name to be consumed -- -- Wildcard (regex) topics are supported by the /librdkafka/ assignor: @@ -22,52 +48,37 @@ -- be regex-matched to the full list of topics in the cluster and matching -- topics will be added to the subscription list. newtype TopicName = - TopicName { unTopicName :: String } -- ^ a simple topic name or a regex if started with @^@ - deriving (Show, Eq, Ord, Read) + TopicName { unTopicName :: Text } -- ^ a simple topic name or a regex if started with @^@ + deriving (Show, Eq, Ord, Read, Generic) +topicType :: TopicName -> TopicType +topicType (TopicName tn) = + if "__" `isPrefixOf` tn then System else User +{-# INLINE topicType #-} + -- | Kafka broker address string (e.g. @broker1:9092@) -newtype BrokerAddress = BrokerAddress { unBrokerAddress :: String } deriving (Show, Eq) +newtype BrokerAddress = BrokerAddress { unBrokerAddress :: Text } deriving (Show, Eq, Generic) -- | Timeout in milliseconds -newtype Timeout = Timeout { unTimeout :: Int } deriving (Show, Eq, Read) +newtype Timeout = Timeout { unTimeout :: Int } deriving (Show, Eq, Read, Generic) -- | Log levels for /librdkafka/. data KafkaLogLevel = KafkaLogEmerg | KafkaLogAlert | KafkaLogCrit | KafkaLogErr | KafkaLogWarning | KafkaLogNotice | KafkaLogInfo | KafkaLogDebug - deriving (Show, Eq) - -instance Enum KafkaLogLevel where - toEnum 0 = KafkaLogEmerg - toEnum 1 = KafkaLogAlert - toEnum 2 = KafkaLogCrit - toEnum 3 = KafkaLogErr - toEnum 4 = KafkaLogWarning - toEnum 5 = KafkaLogNotice - toEnum 6 = KafkaLogInfo - toEnum 7 = KafkaLogDebug - toEnum _ = undefined - - fromEnum KafkaLogEmerg = 0 - fromEnum KafkaLogAlert = 1 - fromEnum KafkaLogCrit = 2 - fromEnum KafkaLogErr = 3 - fromEnum KafkaLogWarning = 4 - fromEnum KafkaLogNotice = 5 - fromEnum KafkaLogInfo = 6 - fromEnum KafkaLogDebug = 7 + deriving (Show, Enum, Eq) -- -- | Any Kafka errors data KafkaError = - KafkaError String + KafkaError Text | KafkaInvalidReturnValue - | KafkaBadSpecification String + | KafkaBadSpecification Text | KafkaResponseError RdKafkaRespErrT - | KafkaInvalidConfigurationValue String - | KafkaUnknownConfigurationKey String + | KafkaInvalidConfigurationValue Text + | KafkaUnknownConfigurationKey Text | KafkaBadConfiguration - deriving (Eq, Show, Typeable) + deriving (Eq, Show, Typeable, Generic) instance Exception KafkaError where displayException (KafkaResponseError err) = @@ -87,10 +98,10 @@ | DebugFetch | DebugFeature | DebugAll - deriving (Eq, Show, Typeable) + deriving (Eq, Show, Typeable, Generic) -kafkaDebugToString :: KafkaDebug -> String -kafkaDebugToString d =case d of +kafkaDebugToText :: KafkaDebug -> Text +kafkaDebugToText d = case d of DebugGeneric -> "generic" DebugBroker -> "broker" DebugTopic -> "topic" @@ -109,10 +120,10 @@ | Gzip | Snappy | Lz4 - deriving (Eq, Show, Typeable) + deriving (Eq, Show, Typeable, Generic) -kafkaCompressionCodecToString :: KafkaCompressionCodec -> String -kafkaCompressionCodecToString c = case c of +kafkaCompressionCodecToText :: KafkaCompressionCodec -> Text +kafkaCompressionCodecToText c = case c of NoCompression -> "none" Gzip -> "gzip" Snappy -> "snappy" diff --git a/tests-it/Kafka/IntegrationSpec.hs b/tests-it/Kafka/IntegrationSpec.hs --- a/tests-it/Kafka/IntegrationSpec.hs +++ b/tests-it/Kafka/IntegrationSpec.hs @@ -8,7 +8,7 @@ import Control.Monad.Loops import qualified Data.ByteString as BS import Data.Either -import Data.Map +import Data.Map (fromList) import Data.Monoid ((<>)) import Kafka.Consumer as C @@ -151,8 +151,10 @@ it "should return all topics metadata" $ \k -> do res <- allTopicsMetadata k (Timeout 1000) res `shouldSatisfy` isRight - (length . kmBrokers) <$> res `shouldBe` Right 1 - (length . kmTopics) <$> res `shouldBe` Right 2 + let filterUserTopics m = m { kmTopics = filter (\t -> topicType (tmTopicName t) == User) (kmTopics m) } + let res' = fmap filterUserTopics res + length . kmBrokers <$> res' `shouldBe` Right 1 + length . kmTopics <$> res' `shouldBe` Right 1 it "should return topic metadata" $ \k -> do res <- topicMetadata k (Timeout 1000) testTopic diff --git a/tests-it/Kafka/TestEnv.hs b/tests-it/Kafka/TestEnv.hs --- a/tests-it/Kafka/TestEnv.hs +++ b/tests-it/Kafka/TestEnv.hs @@ -8,6 +8,7 @@ import Data.Monoid ((<>)) import System.Environment import System.IO.Unsafe +import qualified Data.Text as Text import Control.Concurrent import Kafka.Consumer as C @@ -16,13 +17,13 @@ import Test.Hspec brokerAddress :: BrokerAddress -brokerAddress = unsafePerformIO $ - BrokerAddress <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> (return "localhost:9092") +brokerAddress = unsafePerformIO $ do + (BrokerAddress . Text.pack) <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> (return "localhost:9092") {-# NOINLINE brokerAddress #-} testTopic :: TopicName -testTopic = unsafePerformIO $ - TopicName <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> (return "kafka-client_tests") +testTopic = unsafePerformIO $ do + (TopicName . Text.pack) <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> (return "kafka-client_tests") {-# NOINLINE testTopic #-} testGroupId :: ConsumerGroupId @@ -48,9 +49,7 @@ <> offsetReset Earliest mkProducer :: IO KafkaProducer -mkProducer = do - (Right p) <- newProducer producerProps - return p +mkProducer = newProducer producerProps >>= \(Right p) -> pure p mkConsumerWith :: ConsumerProperties -> IO KafkaConsumer mkConsumerWith props = do diff --git a/tests/Kafka/Consumer/ConsumerRecordMapSpec.hs b/tests/Kafka/Consumer/ConsumerRecordMapSpec.hs --- a/tests/Kafka/Consumer/ConsumerRecordMapSpec.hs +++ b/tests/Kafka/Consumer/ConsumerRecordMapSpec.hs @@ -4,15 +4,16 @@ ) where import Data.Bitraversable +import Data.Text import Kafka.Consumer.Types import Kafka.Types import Test.Hspec -testKey, testValue :: String +testKey, testValue :: Text testKey = "some-key" testValue = "some-value" -testRecord :: ConsumerRecord (Maybe String) (Maybe String) +testRecord :: ConsumerRecord (Maybe Text) (Maybe Text) testRecord = ConsumerRecord { crTopic = TopicName "some-topic" , crPartition = PartitionId 0 @@ -25,7 +26,7 @@ spec :: Spec spec = describe "Kafka.Consumer.ConsumerRecordSpec" $ do it "should exract key" $ - sequenceFirst testRecord `shouldBe` Just (crMapKey (const testKey) testRecord) + bitraverse id pure testRecord `shouldBe` Just (crMapKey (const testKey) testRecord) it "should extract value" $ sequence testRecord `shouldBe` Just (crMapValue (const testValue) testRecord) diff --git a/tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs b/tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs --- a/tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs +++ b/tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs @@ -1,19 +1,21 @@ {-# LANGUAGE OverloadedStrings #-} + module Kafka.Consumer.ConsumerRecordTraverseSpec ( spec ) where import Data.Bifunctor import Data.Bitraversable +import Data.Text import Kafka.Consumer.Types import Kafka.Types import Test.Hspec -testKey, testValue :: String +testKey, testValue :: Text testKey = "some-key" testValue = "some-value" -testRecord :: ConsumerRecord String String +testRecord :: ConsumerRecord Text Text testRecord = ConsumerRecord { crTopic = TopicName "some-topic" , crPartition = PartitionId 0 @@ -29,18 +31,6 @@ liftNothing :: a -> Maybe a liftNothing _ = Nothing -liftValueM :: a -> Maybe (Either String a) -liftValueM = pure . pure - -liftNothingM :: a -> Maybe (Either String a) -liftNothingM _ = Nothing - -testError :: Either String a -testError = Left "test error" - -liftErrorM :: a -> Maybe (Either String a) -liftErrorM _ = Just testError - spec :: Spec spec = describe "Kafka.Consumer.ConsumerRecordTraverseSpec" $ do it "should sequence" $ do @@ -65,23 +55,3 @@ bitraverse liftNothing liftValue testRecord `shouldBe` Nothing bitraverse liftValue liftNothing testRecord `shouldBe` Nothing - it "should traverse key (monadic)" $ - traverseFirstM liftValueM testRecord `shouldBe` Just (Right testRecord) - - it "should traverse value (monadic)" $ - traverseM liftValueM testRecord `shouldBe` Just (Right testRecord) - - it "should traverse KV (monadic)" $ - bitraverseM liftValueM liftValueM testRecord `shouldBe` Just (Right testRecord) - - it "should traverse and report error (monadic)" $ do - traverseFirstM liftErrorM testRecord `shouldBe` Just testError - traverseM liftErrorM testRecord `shouldBe` Just testError - bitraverseM liftValueM liftErrorM testRecord `shouldBe` Just testError - bitraverseM liftErrorM liftValueM testRecord `shouldBe` Just testError - - it "should traverse and report empty container (monadic)" $ do - traverseFirstM liftNothingM testRecord `shouldBe` Nothing - traverseM liftNothingM testRecord `shouldBe` Nothing - bitraverseM liftValueM liftNothingM testRecord `shouldBe` Nothing - bitraverseM liftNothingM liftValueM testRecord `shouldBe` Nothing