Merge pull request #112 from phile314/move_autocommit

Move auto-commit from subscription to consumer

Author: AlexeyRaga

src/Kafka/Consumer/ConsumerProperties.hs

@@ -4,6 +4,7 @@ module Kafka.Consumer.ConsumerProperties
 ( ConsumerProperties(..)
 , CallbackMode(..)
 , brokersList
+, autoCommit
 , noAutoCommit
 , noAutoOffsetStore
 , groupId
@@ -29,7 +30,7 @@ import           Data.Text            (Text)
 import qualified Data.Text            as Text
 import           Kafka.Consumer.Types (ConsumerGroupId (..))
 import           Kafka.Internal.Setup (KafkaConf (..))
-import           Kafka.Types          (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), kafkaCompressionCodecToText, kafkaDebugToText)
+import           Kafka.Types          (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), kafkaCompressionCodecToText, kafkaDebugToText, Millis(..))
 
 import Kafka.Consumer.Callbacks as X
 
@@ -65,6 +66,13 @@ brokersList bs =
   let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)
    in extraProps $ M.fromList [("bootstrap.servers", bs')]
 
+autoCommit :: Millis -> ConsumerProperties
+autoCommit (Millis ms) = extraProps $
+  M.fromList
+    [ ("enable.auto.commit", "true")
+    , ("auto.commit.interval.ms", Text.pack $ show ms)
+    ]
+
 -- | Disables auto commit for the consumer
 noAutoCommit :: ConsumerProperties
 noAutoCommit =

src/Kafka/Consumer/Subscription.hs

@@ -4,7 +4,6 @@ module Kafka.Consumer.Subscription
 ( Subscription(..)
 , topics
 , offsetReset
-, autoCommit
 , extraSubscriptionProps
 )
 where
@@ -44,12 +43,5 @@ offsetReset o =
              Latest   -> "latest"
    in Subscription (Set.empty) (M.fromList [("auto.offset.reset", o')])
 
-autoCommit :: Millis -> Subscription
-autoCommit (Millis ms) = Subscription (Set.empty) $
-  M.fromList
-    [ ("enable.auto.commit", "true")
-    , ("auto.commit.interval.ms", Text.pack $ show ms)
-    ]
-
 extraSubscriptionProps :: Map Text Text -> Subscription
 extraSubscriptionProps = Subscription (Set.empty)