hw-kafka-client 2.1.0 → 2.1.1

raw patch · 16 files changed

+128/−105 lines, 16 files

Files

hw-kafka-client.cabal +7/−10
src/Kafka/Callbacks.hs +1/−0
src/Kafka/Consumer.hs +4/−6
src/Kafka/Consumer/Callbacks.hs +1/−0
src/Kafka/Consumer/ConsumerProperties.hs +1/−0
src/Kafka/Consumer/Types.hs +9/−13
src/Kafka/Dump.hs +51/−0
src/Kafka/Internal/Dump.hs +0/−50
src/Kafka/Internal/RdKafka.chs +1/−2
src/Kafka/Internal/Setup.hs +36/−0
src/Kafka/Internal/Shared.hs +1/−0
src/Kafka/Metadata.hs +1/−0
src/Kafka/Producer.hs +3/−1
src/Kafka/Producer/ProducerProperties.hs +6/−5
src/Kafka/Producer/Types.hs +6/−1
src/Kafka/Types.hs +0/−17

hw-kafka-client.cabal view

@@ -1,5 +1,5 @@ name:                hw-kafka-client-version:             2.1.0+version:             2.1.1 homepage:            https://github.com/haskell-works/hw-kafka-client bug-reports:         https://github.com/haskell-works/hw-kafka-client/issues license:             MIT@@ -65,31 +65,28 @@     build-depends: semigroups    exposed-modules:-    Kafka.Types+    Kafka.Consumer     Kafka.Consumer.ConsumerProperties     Kafka.Consumer.Subscription     Kafka.Consumer.Types-    Kafka.Consumer+    Kafka.Dump     Kafka.Metadata     Kafka.Producer     Kafka.Producer.ProducerProperties     Kafka.Producer.Types+    Kafka.Types   other-modules:     Kafka.Callbacks-    Kafka.Consumer.Convert     Kafka.Consumer.Callbacks+    Kafka.Consumer.Convert     Kafka.Internal.CancellationToken-    Kafka.Internal.Shared+    Kafka.Internal.RdKafka     Kafka.Internal.Setup+    Kafka.Internal.Shared     Kafka.Producer.Convert-    Kafka.Internal.RdKafka-    Kafka.Internal.Dump   hs-source-dirs:      src   default-language:    Haskell2010   ghc-options:         -Wall-  include-dirs:        /usr/local/include/librdkafka-                     , /usr/include/librdkafka-  extra-lib-dirs:      /usr/local/lib   extra-libraries:     rdkafka   if os(darwin)     cpp-options: -D__attribute__(A)= -D_Nullable= -D_Nonnull=

src/Kafka/Callbacks.hs view

@@ -5,6 +5,7 @@ where  import Kafka.Internal.RdKafka+import Kafka.Internal.Setup import Kafka.Types  errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO ()

src/Kafka/Consumer.hs view

@@ -11,11 +11,7 @@ , closeConsumer  -- ReExport Types-, TopicName (..)-, CIT.OffsetCommit (..)-, CIT.PartitionOffset (..)-, CIT.TopicPartition (..)-, CIT.ConsumerRecord (..)+, KafkaConsumer , RdKafkaRespErrT (..) ) where@@ -30,16 +26,18 @@ import qualified Data.Map                         as M import           Foreign                          hiding (void) import           Kafka.Consumer.Convert+import           Kafka.Consumer.Types import           Kafka.Internal.CancellationToken as CToken import           Kafka.Internal.RdKafka import           Kafka.Internal.Setup import           Kafka.Internal.Shared + import qualified Kafka.Consumer.Types as CIT  import Kafka.Consumer.ConsumerProperties as X import Kafka.Consumer.Subscription       as X-import Kafka.Consumer.Types              as X+import Kafka.Consumer.Types              as X hiding (KafkaConsumer) import Kafka.Types                       as X  -- | Runs high-level kafka consumer.

src/Kafka/Consumer/Callbacks.hs view

@@ -10,6 +10,7 @@ import Kafka.Consumer.Convert import Kafka.Consumer.Types import Kafka.Internal.RdKafka+import Kafka.Internal.Setup import Kafka.Types  -- | Sets a callback that is called when rebalance is needed.

src/Kafka/Consumer/ConsumerProperties.hs view

@@ -12,6 +12,7 @@ import           Data.Semigroup import           Kafka.Consumer.Callbacks import           Kafka.Consumer.Types+import           Kafka.Internal.Setup import           Kafka.Types  -- | Properties to create 'KafkaConsumer'.

src/Kafka/Consumer/Types.hs view

