packages feed

haskakafka 1.1.0 → 1.2.0

raw patch · 9 files changed

+97/−54 lines, 9 files

Files

haskakafka.cabal view
@@ -1,5 +1,5 @@ name:                haskakafka-version:             1.1.0+version:             1.2.0 synopsis:            Kafka bindings for Haskell description:         Apache Kafka bindings backed by the librdkafka                      C library. This implies full consumer and producer@@ -26,8 +26,8 @@                      , unix   exposed-modules:     Haskakafka-    Haskakafka.InternalRdKafka     Haskakafka.InternalRdKafkaEnum+    Haskakafka.InternalRdKafka     Haskakafka.InternalSetup     Haskakafka.InternalTypes     Haskakafka.ConsumerExample@@ -39,9 +39,9 @@   hs-source-dirs:      src   default-language:    Haskell2010   ghc-options:         -Wall-  include-dirs:        /usr/local/include/librdkafka-  extra-lib-dirs:      /usr/local/lib-  extra-libraries:     rdkafka+  pkgconfig-depends:   rdkafka+  if os(darwin)+      cpp-options: -D__attribute__(A)= -D_Nullable= -D_Nonnull=  executable simple   main-is:              Simple.hs
src/Haskakafka.hs view
@@ -8,8 +8,13 @@ , produceKeyedMessage , produceMessageBatch , storeOffset+, seekToOffset , getAllMetadata , getTopicMetadata+, handleProduceErr+, producePartitionInteger+, pollEvents+, pollEventsSafe  -- Internal objects , IS.newKafka@@ -22,8 +27,8 @@ , rdKafkaVersionStr  -- Type re-exports-, IT.Kafka-, IT.KafkaTopic+, IT.Kafka(..)+, IT.KafkaTopic(..)  , IT.KafkaOffset(..) , IT.KafkaMessage(..)@@ -38,7 +43,7 @@  , IT.KafkaLogLevel(..) , IT.KafkaError(..)-, RDE.RdKafkaRespErrT+, RDE.RdKafkaRespErrT(..)  -- Pseudo-internal , addBrokers@@ -82,13 +87,7 @@ -- 'consumeMessage' won't work without it. This function is non-blocking. startConsuming :: KafkaTopic -> Int -> KafkaOffset -> IO () startConsuming (KafkaTopic topicPtr _ _) partition offset =-    let trueOffset = case offset of-                        KafkaOffsetBeginning -> (- 2)-                        KafkaOffsetEnd -> (- 1)-                        KafkaOffsetStored -> (- 1000)-                        KafkaOffsetInvalid -> (- 1001)-                        KafkaOffset i -> i-    in throwOnError $ rdKafkaConsumeStart topicPtr partition trueOffset+    throwOnError $ rdKafkaConsumeStart topicPtr partition $ offsetToInt64 offset  -- | Stops consuming for a given topic. You probably do not need to call -- this directly (it is called automatically when 'withKafkaConsumer' terminates).@@ -134,6 +133,18 @@         [] -> return $ Right $ rights ms         l  -> return $ Left $ head l +seekToOffset :: KafkaTopic+    -> Int -- ^ partition number+    -> KafkaOffset -- ^ destination+    -> Int -- ^ timeout in milliseconds+    -> IO (Maybe KafkaError)+seekToOffset (KafkaTopic ptr _ _) p ofs timeout = do+  err <- rdKafkaSeek ptr (fromIntegral p)+      (fromIntegral $ offsetToInt64 ofs) timeout+  case err of+    RdKafkaRespErrNoError -> return Nothing+    e -> return $ Just $ KafkaResponseError e+ -- | Store a partition's offset in librdkafka's offset store. This function only needs to be called -- if auto.commit.enable is false. See <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md> -- for information on how to configure the offset store.@@ -216,6 +227,7 @@                   , offset'RdKafkaMessageT = 0                   , keyLen'RdKafkaMessageT = 0                   , key'RdKafkaMessageT = nullPtr+                  , private'RdKafkaMessageT = nullPtr                   }     produceMessageToMessage (KafkaProduceKeyedMessage kbs bs) =  do         let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr bs@@ -236,6 +248,7 @@                     , offset'RdKafkaMessageT = 0                     , keyLen'RdKafkaMessageT = keyLength                     , key'RdKafkaMessageT = passedKey+                    , private'RdKafkaMessageT = nullPtr                     }  -- | Connects to Kafka broker in producer mode for a given topic, taking a function@@ -394,6 +407,12 @@  pollEvents :: Kafka -> Int -> IO () pollEvents (Kafka kPtr _) timeout = rdKafkaPoll kPtr timeout >> return ()++pollEventsSafe :: Kafka -> Int -> IO ()+pollEventsSafe (Kafka kPtr _) timeout = do+  _ <- withForeignPtr kPtr $ \realPtr -> do+    rdKafkaPollSafe realPtr timeout+  return ()  outboundQueueLength :: Kafka -> IO (Int) outboundQueueLength (Kafka kPtr _) = rdKafkaOutqLen kPtr
src/Haskakafka/Consumer/Internal/Convert.hs view
@@ -3,12 +3,11 @@ where  import           Control.Monad-import           Data.Int import           Foreign import           Foreign.C.String import           Haskakafka.Consumer.Internal.Types import           Haskakafka.InternalRdKafka-import           Haskakafka.InternalTypes+import           Haskakafka.InternalShared  -- | Converts offsets sync policy to integer (the way Kafka understands it): --@@ -24,24 +23,6 @@         OffsetSyncImmediate -> 0         OffsetSyncInterval ms -> ms {-# INLINE offsetSyncToInt #-}--offsetToInt64 :: KafkaOffset -> Int64-offsetToInt64 o = case o of-    KafkaOffsetBeginning -> -2-    KafkaOffsetEnd       -> -1-    KafkaOffset off      -> off-    KafkaOffsetStored    -> -1000-    KafkaOffsetInvalid   -> -1001-{-# INLINE offsetToInt64 #-}--int64ToOffset :: Int64 -> KafkaOffset-int64ToOffset o-    | o == -2    = KafkaOffsetBeginning-    | o == -1    = KafkaOffsetEnd-    | o == -1000 = KafkaOffsetStored-    | o >= 0     = KafkaOffset o-    | otherwise  = KafkaOffsetInvalid-{-# INLINE int64ToOffset #-}  fromNativeTopicPartitionList :: RdKafkaTopicPartitionListT -> IO [KafkaTopicPartition] fromNativeTopicPartitionList pl =
src/Haskakafka/ConsumerExample.hs view
@@ -5,7 +5,7 @@ import           Control.Arrow                  ((&&&)) import           Haskakafka import           Haskakafka.Consumer-import           Haskakafka.InternalRdKafkaEnum+import           Haskakafka.InternalRdKafkaEnum ()  iterator :: [Integer] iterator = [0 .. 20]
src/Haskakafka/InternalRdKafka.chs view
@@ -15,8 +15,10 @@ import System.Posix.IO import System.Posix.Types -#include "rdkafka.h"+#include "librdkafka/rdkafka.h" +{#import Haskakafka.InternalRdKafkaEnum#}+ type CInt64T = {#type int64_t #} type CInt32T = {#type int32_t #} @@ -129,6 +131,7 @@     , offset'RdKafkaMessageT :: Int64     , payload'RdKafkaMessageT :: Word8Ptr     , key'RdKafkaMessageT :: Word8Ptr+    , private'RdKafkaMessageT :: Ptr ()     }     deriving (Show, Eq) @@ -144,6 +147,7 @@         <*> liftM fromIntegral  ({#get rd_kafka_message_t->offset #} p)         <*> liftM castPtr       ({#get rd_kafka_message_t->payload #} p)         <*> liftM castPtr       ({#get rd_kafka_message_t->key #} p)+        <*> liftM castPtr       ({#get rd_kafka_message_t->_private #} p)     poke p x = do       {#set rd_kafka_message_t.err#}        p (enumToCInt   $ err'RdKafkaMessageT x)       {#set rd_kafka_message_t.rkt#}        p (castPtr      $ topic'RdKafkaMessageT x)@@ -153,6 +157,7 @@       {#set rd_kafka_message_t.offset#}     p (fromIntegral $ offset'RdKafkaMessageT x)       {#set rd_kafka_message_t.payload#}    p (castPtr      $ payload'RdKafkaMessageT x)       {#set rd_kafka_message_t.key#}        p (castPtr      $ key'RdKafkaMessageT x)+      {#set rd_kafka_message_t._private#}   p (castPtr      $ private'RdKafkaMessageT x)  {#pointer *rd_kafka_message_t as RdKafkaMessageTPtr foreign -> RdKafkaMessageT #} @@ -167,7 +172,7 @@  instance Storable RdKafkaMetadataBrokerT where   alignment _ = {#alignof rd_kafka_metadata_broker_t#}-  sizeOf _ = 24+  sizeOf _ = {#sizeof rd_kafka_metadata_broker_t#}   peek p = RdKafkaMetadataBrokerT     <$> liftM fromIntegral ({#get rd_kafka_metadata_broker_t->id #} p)     <*> liftM id ({#get rd_kafka_metadata_broker_t->host #} p)@@ -209,7 +214,7 @@  instance Storable RdKafkaMetadataTopicT where   alignment _ = {#alignof rd_kafka_metadata_topic_t#}-  sizeOf _ = 32+  sizeOf _ = {#sizeof rd_kafka_metadata_topic_t#}   peek p = RdKafkaMetadataTopicT     <$> liftM id ({#get rd_kafka_metadata_topic_t->topic #} p)     <*> liftM fromIntegral ({#get rd_kafka_metadata_topic_t->partition_cnt #} p)@@ -514,9 +519,13 @@     {`RdKafkaTPtr', `RdKafkaMessageTPtr', `Int'}     -> `RdKafkaRespErrT' cIntToEnum #} -{#fun unsafe rd_kafka_position as ^+{#fun unsafe rd_kafka_committed as ^     {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Int'}     -> `RdKafkaRespErrT' cIntToEnum #}++{#fun unsafe rd_kafka_position as ^+    {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'}+    -> `RdKafkaRespErrT' cIntToEnum #} ------------------------------------------------------------------------------------------------- ---- Groups data RdKafkaGroupMemberInfoT = RdKafkaGroupMemberInfoT@@ -623,6 +632,9 @@ foreign import ccall unsafe "rdkafka.h rd_kafka_message_destroy"     rdKafkaMessageDestroy :: Ptr RdKafkaMessageT -> IO () +{#fun unsafe rd_kafka_message_timestamp as ^+    {`RdKafkaMessageTPtr', `RdKafkaTimestampTypeTPtr'} -> `CInt64T' cIntConv #}+ -- rd_kafka_conf {#fun unsafe rd_kafka_conf_new as ^     {} -> `RdKafkaConfTPtr' #}@@ -720,6 +732,10 @@   {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Ptr (Ptr RdKafkaMessageT)', cIntConv `CSize'}   -> `CSize' cIntConv #} +{#fun rd_kafka_seek as ^+  {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T', `Int'}+  -> `RdKafkaRespErrT' cIntToEnum #}+ rdKafkaConsumeStop :: RdKafkaTopicTPtr -> Int -> IO (Maybe String) rdKafkaConsumeStop topicPtr partition = do     i <- rdKafkaConsumeStopInternal topicPtr (fromIntegral partition)@@ -754,7 +770,10 @@ {# fun unsafe rd_kafka_metadata_destroy as ^    {castPtr `Ptr RdKafkaMetadataT'} -> `()' #} -{#fun unsafe rd_kafka_poll as ^+foreign import ccall safe "rd_kafka.h rd_kafka_poll"+  rdKafkaPollSafe :: Ptr RdKafkaT -> Int -> IO Int++{# fun unsafe rd_kafka_poll as ^     {`RdKafkaTPtr', `Int'} -> `Int' #}  {#fun unsafe rd_kafka_outq_len as ^
src/Haskakafka/InternalRdKafkaEnum.chs view
@@ -3,8 +3,11 @@  module Haskakafka.InternalRdKafkaEnum where -#include "rdkafka.h"+#include "librdkafka/rdkafka.h"  {#enum rd_kafka_type_t as ^ {underscoreToCase} deriving (Show, Eq) #} {#enum rd_kafka_conf_res_t as ^ {underscoreToCase} deriving (Show, Eq) #} {#enum rd_kafka_resp_err_t as ^ {underscoreToCase} deriving (Show, Eq) #}+{#enum rd_kafka_timestamp_type_t as ^ {underscoreToCase} deriving (Show, Eq) #}++{#pointer *rd_kafka_timestamp_type_t as RdKafkaTimestampTypeTPtr foreign -> RdKafkaTimestampTypeT #}
src/Haskakafka/InternalShared.hs view
@@ -66,3 +66,22 @@     _ -> Left err {-# INLINE kafkaErrorToEither #-} +offsetToInt64 :: KafkaOffset -> Int64+offsetToInt64 o = case o of+    KafkaOffsetBeginning -> -2+    KafkaOffsetEnd       -> -1+    KafkaOffset off      -> off+    KafkaOffsetStored    -> -1000+    KafkaOffsetInvalid   -> -1001+    KafkaOffsetTail i    -> -2000 - i+{-# INLINE offsetToInt64 #-}++int64ToOffset :: Int64 -> KafkaOffset+int64ToOffset o+    | o == -2    = KafkaOffsetBeginning+    | o == -1    = KafkaOffsetEnd+    | o == -1000 = KafkaOffsetStored+    | o >= 0     = KafkaOffset o+    | o <= -2000 = KafkaOffsetTail (-2000 - o)+    | otherwise  = KafkaOffsetInvalid+{-# INLINE int64ToOffset #-}
src/Haskakafka/InternalTypes.hs view
@@ -49,6 +49,7 @@   -- <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md librdkafka's documentation>   -- for offset store configuration.   | KafkaOffsetStored+  | KafkaOffsetTail Int64   | KafkaOffsetInvalid   deriving (Eq, Show)
tests/TestMain.hs view
@@ -74,24 +74,25 @@    describe "Kafka topic configuration" $ do     it "should allow dumping" $ do-      kConf <- newKafkaTopicConf-      kvs <- dumpKafkaTopicConf kConf+      kTopicConf <- newKafkaTopicConf+      kvs <- dumpKafkaTopicConf kTopicConf       (Map.size kvs) `shouldSatisfy` (>0)      it "should change when set is called" $ do-      kConf <- newKafkaTopicConf-      setKafkaTopicConfValue kConf "request.timeout.ms" "20000"-      kvs <- dumpKafkaTopicConf kConf+      kTopicConf <- newKafkaTopicConf+      setKafkaTopicConfValue kTopicConf "request.timeout.ms" "20000"+      kvs <- dumpKafkaTopicConf kTopicConf       (kvs Map.! "request.timeout.ms") `shouldBe` "20000"      it "should throw an exception on unknown property" $ do-      kConf <- newKafkaTopicConf-      (setKafkaTopicConfValue kConf "blippity.blop.cosby" "120") `shouldThrow`+      kTopicConf <- newKafkaTopicConf+      (setKafkaTopicConfValue kTopicConf "blippity.blop.cosby" "120") `shouldThrow`         (\(KafkaUnknownConfigurationKey str) -> (length str) > 0)      it "should throw an exception on an invalid value" $ do-      kConf <- newKafkaTopicConf-      (setKafkaTopicConfValue kConf "request.timeout.ms" "mono...doh!") `shouldThrow`+      kTopicConf <- newKafkaTopicConf+      (setKafkaTopicConfValue kTopicConf "request.timeout.ms" "mono...doh!")+        `shouldThrow`         (\(KafkaInvalidConfigurationValue str) -> (length str) > 0)    describe "Logging" $ do@@ -150,7 +151,7 @@                   produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages         errs `shouldBe` [] -        et <- consumeMessageBatch topic 0 (5000) 3+        et <- consumeMessageBatch topic 0 kafkaDelay 3         case et of           (Left err) -> error $ show err           (Right oms) -> do@@ -161,7 +162,7 @@     it "should not fail on batch consume when no messages are available #12" $ getAddressTopic $ \a t -> do       withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do         primeEOF topic-        et <- consumeMessageBatch topic 0 (5000) 3+        et <- consumeMessageBatch topic 0 kafkaDelay 3         case et of           (Left err) -> error $ show err           (Right oms) -> do@@ -169,7 +170,7 @@      it "should return EOF on batch consume if necessary" $ getAddressTopic $ \a t -> do       withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do-        et <- consumeMessageBatch topic 0 (5000) 10+        et <- consumeMessageBatch topic 0 kafkaDelay 10         case et of           (Left err) -> print err           (Right _oms) -> error "should return EOF"