diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -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.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,3 @@
+import           Distribution.Simple
+
+main = defaultMain
diff --git a/pipes-kafka.cabal b/pipes-kafka.cabal
new file mode 100644
--- /dev/null
+++ b/pipes-kafka.cabal
@@ -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
diff --git a/src/Pipes/Kafka.hs b/src/Pipes/Kafka.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Kafka.hs
@@ -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)