hw-kafka-client 1.1.3 → 1.1.4

raw patch · 3 files changed

+31/−20 lines, 3 files

Files

• hw-kafka-client.cabal +1/−1
• src/Kafka/Consumer.hs +20/−10
• src/Kafka/Consumer/Types.hs +10/−9

@@ -1,5 +1,5 @@ name:                hw-kafka-client-version:             1.1.3+version:             1.1.4 homepage:            https://github.com/haskell-works/hw-kafka-client bug-reports:         https://github.com/haskell-works/hw-kafka-client/issues license:             MIT

@@ -7,6 +7,7 @@ , pollMessage , commitOffsetMessage , commitAllOffsets+, commitPartitionsOffsets , closeConsumer  -- ReExport Types@@ -20,24 +21,24 @@ where  import           Control.Exception-import           Control.Monad (forM_)+import           Control.Monad                     (forM_) import           Control.Monad.IO.Class-import qualified Data.ByteString as BS-import qualified Data.Map as M-import           Foreign                         hiding (void)+import qualified Data.ByteString                   as BS+import qualified Data.Map                          as M+import           Foreign                           hiding (void) import           Kafka.Consumer.Convert import           Kafka.Internal.RdKafka import           Kafka.Internal.RdKafkaEnum import           Kafka.Internal.Setup import           Kafka.Internal.Shared -import qualified Kafka.Consumer.Types   as CIT-import qualified Kafka.Internal.RdKafkaEnum      as RDE+import qualified Kafka.Consumer.Types              as CIT+import qualified Kafka.Internal.RdKafkaEnum        as RDE -import Kafka.Types as X-import Kafka.Consumer.Types as X-import Kafka.Consumer.Subscription as X-import Kafka.Consumer.ConsumerProperties as X+import           Kafka.Consumer.ConsumerProperties as X+import           Kafka.Consumer.Subscription       as X+import           Kafka.Consumer.Types              as X+import           Kafka.Types                       as X  -- | Runs high-level kafka consumer. --@@ -106,6 +107,15 @@                  -> m (Maybe KafkaError) commitAllOffsets o k =   liftIO $ newForeignPtr_ nullPtr >>= commitOffsets o k++-- | Commit offsets for all currently assigned partitions.+commitPartitionsOffsets :: MonadIO m+                 => OffsetCommit+                 -> KafkaConsumer+                 -> [TopicPartition]+                 -> m (Maybe KafkaError)+commitPartitionsOffsets o k ps =+  liftIO $ toNativeTopicPartitionList ps >>= commitOffsets o k  -- | Assigns specified partitions to a current consumer. -- Assigning an empty list means unassigning from all partitions that are currently assigned.

@@ -1,15 +1,16 @@-{-# LANGUAGE DeriveDataTypeable, GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} module Kafka.Consumer.Types  where -import Data.Bifunctor-import Data.Bifoldable-import Data.Bitraversable-import Data.Int-import Data.Typeable-import Kafka.Types-import Kafka.Internal.RdKafka+import           Data.Bifoldable+import           Data.Bifunctor+import           Data.Bitraversable+import           Data.Int+import           Data.Typeable+import           Kafka.Internal.RdKafka+import           Kafka.Types  data KafkaConsumer = KafkaConsumer { kcKafkaPtr :: !RdKafkaTPtr, kcKafkaConf :: !RdKafkaConfTPtr} deriving (Show) @@ -18,7 +19,7 @@  newtype ConsumerGroupId = ConsumerGroupId String deriving (Show, Eq) newtype Offset          = Offset Int64 deriving (Show, Eq, Read)-newtype PartitionId     = PartitionId Int deriving (Show, Eq, Read)+newtype PartitionId     = PartitionId Int deriving (Show, Eq, Read, Ord) newtype Millis          = Millis Int deriving (Show, Eq, Ord, Num) newtype ClientId        = ClientId String deriving (Show, Eq, Ord) data OffsetReset        = Earliest | Latest deriving (Show, Eq)