packages feed

hw-kafka-client 2.5.0 → 2.6.0

raw patch · 27 files changed

+640/−370 lines, 27 filesdep +text

Dependencies added: text

Files

README.md view
@@ -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
example/ConsumerExample.hs view
@@ -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 ())
example/ProducerExample.hs view
@@ -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
hw-kafka-client.cabal view
@@ -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
src/Kafka/Callbacks.hs view
@@ -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 =
src/Kafka/Consumer.hs view
@@ -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
src/Kafka/Consumer/Callbacks.hs view
@@ -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
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>-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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>-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 #-}
src/Kafka/Consumer/Convert.hs view
@@ -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
src/Kafka/Consumer/Subscription.hs view
@@ -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)
src/Kafka/Consumer/Types.hs view
@@ -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'))
src/Kafka/Dump.hs view
@@ -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"
src/Kafka/Internal/CancellationToken.hs view
@@ -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)
src/Kafka/Internal/RdKafka.chs view
@@ -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 <librdkafka/rdkafka.h> @@ -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
src/Kafka/Internal/Setup.hs view
@@ -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)
src/Kafka/Internal/Shared.hs view
@@ -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
src/Kafka/Metadata.hs view
@@ -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)
src/Kafka/Producer.hs view
@@ -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
src/Kafka/Producer/Callbacks.hs view
@@ -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 ()
src/Kafka/Producer/Convert.hs view
@@ -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
src/Kafka/Producer/ProducerProperties.hs view
@@ -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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>-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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>-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)]
src/Kafka/Producer/Types.hs view
@@ -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)
src/Kafka/Types.hs view
@@ -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"
tests-it/Kafka/IntegrationSpec.hs view
@@ -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
tests-it/Kafka/TestEnv.hs view
@@ -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
tests/Kafka/Consumer/ConsumerRecordMapSpec.hs view
@@ -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)
tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs view
@@ -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