packages feed

kafka-client-sync (empty) → 0.1.0.0

raw patch · 5 files changed

+355/−0 lines, 5 filesdep +basedep +containersdep +hw-kafka-client

Dependencies added: base, containers, hw-kafka-client, kafka-client-sync, monad-parallel, text

Files

+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for hedgehog-servant++## 0.1.0.0 -- 2019-11-10++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,29 @@+BSD 3-Clause License++Copyright (c) 2019, Felix Mulder+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this+   list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its+   contributors may be used to endorse or promote products derived from+   this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ kafka-client-sync.cabal view
@@ -0,0 +1,88 @@+name:+  kafka-client-sync+version:+  0.1.0.0+synopsis:+  Synchronous Kafka Client+description:+  A synchronous API on top of librdkafka using hw-kafka-client+bug-reports:+  https://github.com/felixmulder/kafka-client-sync+cabal-version:+  1.24+license:+  BSD3+license-file:+  LICENSE+author:+  Felix Mulder+maintainer:+  felix.mulder@gmail.com+copyright:+  Felix Mulder+category:+  Network+build-type:+  Simple+extra-source-files:+  CHANGELOG.md++source-repository head+  type: git+  location: git://github.com/felixmulder/kafka-client-sync.git++library+  default-extensions:+    DefaultSignatures+    LambdaCase+    OverloadedStrings+    RecordWildCards++  ghc-options:+    -Wall -Wredundant-constraints -fhide-source-paths++  exposed-modules:+    Kafka.Producer.Sync++  build-depends:+      base                      >= 4.9      && < 5+    , containers                >= 0.6      && < 0.7+    , hw-kafka-client           >= 2.6      && < 3.1++  hs-source-dirs:+    src++  default-language:+    Haskell2010++test-suite kafka-client-sync-tests+  type:+    exitcode-stdio-1.0++  default-language:+    Haskell2010++  main-is:+    Main.hs++  ghc-options:+    -Wall -Wredundant-constraints -fhide-source-paths -threaded++  hs-source-dirs:+    test++  other-modules:++  default-extensions:+    DefaultSignatures+    LambdaCase+    OverloadedStrings+    RecordWildCards++  build-depends:+      kafka-client-sync++    , base                      >= 4.9      && < 5+    , hw-kafka-client           >= 2.6      && < 3.1+    , monad-parallel            >= 0.7      && < 0.8+    , text                      >= 1.2      && < 1.3
+ src/Kafka/Producer/Sync.hs view
@@ -0,0 +1,187 @@+-- | This module provides a synchronous interface on top of the hw-kafka-client+--+--   It works by using MVars managed in two different queues. Each request is+--   sent as soon as there are no other effectively equal Kafka records+--   in-flight. This is done in order to make sure that there is no ambiguity+--   as to which MVar to resolve.+--+--   Currently, this implements fair sending. For all requests, the oldest+--   pending request should be sent first.+--+module Kafka.Producer.Sync+  ( -- * Sync producer+    SyncKafkaProducer+  , newSyncProducer+  , produceRecord++    -- * Re-exports+  , KafkaError(..)+  , ProducerRecord(..)+  , ProducePartition(..)+  , TopicName(..)+  )+  where++import           Prelude++import           Control.Concurrent.MVar (MVar, newMVar, takeMVar, newEmptyMVar, putMVar)+import           Control.Monad.IO.Class (MonadIO(..))+import           Data.Foldable (find)+import           Data.Functor ((<&>))+import           Data.Maybe (isJust)+import           Data.Sequence (Seq(..), (<|), (|>))+import qualified Kafka.Producer as KP (deliveryCallback, flushProducer, newProducer)+import qualified Kafka.Producer as KP (produceMessage, setCallback)+import           Kafka.Producer.ProducerProperties (ProducerProperties)+import           Kafka.Producer.Types (KafkaProducer, ProducerRecord(..), DeliveryReport(..))+import           Kafka.Producer.Types (ProducePartition(..))+import           Kafka.Types (KafkaError(..), TopicName(..))++produceRecord :: MonadIO m => SyncKafkaProducer -> ProducerRecord -> m (Either KafkaError ())+produceRecord syncProducer record =+  -- Produce our message synchronously, then send pending:+  liftIO $ sendProducerRecord record syncProducer <* sendPending syncProducer+++-- | A producer for sending messages to Kafka and waiting for the 'DeliveryReport'+--+data SyncKafkaProducer = SyncKafkaProducer+  { requests :: MVar Requests+  , producer :: KafkaProducer+  }++-- | A variable containing the MVar that needs to be resolved in order for the+--   caller to proceed+--+type ResultVar = MVar (Either KafkaError ())++data Requests = Requests+  { pending :: Seq (ResultVar, ProducerRecord)+    -- ^ This sequence contains records that are effectively equal to something+    --   that is already being sent. In order for us not to have ambiguities on+    --   what 'MVar' to resolve on 'DeliverReport's - this separation is needed.+    --+    --   When a sent record has it's 'ResultVar' resolved, an effectively equal+    --   record is removed from this @pending@ queue and produced via the+    --   'KafkaProducer'. Once it's produced, it is moved to @sent@.+  , sent :: Seq (ResultVar, ProducerRecord)+    -- ^ This structure keeps track of in-flight messages sent to the Kafka+    --   broker, but not currently acknowledged. Once they are acknowledged+    --   via the callback action - they are removed from this structure and+    --   their 'ResultVar' is resolved.+  }++instance Show Requests where+  show Requests{..} =+    "Requests { pending = " <> show (snd <$> pending) <> ", sent = " <> show (snd <$> sent) <> " }"++-- | Create a new 'SyncKafkaProducer'+--+--   /Note/: since this library wraps the regular hw-kafka-client, please be+--   aware that you should not set the delivery report callback. As it is set+--   internally.+--+newSyncProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError SyncKafkaProducer)+newSyncProducer props = liftIO $ do+  reqs <- newMVar Requests { pending = mempty, sent = mempty }++  let+    -- A handler that removes requests from sent and resolves callbacks by+    -- putting to mvars:+    callbackAction =+      handleDeliveryReport reqs++    -- The regular hw-kafka-client KafkaProducer, with the sync callback handler:+    producer =+      KP.newProducer $ props <> KP.setCallback (KP.deliveryCallback callbackAction)++  producer <&> fmap (SyncKafkaProducer reqs)++-- | Sends one pending producer record+--+--   This will cause the completed send request to send an additional pending+--   record, and so on, eventually emptying the pending queue.+--+sendPending :: SyncKafkaProducer -> IO ()+sendPending SyncKafkaProducer{..} = do+  reqs <- takeMVar requests+  case pending reqs of+    (mvar, rec) :<| rest -> do+      KP.produceMessage producer rec >>= \case+        Just err -> putMVar mvar . Left $ err+        Nothing -> pure ()+      putMVar requests reqs { pending = rest, sent = sent reqs |> (mvar, rec) }+      KP.flushProducer producer+    Empty ->+      putMVar requests reqs++-- | Sends a producer record and waits for its delivery report+sendProducerRecord :: ProducerRecord -> SyncKafkaProducer -> IO (Either KafkaError ())+sendProducerRecord record SyncKafkaProducer{..} =+  takeMVar requests >>= \reqs ->+    if hasEffectivelyEqual record (sent reqs) then do+      var <- newEmptyMVar+      putMVar requests reqs { pending = pending reqs |> (var, record) }+      takeMVar var+    else KP.produceMessage producer record >>= \case+      Just err -> do+        putMVar requests reqs+        pure (Left err)+      Nothing -> do+        var <- newEmptyMVar+        putMVar requests reqs { sent = sent reqs |> (var, record) }+        KP.flushProducer producer+        takeMVar var++hasEffectivelyEqual :: ProducerRecord -> Seq (a, ProducerRecord) -> Bool+hasEffectivelyEqual record+  = isJust+  . find (effectivelyEqual record)+  . fmap snd++handleDeliveryReport :: MVar Requests -> (DeliveryReport -> IO ())+handleDeliveryReport mvarRequests = \case+  DeliverySuccess record _offset -> do+    reqs <- takeMVar mvarRequests+    case getAndRemove record (sent reqs) of+      Just (mvar, rest) -> do+        putMVar mvarRequests reqs { sent = rest }+        putMVar mvar $ Right ()+      Nothing ->+        error+          $ "Illegal state ocurred, record was not in sent: "+          <> show reqs+  DeliveryFailure record err -> do+    reqs <- takeMVar mvarRequests+    case getAndRemove record (sent reqs) of+      Just (mvar, rest) -> do+        putMVar mvarRequests reqs { sent = rest }+        putMVar mvar . Left $ err+      Nothing ->+        error+          $ "Illegal state ocurred, record was not in sent: "+          <> show reqs+  NoMessageError err ->+    error $ "Illegal state ocurred, NoMessageError received: " <> show err++getAndRemove ::+     ProducerRecord+  -> Seq (ResultVar, ProducerRecord)+  -> Maybe (ResultVar, Seq (ResultVar, ProducerRecord))+getAndRemove record xs =+  let+    splitRight acc = \case+      rest :|> current ->+        if snd current `effectivelyEqual` record then+          Just (fst current, rest <> acc)+        else+          splitRight (current <| acc) rest+      Empty -> Nothing+  in+    splitRight Empty xs++effectivelyEqual :: ProducerRecord -> ProducerRecord -> Bool+effectivelyEqual this other =+  prTopic this == prTopic other &&+  prKey this == prKey other &&+  prValue this == prValue other
+ test/Main.hs view
@@ -0,0 +1,46 @@+module Main (main) where++import           Prelude hiding (sequence)++import           Control.Monad (when)+import           Control.Monad.Parallel (sequence)+import           Data.Foldable (for_)+import           Data.Either (isLeft)+import           Data.Text (pack)+import           Data.Text.Encoding (encodeUtf8)+import           Kafka.Producer.Sync+import           Kafka.Producer++producerProps :: ProducerProperties+producerProps = brokersList [BrokerAddress "localhost:9092"]+  <> sendTimeout (Timeout 10000)+  <> logLevel KafkaLogDebug++message :: Int -> ProducerRecord+message suffix = ProducerRecord+  { prTopic = TopicName "topic"+  , prPartition = UnassignedPartition+  , prKey = Nothing+  , prValue = Just . encodeUtf8 . pack . show $ suffix+  }++main :: IO ()+main =+  newSyncProducer producerProps >>= \case+    Left _ -> error "Couldn't create sync producer"+    Right producer -> do+      putStrLn "Running Kafka sync tests"++      let action = produceRecord producer . message+      res <- sequence $+        fmap action [1..100] +++        fmap action (replicate 1 100)++      for_ res $+        either (putStrLn . (<>) "got error: " . show) pure++      when+        (any isLeft res)+        (error "Couldn't publish messages in parallel")++      putStrLn "Successfully published messages"