@@ -8,6 +8,7 @@ import Data.Bitraversable import Data.Int import Data.Typeable+import Kafka.Internal.Setup import Kafka.Types  data KafkaConsumer = KafkaConsumer@@ -27,6 +28,14 @@ newtype Offset          = Offset Int64 deriving (Show, Eq, Read) data OffsetReset        = Earliest | Latest deriving (Show, Eq) +data PartitionOffset =+    PartitionOffsetBeginning+  | PartitionOffsetEnd+  | PartitionOffset Int64+  | PartitionOffsetStored+  | PartitionOffsetInvalid+  deriving (Eq, Show)+ data SubscribedPartitions   = SubscribedPartitions [PartitionId]   | SubscribedPartitionsAll@@ -142,16 +151,3 @@ bitraverseM f g r = bisequenceA <$> bimapM f g r {-# INLINE bitraverseM #-} -data PartitionOffset =-    PartitionOffsetBeginning-  | PartitionOffsetEnd-  | PartitionOffset Int64-  | PartitionOffsetStored-  | PartitionOffsetInvalid-  deriving (Eq, Show)--data CommitPartitionOffset = CommitPartitionOffset-  { cpoTopicName   :: !TopicName-  , cpoPartitionId :: !PartitionId-  , cpoOffset      :: !Offset-  } deriving (Eq, Show)

+ src/Kafka/Dump.hs view

@@ -0,0 +1,51 @@+module Kafka.Dump+( hPrintSupportedKafkaConf+, hPrintKafka+, dumpKafkaConf+, dumpTopicConf+)+where++--+import Kafka.Internal.RdKafka+import Kafka.Internal.Setup++import           Control.Monad+import           Control.Monad.IO.Class+import           Data.Map.Strict        (Map)+import qualified Data.Map.Strict        as Map+import           Foreign+import           Foreign.C.String+import           System.IO++-- | Prints out all supported Kafka conf properties to a handle+hPrintSupportedKafkaConf :: MonadIO m => Handle -> m ()+hPrintSupportedKafkaConf h = liftIO $ handleToCFile h "w" >>= rdKafkaConfPropertiesShow++-- | Prints out all data associated with a specific kafka object to a handle+hPrintKafka :: (MonadIO m, HasKafka k) => Handle -> k -> m ()+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 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 k = liftIO $ parseDump (rdKafkaConfDump (getRdKafkaConf k))++parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String)+parseDump cstr = alloca $ \sizeptr -> do+    strPtr <- cstr sizeptr+    size <- peek sizeptr++    keysAndValues <- mapM (peekCString <=< 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"

− src/Kafka/Internal/Dump.hs

@@ -1,50 +0,0 @@-module Kafka.Internal.Dump-where-----import           Kafka.Internal.RdKafka--import           Control.Monad-import           Data.Map.Strict            (Map)-import           Foreign-import           Foreign.C.String-import           System.IO-import qualified Data.Map.Strict            as Map---- | Prints out all supported Kafka conf properties to a handle-hPrintSupportedKafkaConf :: Handle -> IO ()-hPrintSupportedKafkaConf h = handleToCFile h "w" >>= rdKafkaConfPropertiesShow---- | Prints out all data associated with a specific kafka object to a handle-hPrintKafka :: Handle -> RdKafkaTPtr -> IO ()-hPrintKafka h k = handleToCFile h "w" >>= \f -> rdKafkaDump f k---- | Returns a map of the current kafka configuration-dumpConfFromKafka :: RdKafkaConfTPtr -> IO (Map String String)-dumpConfFromKafka = dumpKafkaConf---- | Returns a map of the current topic configuration-dumpConfFromKafkaTopic :: RdKafkaTopicConfTPtr -> IO (Map String String)-dumpConfFromKafkaTopic = dumpTopicConf--dumpTopicConf :: RdKafkaTopicConfTPtr -> IO (Map String String)-dumpTopicConf kptr = parseDump (rdKafkaTopicConfDump kptr)--dumpKafkaConf :: RdKafkaConfTPtr -> IO (Map String String)-dumpKafkaConf kptr = parseDump (rdKafkaConfDump kptr)--parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String)-parseDump cstr = alloca $ \sizeptr -> do-    strPtr <- cstr sizeptr-    size <- peek sizeptr--    keysAndValues <- mapM (peekCString <=< 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"

src/Kafka/Internal/RdKafka.chs view

@@ -3,7 +3,6 @@  module Kafka.Internal.RdKafka where ---import Control.Applicative import Control.Monad import Data.Word import Foreign@@ -14,7 +13,7 @@ import System.Posix.IO import System.Posix.Types -#include "rdkafka.h"+#include <librdkafka/rdkafka.h>  type CInt64T = {#type int64_t #} type CInt32T = {#type int32_t #}

src/Kafka/Internal/Setup.hs view

@@ -14,6 +14,42 @@ -- newtype KafkaProps = KafkaProps [(String, String)] deriving (Show, Eq) newtype TopicProps = TopicProps [(String, String)] deriving (Show, Eq)+newtype Kafka      = Kafka RdKafkaTPtr deriving Show+data KafkaConf     = KafkaConf RdKafkaConfTPtr CancellationToken+newtype TopicConf  = TopicConf RdKafkaTopicConfTPtr deriving Show++class HasKafka a where+  getKafka :: a -> Kafka++class HasKafkaConf a where+  getKafkaConf :: a -> KafkaConf++class HasTopicConf a where+  getTopicConf :: a -> TopicConf++instance HasKafkaConf KafkaConf where+  getKafkaConf = id+  {-# INLINE getKafkaConf #-}++instance HasKafka Kafka where+  getKafka = id+  {-# INLINE getKafka #-}++instance HasTopicConf TopicConf where+  getTopicConf = id+  {-# INLINE getTopicConf #-}++getRdKafka :: HasKafka k => k -> RdKafkaTPtr+getRdKafka k = let (Kafka k') = getKafka k in k'+{-# INLINE getRdKafka #-}++getRdKafkaConf :: HasKafkaConf k => k -> RdKafkaConfTPtr+getRdKafkaConf k = let (KafkaConf k' _) = getKafkaConf k in k'+{-# INLINE getRdKafkaConf #-}++getRdTopicConf :: HasTopicConf t => t -> RdKafkaTopicConfTPtr+getRdTopicConf t = let (TopicConf t') = getTopicConf t in t'+{-# INLINE getRdTopicConf #-}  newTopicConf :: IO TopicConf newTopicConf = TopicConf <$> newRdKafkaTopicConfT

src/Kafka/Internal/Shared.hs view

@@ -9,6 +9,7 @@ import           Foreign.C.Error import           Kafka.Internal.CancellationToken as CToken import           Kafka.Internal.RdKafka+import           Kafka.Internal.Setup import           Kafka.Types  runEventLoop :: HasKafka a => a -> CancellationToken -> Maybe Timeout -> IO ()

src/Kafka/Metadata.hs view

@@ -24,6 +24,7 @@ import           Kafka.Consumer.Convert import           Kafka.Consumer.Types import           Kafka.Internal.RdKafka+import           Kafka.Internal.Setup import           Kafka.Internal.Shared import           Kafka.Types

src/Kafka/Producer.hs view

@@ -6,6 +6,7 @@ , produceMessage, produceMessageBatch , flushProducer , closeProducer+, KafkaProducer , RdKafkaRespErrT (..) ) where@@ -26,9 +27,10 @@ import           Kafka.Internal.Setup import           Kafka.Internal.Shared import           Kafka.Producer.Convert+import           Kafka.Producer.Types  import Kafka.Producer.ProducerProperties as X-import Kafka.Producer.Types              as X+import Kafka.Producer.Types              as X hiding (KafkaProducer) import Kafka.Types                       as X  -- | Runs Kafka Producer.

src/Kafka/Producer/ProducerProperties.hs view

@@ -5,11 +5,12 @@ where  import           Control.Monad-import qualified Data.List       as L-import           Data.Map        (Map)-import qualified Data.Map        as M+import qualified Data.List            as L+import           Data.Map             (Map)+import qualified Data.Map             as M import           Data.Semigroup import           Kafka.Callbacks+import           Kafka.Internal.Setup import           Kafka.Types  -- | Properties to create 'KafkaProducer'.@@ -19,12 +20,12 @@   , ppLogLevel   :: Maybe KafkaLogLevel   , callbacks    :: [KafkaConf -> IO ()]   }-  + -- | /Right biased/ so we prefer newer properties over older ones. instance Semigroup ProducerProperties where   (<>) (ProducerProperties k1 t1 ll1 cb1) (ProducerProperties k2 t2 ll2 cb2) =     ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb2 `mplus` cb1)-    + instance Monoid ProducerProperties where   mempty = ProducerProperties M.empty M.empty Nothing []   mappend = (<>)

