pipes-kafka (empty) → 0.2.0.0

raw patch · 4 files changed

+142/−0 lines, 4 filesdep +basedep +bytestringdep +exceptionssetup-changed

Dependencies added: base, bytestring, exceptions, hw-kafka-client, monad-logger, pipes, pipes-safe, text, transformers, transformers-base

Files

• LICENSE +20/−0
• Setup.hs +3/−0
• pipes-kafka.cabal +36/−0
• src/Pipes/Kafka.hs +83/−0

@@ -0,0 +1,20 @@+Copyright (c) 2015 Ben Ford++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

@@ -0,0 +1,3 @@+import           Distribution.Simple++main = defaultMain

@@ -0,0 +1,36 @@+-- Initial pipes-kafka.cabal generated by cabal init.  For further +-- documentation, see http://haskell.org/cabal/users-guide/++name:                pipes-kafka+version:             0.2.0.0+synopsis:            Kafka in the Pipes ecosystem+-- description:         +homepage:            https://github.com/boothead/pipes-kafka+license:             MIT+license-file:        LICENSE+author:              Ben Ford+maintainer:          ben@perurbis.com+-- copyright:           +category:            Network+build-type:          Simple+-- extra-source-files:  +cabal-version:       >= 1.10 ++library+  exposed-modules:     Pipes.Kafka     +  -- other-modules:       +  -- other-extensions:    +  build-depends:       base >= 4 && < 5+                     , bytestring+                     , exceptions+                     , hw-kafka-client+                     , monad-logger+                     , pipes+                     , pipes-safe+                     , text+                     , transformers+                     , transformers-base++  hs-source-dirs:      src+  ghc-options:         -Wall+  default-language:    Haskell2010

@@ -0,0 +1,83 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell   #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++module Pipes.Kafka where++import           Control.Concurrent        (threadDelay)+import           Control.Monad             (forever)+import           Control.Monad.Catch       (throwM)+import           Control.Monad.IO.Class    (MonadIO, liftIO)+import           Control.Monad.Logger      (MonadLogger, logDebug, logWarn)+import           Control.Monad.Trans.Class (lift)+import qualified Data.ByteString           as BS++import           Data.Text                 (pack)+import           Kafka.Consumer            (ConsumerProperties, ConsumerRecord,+                                            Subscription, Timeout (..),+                                            TopicName (..), closeConsumer,+                                            newConsumer, pollMessage)+import           Kafka.Producer            (ProducePartition (..),+                                            ProducerProperties,+                                            ProducerRecord (..),+                                            RdKafkaRespErrT (..), closeProducer,+                                            newProducer, produceMessage)+import           Kafka.Types               (KafkaError (..))+import           Pipes                     (Consumer, Producer)+import qualified Pipes                     as P++import           Pipes.Safe                (MonadSafe)+import qualified Pipes.Safe                as PS++kafkaSink ::+     (MonadIO m, MonadSafe m)+  => ProducerProperties+  -> TopicName+  -> Consumer BS.ByteString m ()+kafkaSink props topic = PS.bracket connect release publish+  where+    connect = do+      res <- liftIO $ newProducer props+      case res of+        Left err       -> liftIO $ throwM err+        Right producer -> pure producer+    release producer = liftIO $ closeProducer producer+    publish producer =+      forever $ do+        m <- P.await+        liftIO $ produceMessage producer $ mkMessage topic Nothing (Just m)+    mkMessage t k v =+      ProducerRecord+      {prTopic = t, prPartition = UnassignedPartition, prKey = k, prValue = v}++kafkaSource ::+     (MonadIO m, MonadSafe m, MonadLogger m)+  => ConsumerProperties+  -> Subscription+  -> Timeout+  -> Producer (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)) m ()+kafkaSource props subs timeout@(Timeout timeoutms) =+  PS.bracket connect release consume+  where+    connect = do+      res <- liftIO $ newConsumer props subs+      case res of+        Left err       -> liftIO $ throwM err+        Right consumer -> pure consumer+    release consumer = liftIO $ closeConsumer consumer+    consume consumer =+      forever $ do+        res <- liftIO $ pollMessage consumer timeout+        case res of+          Left (KafkaResponseError RdKafkaRespErrPartitionEof) ->+            waitForMessages timeoutms+          Left (KafkaResponseError RdKafkaRespErrTimedOut) -> waitForMessages 0+          Left err -> do+            lift . $(logWarn) . pack . show $ err+            waitForMessages timeoutms+          Right m -> P.yield m+    waitForMessages t = do+      lift $ $(logDebug) "waiting for new messages"+      liftIO . threadDelay $ 1000 * t++instance MonadLogger m => MonadLogger (PS.SafeT m)