diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for hedgehog-servant + +## 0.1.0.0 -- 2019-11-10 + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/kafka-client-sync.cabal b/kafka-client-sync.cabal new file mode 100644 --- /dev/null +++ b/kafka-client-sync.cabal @@ -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 diff --git a/src/Kafka/Producer/Sync.hs b/src/Kafka/Producer/Sync.hs new file mode 100644 --- /dev/null +++ b/src/Kafka/Producer/Sync.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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"