src/Kafka/Producer/Types.hs view

@@ -3,8 +3,9 @@  where -import qualified Data.ByteString as BS+import qualified Data.ByteString      as BS import           Data.Typeable+import           Kafka.Internal.Setup import           Kafka.Types  -- | Main pointer to Kafka object, which contains our brokers@@ -21,6 +22,10 @@ instance HasKafkaConf KafkaProducer where   getKafkaConf = kpKafkaConf   {-# INLINE getKafkaConf #-}++instance HasTopicConf KafkaProducer where+  getTopicConf = kpTopicConf+  {-# INLINE getTopicConf #-}  -- | Represents messages /to be enqueued/ onto a Kafka broker (i.e. used for a producer) data ProducerRecord = ProducerRecord

src/Kafka/Types.hs view

@@ -9,31 +9,14 @@ import Kafka.Internal.CancellationToken import Kafka.Internal.RdKafka -class HasKafka a where-  getKafka :: a -> Kafka--class HasKafkaConf a where-  getKafkaConf :: a -> KafkaConf- newtype BrokerId =   BrokerId Int   deriving (Show, Eq, Ord, Read) -newtype Kafka     = Kafka RdKafkaTPtr deriving Show-data KafkaConf    = KafkaConf RdKafkaConfTPtr CancellationToken-newtype TopicConf = TopicConf RdKafkaTopicConfTPtr deriving Show  newtype PartitionId = PartitionId Int deriving (Show, Eq, Read, Ord, Enum) newtype Millis      = Millis Int64 deriving (Show, Read, Eq, Ord, Num) newtype ClientId    = ClientId String deriving (Show, Eq, Ord)--instance HasKafkaConf KafkaConf where-  getKafkaConf = id-  {-# INLINE getKafkaConf #-}--instance HasKafka Kafka where-  getKafka = id-  {-# INLINE getKafka #-}  -- | Topic name to be consumed --