diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -15,34 +15,41 @@ ```haskell main :: IO () -main = runPulsar resources $ \(Consumer {..}, Producer {..}) -> - let c = forever $ fetch >>= \(Message i m) -> msgDecoder m >> ack i - p = forever $ sleep 5 >> traverse_ produce messages - in concurrently_ c p +main = runPulsar conn $ do + c <- newConsumer topic sub + p <- newProducer topic + liftIO $ program c p -resources :: Pulsar (Consumer IO, Producer IO) -resources = do - ctx <- connect defaultConnectData - consumer <- newConsumer ctx topic "test-sub" - producer <- newProducer ctx topic - return (consumer, producer) +conn :: PulsarConnection +conn = connect defaultConnectData + +topic :: Topic +topic = defaultTopic "app" + +sub :: Subscription +sub = Subscription Exclusive "test-sub" + +program :: Consumer IO -> Producer IO -> IO () +program Consumer {..} Producer {..} = + let c = fetch >>= \(Message i m) -> msgDecoder m >> ack i >> c + p = sleep 3 >> traverse_ send messages >> p + in concurrently_ c p ``` A `Message` contains a `MessageID` you need for `ack`ing and a payload defined as a lazy `ByteString`. +> Note that we wait a few seconds before publishing a message to make sure the consumer is already subscribed. Otherwise, it might miss some messages. + Run it with the following command: ```shell cabal new-run supernova-tests ``` -By default, it logs to the standard output in DEBUG level. You can change it by suppling `LogOptions`. +By default, it logs to the standard output in DEBUG level. You can change it by suppling `LogOptions` to the alternative function `runPulsar'`. ```haskell -logOpts :: LogOptions -logOpts = LogOptions Info StdOut - -runPulsar' logOpts resources +runPulsar' :: LogOptions -> PulsarConnection -> Pulsar a -> IO () ``` ### Streaming @@ -53,10 +60,10 @@ import Streamly import qualified Streamly.Prelude as S -main :: IO () -main = runPulsar resources $ \(Consumer {..}, Producer {..}) -> - let c = forever $ fetch >>= \(Message i p) -> msgDecoder p >> ack i - p = forever $ sleep 5 >> traverse_ produce messages +program :: Consumer IO -> Producer IO -> IO () +program Consumer {..} Producer {..} = + let c = fetch >>= \(Message i m) -> msgDecoder m >> ack i >> c + p = sleep 3 >> traverse_ send messages >> p in S.drain . asyncly . maxThreads 10 $ S.yieldM c <> S.yieldM p ``` @@ -72,4 +79,17 @@ ```shell cabal new-build +``` + +#### Generate Hackage tarball + +```shell +cabal new-sdist +``` + +#### Upload documentation + +```shell +cabal new-haddock --haddock-for-hackage --enable-doc +cabal upload -d dist-newstyle/supernova-0.0.2-docs.tar.gz --publish ``` diff --git a/src/Pulsar.hs b/src/Pulsar.hs --- a/src/Pulsar.hs +++ b/src/Pulsar.hs @@ -5,35 +5,56 @@ Maintainer : gabriel.volpe@chatroulette.com Stability : experimental +In the following example, we will create a quick example showcasing a consumer and producer running concurrently, step by step. + Consider the following imports (needs the [async](http://hackage.haskell.org/package/async) library). @ import Control.Concurrent ( threadDelay ) import Control.Concurrent.Async ( concurrently_ ) -import Control.Monad ( forever ) import Pulsar @ -A quick example of a consumer and producer running concurrently. +Firstly, we create a connection to Pulsar, defined as 'PulsarConnection'. @ -resources :: Pulsar (Consumer IO, Producer IO) -resources = do - ctx <- connect defaultConnectData - consumer <- newConsumer ctx topic "test-sub" - producer <- newProducer ctx topic - return (consumer, producer) +conn :: PulsarConnection +conn = connect defaultConnectData @ -A Pulsar connection, consumers, and producers are long-lived resources that are managed accordingly for you. Once the program exits, the resources will be released in the respective order (always opposite to the order of acquisition). +Then a consumer and a producer, which operate in the 'Pulsar' monad. @ -main :: IO () -main = runPulsar resources $ \(Consumer {..}, Producer {..}) -> - let c = forever $ fetch >>= \(Message i m) -> print m >> ack i - p = forever $ threadDelay (5 * 1000000) >> produce "hello world" +pulsar :: Pulsar () +pulsar = do + c <- newConsumer topic sub + p <- newProducer topic + liftIO $ program c p + where + topic = defaultTopic "app" + sub = Subscription Exclusive "test-sub" +@ + +And the main user program that consume and produce messages concurrently, running in 'IO'. + +@ +program :: Consumer IO -> Producer IO -> IO () +program Consumer {..} Producer {..} = + let c = fetch >>= \(Message i m) -> print m >> ack i >> c + p = threadDelay (3 * 1000000) >> send "Hello World!" >> p in concurrently_ c p @ + +We have a delay of 3 seconds before publishing to make sure the consumer is already running. Otherwise, it might miss some messages. + +Finally, we put it all together and call 'runPulsar' with the connection and the program in the 'Pulsar' monad. + +@ +main :: IO () +main = runPulsar conn pulsar +@ + +Since a Pulsar connection, consumers, and producers are long-lived resources, Supernova manages them accordingly for you. Once the program exits, the resources will be released in the respective order (always opposite to the order of acquisition). -} module Pulsar ( connect @@ -45,8 +66,8 @@ , Consumer(..) , Producer(..) , Pulsar - , PulsarCtx - , ConnectData + , PulsarConnection + , ConnectData(..) , LogLevel(..) , LogOptions(..) , LogOutput(..) diff --git a/src/Pulsar/AppState.hs b/src/Pulsar/AppState.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/AppState.hs @@ -0,0 +1,99 @@ +{-# OPTIONS_GHC -Wno-missing-signatures #-} +{-# LANGUAGE GeneralizedNewtypeDeriving, TemplateHaskell #-} + +module Pulsar.AppState where + +import Control.Concurrent.Async ( Async ) +import Control.Concurrent.Chan +import Control.Concurrent.MVar +import Control.Monad.IO.Class +import qualified Data.Binary as B +import Data.Foldable ( traverse_ ) +import Data.IORef +import Lens.Family +import Lens.Family.TH +import Pulsar.Protocol.Frame ( Response(..) ) + +newtype ReqId = ReqId B.Word64 deriving (Eq, Num, Show) +newtype SeqId = SeqId B.Word64 deriving (Eq, Num, Show) +newtype ProducerId = PId B.Word64 deriving (Eq, Num, Show) +newtype ConsumerId = CId B.Word64 deriving (Eq, Num, Show) + +newtype Permits = Permits B.Word32 deriving (Eq, Num, Show) + +{- | It represents a running worker in the background along with a synchronizer. -} +type Worker = (Async (), MVar ()) + +{- | It represents a list of registered sequence ids for a producer. -} +type ProducerSeqs = [(SeqId, MVar Response)] + +data AppState = AppState + { _appConsumers :: [(ConsumerId, Chan Response)] -- a list of consumer identifiers associated with a communication channel + , _appConsumerId :: ConsumerId -- an incremental counter to assign unique consumer ids + , _appProducerId :: ProducerId -- an incremental counter to assign unique producer ids + , _appRequestId :: ReqId -- an incremental counter to assign unique request ids for all commands + , _appWorkers :: [Worker] -- a list of workers for consumers and producers that run in the background + , _appResponse :: [(ReqId, MVar Response)] -- a list of registered requests that need a Request Id + , _appSendReceipts :: [(ProducerId, ProducerSeqs)] -- a list of registered messages sent by a specific producer + } +$(makeLenses ''AppState) + +mkConsumerId :: MonadIO m => Chan Response -> IORef AppState -> m ConsumerId +mkConsumerId chan ref = liftIO $ atomicModifyIORef ref $ \app -> + let cid = app ^. appConsumerId + f = over appConsumers ((cid, chan) :) app + in (over appConsumerId (+ 1) f, cid) + +mkProducerId :: MonadIO m => IORef AppState -> m ProducerId +mkProducerId ref = liftIO $ atomicModifyIORef ref $ \app -> + let pid = app ^. appProducerId + f = over appSendReceipts ((pid, []) :) app + in (over appProducerId (+ 1) f, pid) + +mkRequestId :: MonadIO m => IORef AppState -> m (ReqId, MVar Response) +mkRequestId ref = liftIO $ do + var <- newEmptyMVar + atomicModifyIORef ref $ \app -> + let req = app ^. appRequestId + f = over appResponse ((req, var) :) app + in (over appRequestId (+ 1) f, (req, var)) + +addWorker :: MonadIO m => IORef AppState -> (Async (), MVar ()) -> m () +addWorker ref nw = + liftIO $ atomicModifyIORef ref $ \app -> (over appWorkers (nw :) app, ()) + +{- | Register a response for a request and unregister request. -} +registerReqResponse :: MonadIO m => IORef AppState -> ReqId -> Response -> m () +registerReqResponse ref rid resp = liftIO $ do + maybeVar <- atomicModifyIORef ref + $ \app -> (over appResponse h app, lookup rid $ app ^. appResponse) + traverse_ (`putMVar` resp) maybeVar + where h = filter ((rid /=) . fst) -- unregister request + +getProducerSeqs pid xs = filter (\(p, _) -> pid == p) xs >>= snd + +updateProducerSeqs pid g xs = + (\(p, ys) -> if p == pid then (p, g) else (p, ys)) <$> xs + +registerSeqId + :: MonadIO m => IORef AppState -> ProducerId -> SeqId -> m (MVar Response) +registerSeqId ref pid sid = liftIO $ do + var <- newEmptyMVar + atomicModifyIORef ref $ \app -> + let xs = app ^. appSendReceipts + g = (sid, var) : getProducerSeqs pid xs + h = updateProducerSeqs pid g xs + in (set appSendReceipts h app, var) + +{- | Register a response for a message sent and unregister sequence id. -} +registerSendReceipt + :: MonadIO m => IORef AppState -> ProducerId -> SeqId -> Response -> m () +registerSendReceipt ref pid sid resp = liftIO $ do + maybeVar <- atomicModifyIORef ref + $ \app -> (over appSendReceipts h app, g $ app ^. appSendReceipts) + traverse_ (`putMVar` resp) maybeVar + where + h xs = + let f = filter (\(s, _) -> s /= sid) $ getProducerSeqs pid xs + in updateProducerSeqs pid f xs + g = lookup sid . getProducerSeqs pid diff --git a/src/Pulsar/Connection.hs b/src/Pulsar/Connection.hs --- a/src/Pulsar/Connection.hs +++ b/src/Pulsar/Connection.hs @@ -1,12 +1,31 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving, LambdaCase, OverloadedStrings #-} +{-# LANGUAGE LambdaCase, OverloadedStrings #-} module Pulsar.Connection where -import Control.Monad ( forever ) -import Control.Monad.Catch ( MonadThrow ) -import Control.Monad.Managed -import qualified Data.Binary as B +import Control.Applicative ( (<|>) ) +import Control.Concurrent ( forkIO + , killThread + , threadDelay + ) +import Control.Concurrent.Async ( async + , concurrently_ + ) +import Control.Concurrent.Chan +import Control.Concurrent.MVar +import Control.Exception ( throwIO ) +import Control.Monad ( forever + , when + ) +import Control.Monad.Catch ( MonadThrow + , bracket + ) +import Control.Monad.IO.Class +import Control.Monad.Managed ( MonadManaged + , managed + , runManaged + ) import Data.Foldable ( traverse_ ) +import Data.Functor ( void ) import Data.IORef import Lens.Family import qualified Network.Socket as NS @@ -16,6 +35,7 @@ , MessageMetadata ) import qualified Proto.PulsarApi_Fields as F +import Pulsar.AppState import Pulsar.Internal.Logger import Pulsar.Internal.TCPClient ( acquireSocket ) import qualified Pulsar.Protocol.Commands as P @@ -27,52 +47,9 @@ , getCommand ) import System.Timeout ( timeout ) -import UnliftIO.Async ( concurrently_ ) -import UnliftIO.Chan -import UnliftIO.Concurrent ( forkIO - , killThread - , threadDelay - ) -import UnliftIO.Exception ( bracket - , throwIO - ) newtype Connection = Conn NS.Socket -newtype ReqId = ReqId B.Word64 deriving (Num, Show) -newtype SeqId = SeqId B.Word64 deriving (Num, Show) -newtype ProducerId = PId B.Word64 deriving (Num, Show) -newtype ConsumerId = CId B.Word64 deriving (Num, Show) - -data AppState = AppState - { appConsumers :: [(ConsumerId, Chan Response)] -- a list of consumer identifiers associated with a communication channel - , appConsumerId :: ConsumerId -- an incremental counter to assign unique consumer ids - , appProducers :: [(ProducerId, Chan Response)] -- a list of producer identifiers associated with a communication channel - , appProducerId :: ProducerId -- an incremental counter to assign unique producer ids - , appRequestId :: ReqId -- an incremental counter to assign unique request ids for all commands - } - -mkConsumerId :: MonadIO m => Chan Response -> IORef AppState -> m ConsumerId -mkConsumerId chan ref = liftIO $ atomicModifyIORef - ref - (\(AppState cs cid ps pid rid) -> - let cid' = cid + 1 in (AppState ((cid', chan) : cs) cid' ps pid rid, cid) - ) - -mkProducerId :: MonadIO m => Chan Response -> IORef AppState -> m ProducerId -mkProducerId chan ref = liftIO $ atomicModifyIORef - ref - (\(AppState cs cid ps pid rid) -> - let pid' = pid + 1 in (AppState cs cid ((pid', chan) : ps) pid' rid, pid) - ) - -mkRequestId :: MonadIO m => IORef AppState -> m ReqId -mkRequestId ref = liftIO $ atomicModifyIORef - ref - (\(AppState cs cid ps pid req) -> - let req' = req + 1 in (AppState cs cid ps pid req', req) - ) - {- | Connection details: host and port. -} data ConnectData = ConnData { connHost :: NS.HostName @@ -83,6 +60,7 @@ data PulsarCtx = Ctx { ctxConn :: Connection , ctxState :: IORef AppState + , ctxConnWorker :: Worker } {- | Default connection data: "127.0.0.1:6650" -} @@ -91,42 +69,82 @@ {- | Starts a Pulsar connection with the supplied 'ConnectData' -} connect - :: (MonadThrow m, MonadIO m, MonadManaged m) => ConnectData -> m PulsarCtx + :: (MonadIO m, MonadThrow m, MonadManaged m) => ConnectData -> m PulsarCtx connect (ConnData h p) = do socket <- acquireSocket h p liftIO $ sendSimpleCmd socket P.connect + checkConnection socket + app <- liftIO initAppState + kchan <- liftIO newChan + var <- liftIO newEmptyMVar + let + dispatcher = recvDispatch socket app kchan + task = concurrently_ dispatcher (keepAlive socket kchan) + handler = + managed (bracket (forkIO task) (\i -> readMVar var >> killThread i)) + worker <- liftIO $ async (runManaged $ void handler) + return $ Ctx (Conn socket) app (worker, var) + +checkConnection :: (MonadIO m, MonadThrow m) => NS.Socket -> m () +checkConnection socket = do resp <- receive socket case getCommand resp ^. F.maybe'connected of Just _ -> logResponse resp - Nothing -> throwIO $ userError "Could not connect" - app <- liftIO initAppState - kchan <- liftIO newChan - let ctx = Ctx (Conn socket) app - dispatcher = recvDispatch socket app kchan - task = concurrently_ dispatcher (keepAlive socket kchan) - using $ ctx <$ managed (bracket (forkIO task) killThread) + Nothing -> liftIO . throwIO $ userError "Could not connect" initAppState :: MonadIO m => m (IORef AppState) -initAppState = liftIO . newIORef $ AppState [] 0 [] 0 0 +initAppState = liftIO . newIORef $ AppState [] 0 0 0 [] [] [] -recvDispatch - :: MonadIO m => NS.Socket -> IORef AppState -> Chan BaseCommand -> m () +responseForRequest :: BaseCommand -> Maybe ReqId +responseForRequest cmd = + let cmd1 = view F.requestId <$> cmd ^. F.maybe'success + cmd2 = view F.requestId <$> cmd ^. F.maybe'producerSuccess + cmd3 = view F.requestId <$> cmd ^. F.maybe'lookupTopicResponse + in ReqId <$> (cmd1 <|> cmd2 <|> cmd3) + +responseForSendReceipt :: BaseCommand -> Maybe (ProducerId, SeqId) +responseForSendReceipt cmd = + let cmd' = cmd ^. F.maybe'sendReceipt + pid = PId . view F.producerId <$> cmd' + sid = SeqId . view F.sequenceId <$> cmd' + in (,) <$> pid <*> sid + +pongResponse :: BaseCommand -> Chan BaseCommand -> IO (Maybe ()) +pongResponse cmd chan = + traverse (const $ writeChan chan cmd) (cmd ^. F.maybe'pong) + +messageResponse :: BaseCommand -> Maybe ConsumerId +messageResponse cmd = + let cmd' = cmd ^. F.maybe'message + cid = view F.consumerId <$> cmd' + in CId <$> cid + +{- | It listens to incoming messages directly from the network socket and it writes them to all the + - consumers and producers' communication channels. -} +recvDispatch :: NS.Socket -> IORef AppState -> Chan BaseCommand -> IO () recvDispatch s ref chan = forever $ do - resp <- receive s - (AppState cs _ ps _ _) <- liftIO $ readIORef ref - case getCommand resp ^. F.maybe'pong of - Just _ -> writeChan chan (getCommand resp) - Nothing -> traverse_ (`writeChan` resp) ((snd <$> cs) ++ (snd <$> ps)) + resp <- receive s + cs <- _appConsumers <$> readIORef ref + let + f = \rid -> registerReqResponse ref rid resp + g = (\(pid, sid) -> registerSendReceipt ref pid sid resp) + h = \cid -> + traverse (\(cid', cn) -> when (cid == cid') (writeChan cn resp)) cs + cmd = getCommand resp + traverse_ f (responseForRequest cmd) + traverse_ g (responseForSendReceipt cmd) + traverse_ h (messageResponse cmd) + pongResponse cmd chan {- Emit a PING and expect a PONG every 29 seconds. If a PONG is not received, interrupt connection -} -keepAlive :: MonadIO m => NS.Socket -> Chan BaseCommand -> m () +keepAlive :: NS.Socket -> Chan BaseCommand -> IO () keepAlive s chan = forever $ do threadDelay (29 * 1000000) logRequest P.ping sendSimpleCmd s P.ping - liftIO $ timeout (2 * 1000000) (readChan chan) >>= \case + timeout (2 * 1000000) (readChan chan) >>= \case Just cmd -> logResponse cmd - Nothing -> throwIO (userError "Keep Alive interruption") + Nothing -> throwIO $ userError "Keep Alive interruption" sendSimpleCmd :: MonadIO m => NS.Socket -> BaseCommand -> m () sendSimpleCmd s cmd = diff --git a/src/Pulsar/Consumer.hs b/src/Pulsar/Consumer.hs --- a/src/Pulsar/Consumer.hs +++ b/src/Pulsar/Consumer.hs @@ -1,23 +1,71 @@ -{-# LANGUAGE LambdaCase, OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts, LambdaCase, OverloadedStrings #-} -module Pulsar.Consumer where +{- | +Module : Pulsar.Consumer +Description : Apache Pulsar client +License : Apache-2.0 +Maintainer : gabriel.volpe@chatroulette.com +Stability : experimental -import Control.Monad ( forever ) -import qualified Control.Monad.Catch as E -import Control.Monad.Managed -import Lens.Family +The basic consumer interaction looks as follows: http://pulsar.apache.org/docs/en/develop-binary-protocol/#consumer + +>>> LOOKUP +<<< LOOKUP_RESPONSE +>>> SUBSCRIBE +<<< SUCCESS +>>> FLOW 1000 +<<< MESSAGE 1 +<<< MESSAGE 2 +>>> ACK 1 +>>> ACK 2 + +When half of the messages have been consumed from our internal queue (Chan), we ask the broker to send more events and continue processing events. + +>>> FLOW 500 + +When the program finishes, either succesfully or due to a failure, we unsubscribe and close the consumer. + +>>> CLOSE_CONSUMER +<<< SUCCESS +-} +module Pulsar.Consumer + ( Consumer(..) + , newConsumer + ) +where + +import Control.Concurrent ( forkIO + , killThread + ) +import Control.Concurrent.Async ( async ) +import Control.Concurrent.Chan +import Control.Concurrent.MVar +import Control.Monad ( forever + , when + ) +import Control.Monad.Catch ( bracket ) +import Control.Monad.IO.Class ( MonadIO + , liftIO + ) +import Control.Monad.Managed ( managed + , runManaged + ) +import Control.Monad.Reader ( MonadReader + , ask + ) +import Data.Foldable ( for_ ) +import Data.IORef +import Data.Functor ( void ) +import Lens.Family hiding ( reset ) import qualified Proto.PulsarApi_Fields as F import qualified Pulsar.Core as C -import Pulsar.Connection +import Pulsar.AppState +import Pulsar.Connection ( PulsarCtx(..) ) import Pulsar.Internal.Logger ( logResponse ) import Pulsar.Protocol.Frame ( Payload(..) , Response(..) ) import Pulsar.Types -import UnliftIO.Chan -import UnliftIO.Concurrent ( forkIO - , killThread - ) {- | An abstract 'Consumer' able to 'fetch' messages and 'ack'nowledge them. -} data Consumer m = Consumer @@ -25,36 +73,65 @@ , ack :: MsgId -> m () -- ^ Acknowledges a single message. } +{- | The protocol expects the implementation to use some kind of queue to store events sent by the broker. -} +defaultQueueSize :: Int +defaultQueueSize = 1000 + +{- | It keeps track of the size of our internal messages queue . -} +updateQueueSize :: IORef Int -> (Int -> Int) -> IO () +updateQueueSize ref f = atomicModifyIORef ref (\x -> (f x, ())) + {- | Create a new 'Consumer' by supplying a 'PulsarCtx' (returned by 'Pulsar.connect'), a 'Topic' and a 'SubscriptionName'. -} newConsumer - :: (MonadManaged m, MonadIO f) - => PulsarCtx - -> Topic - -> SubscriptionName + :: (MonadIO m, MonadIO f, MonadReader PulsarCtx m) + => Topic + -> Subscription -> m (Consumer f) -newConsumer (Ctx conn app) topic sub = do - chan <- newChan - cid <- mkConsumerId chan app - fchan <- newChan - using $ Consumer (readChan fchan) (acker cid) <$ managed - (E.bracket - (mkSubscriber chan cid >> forkIO (fetcher chan fchan)) - (\i -> newReq >>= \r -> C.closeConsumer conn chan r cid >> killThread i) - ) +newConsumer topic sub = do + (Ctx conn app _) <- ask + chan <- liftIO newChan + cid <- mkConsumerId chan app + fchan <- liftIO newChan + ref <- liftIO $ newIORef 0 + var <- liftIO newEmptyMVar + let permits = issuePermits conn cid + acquire = do + mkSubscriber conn cid app + forkIO (fetcher chan fchan ref permits) + release i = + killThread i >> newReq app >>= \(r, v) -> C.closeConsumer conn v cid r + handler = managed (bracket acquire release) >> liftIO (readMVar var) + worker <- liftIO $ async (runManaged $ void handler) + addWorker app (worker, var) + return $ Consumer (liftIO $ readChan fchan) (acker conn cid) where - fetcher chan fc = liftIO . forever $ readChan chan >>= \case - PayloadResponse cmd _ p -> case cmd ^. F.maybe'message of - Just msg -> - let msgId = msg ^. F.messageId - pm = Message (MsgId msgId) $ maybe "" (\(Payload x) -> x) p - in logResponse cmd >> writeChan fc pm - Nothing -> return () - _ -> return () - newReq = mkRequestId app - acker cid (MsgId mid) = liftIO $ C.ack conn cid mid - mkSubscriber chan cid = do - req1 <- newReq - C.lookup conn chan req1 topic - req2 <- newReq - C.newSubscriber conn chan req2 cid topic sub - C.flow conn cid + newReq app = mkRequestId app + acker conn cid (MsgId mid) = liftIO $ C.ack conn cid mid + issuePermits conn cid = + C.flow conn cid (Permits $ fromIntegral (defaultQueueSize `div` 2)) + mkSubscriber conn cid app = do + (req1, var1) <- newReq app + C.lookup conn var1 req1 topic + (req2, var2) <- newReq app + C.newSubscriber conn var2 req2 cid topic sub + C.flow conn cid (Permits $ fromIntegral defaultQueueSize) + +{- | It reads responses from the main communication channel and whenever it corresponds to a + - 'PayloadResponse', it creates a 'Message' and it writes it to the fetcher channel, which + - is the one the 'fetch' function is listening on. + - + - It also keeps count of the internal fetcher channel size and issues new permits (FLOW) + - whenever necessary. + -} +fetcher :: Chan Response -> Chan Message -> IORef Int -> IO a -> IO b +fetcher chan fc ref f = forever $ readChan chan >>= \case + PayloadResponse cmd _ p -> for_ (cmd ^. F.maybe'message) $ \msg -> do + let msgId = msg ^. F.messageId + pm = Message (MsgId msgId) $ maybe "" (\(Payload x) -> x) p + reset = updateQueueSize ref ((defaultQueueSize `div` 2) -) + logResponse cmd + updateQueueSize ref (+ 1) + size <- readIORef ref + when (size >= defaultQueueSize `div` 2) (f >> reset) + writeChan fc pm + _ -> return () diff --git a/src/Pulsar/Core.hs b/src/Pulsar/Core.hs --- a/src/Pulsar/Core.hs +++ b/src/Pulsar/Core.hs @@ -1,18 +1,18 @@ -{-# LANGUAGE DataKinds, FlexibleContexts, LambdaCase, OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings #-} {- Defines a set of transactional commands, communicating via internal channels -} module Pulsar.Core where +import Control.Concurrent.Chan +import Control.Concurrent.MVar import Control.Exception ( throwIO ) import Control.Monad.Catch ( MonadThrow ) import Control.Monad.IO.Class -import qualified Data.Binary as B -import Data.Functor ( void ) -import Data.ProtoLens.Field ( HasField ) import Data.Text ( Text ) import Lens.Family -import Proto.PulsarApi +import Proto.PulsarApi hiding ( Subscription ) import qualified Proto.PulsarApi_Fields as F +import Pulsar.AppState import Pulsar.Connection import Pulsar.Internal.Logger import qualified Pulsar.Protocol.Commands as P @@ -21,82 +21,62 @@ , getCommand ) import Pulsar.Types -import UnliftIO.Chan ------ Simple commands ------ -verifyResponse - :: (HasField a "requestId" B.Word64, Show a) - => ReqId - -> Chan Response - -> LensLike' (Constant (Maybe a)) BaseCommand (Maybe a) - -> IO (Maybe a) -verifyResponse r@(ReqId req) chan lens = do - resp <- readChan chan - let cmd' = getCommand resp ^. lens - req' = view F.requestId <$> cmd' - rewrite = writeChan chan resp - loop = verifyResponse r chan lens - checkEq (_, rq) | rq == req = cmd' <$ logResponse resp - | otherwise = rewrite >> loop - maybe loop checkEq $ (,) <$> cmd' <*> req' - -lookup :: Connection -> Chan Response -> ReqId -> Topic -> IO () -lookup (Conn s) chan r@(ReqId req) topic = do +lookup :: Connection -> MVar Response -> ReqId -> Topic -> IO () +lookup (Conn s) var (ReqId req) topic = do logRequest $ P.lookup req topic sendSimpleCmd s $ P.lookup req topic -- TODO: we need to analyze it and might need to re-issue another lookup - void $ verifyResponse r chan F.maybe'lookupTopicResponse + readMVar var >>= logResponse newProducer - :: Connection -> Chan Response -> ReqId -> ProducerId -> Topic -> IO Text -newProducer (Conn s) chan r@(ReqId req) (PId pid) topic = do + :: Connection -> MVar Response -> ReqId -> ProducerId -> Topic -> IO Text +newProducer (Conn s) var (ReqId req) (PId pid) topic = do logRequest $ P.producer req pid topic sendSimpleCmd s $ P.producer req pid topic - verifyResponse r chan F.maybe'producerSuccess >>= \case + resp <- readMVar var + logResponse resp + case getCommand resp ^. F.maybe'producerSuccess of Just ps -> return $ ps ^. F.producerName Nothing -> return "" -closeProducer :: Connection -> Chan Response -> ReqId -> ProducerId -> IO () -closeProducer (Conn s) chan r@(ReqId req) (PId pid) = do +closeProducer :: Connection -> MVar Response -> ProducerId -> ReqId -> IO () +closeProducer (Conn s) var (PId pid) (ReqId req) = do logRequest $ P.closeProducer req pid sendSimpleCmd s $ P.closeProducer req pid - void $ verifyResponse r chan F.maybe'success + readMVar var >>= logResponse newSubscriber :: Connection - -> Chan Response + -> MVar Response -> ReqId -> ConsumerId -> Topic - -> SubscriptionName + -> Subscription -> IO () -newSubscriber (Conn s) chan r@(ReqId req) (CId cid) topic subs = do - logRequest $ P.subscribe req cid topic subs - sendSimpleCmd s $ P.subscribe req cid topic subs - -- TODO: we may need to check for failure too - void $ verifyResponse r chan F.maybe'success +newSubscriber (Conn s) var (ReqId req) (CId cid) topic (Subscription stype sname) + = do + logRequest $ P.subscribe req cid topic stype sname + sendSimpleCmd s $ P.subscribe req cid topic stype sname + readMVar var >>= logResponse -flow :: Connection -> ConsumerId -> IO () -flow (Conn s) (CId cid) = do - logRequest $ P.flow cid - sendSimpleCmd s $ P.flow cid +flow :: Connection -> ConsumerId -> Permits -> IO () +flow (Conn s) (CId cid) (Permits p) = do + logRequest $ P.flow cid p + sendSimpleCmd s $ P.flow cid p ack :: MonadIO m => Connection -> ConsumerId -> MessageIdData -> m () ack (Conn s) (CId cid) msgId = do logRequest $ P.ack cid msgId sendSimpleCmd s $ P.ack cid msgId -closeConsumer :: Connection -> Chan Response -> ReqId -> ConsumerId -> IO () -closeConsumer (Conn s) _ (ReqId req) (CId cid) = do +closeConsumer :: Connection -> MVar Response -> ConsumerId -> ReqId -> IO () +closeConsumer (Conn s) var (CId cid) (ReqId req) = do logRequest $ P.closeConsumer req cid sendSimpleCmd s $ P.closeConsumer req cid - -- FIXME: this is a workaround but the problem is the response for close consumer never comes on a SIGTERM when consuming - -- from the Chan, since the writer gets interrupted and no messages come in. - resp <- receive s - case getCommand resp ^. F.maybe'success of - Just _ -> logResponse resp - Nothing -> return () + readMVar var >>= logResponse ------ Keep Alive ------- @@ -104,7 +84,7 @@ ping (Conn s) chan = do logRequest P.ping sendSimpleCmd s P.ping - cmd <- getCommand <$> readChan chan + cmd <- getCommand <$> liftIO (readChan chan) case cmd ^. F.maybe'pong of Just p -> logResponse p Nothing -> liftIO . throwIO $ userError "Failed to get PONG" @@ -118,23 +98,12 @@ send :: Connection - -> Chan Response + -> MVar Response -> ProducerId -> SeqId -> PulsarMessage -> IO () -send (Conn s) chan (PId pid) (SeqId sid) (PulsarMessage msg) = do +send (Conn s) var (PId pid) (SeqId sid) (PulsarMessage msg) = do logRequest $ P.send pid sid sendPayloadCmd s (P.send pid sid) P.messageMetadata (Just $ Payload msg) - confirmReception - where - confirmReception = do - resp <- readChan chan - let cmd' = getCommand resp ^. F.maybe'sendReceipt - pid' = view F.producerId <$> cmd' - sid' = view F.sequenceId <$> cmd' - rewrite = writeChan chan resp - loop = confirmReception - checkEq (_, pd, sd) | pd == pid && sd == sid = logResponse resp - | otherwise = rewrite >> loop - maybe loop checkEq $ (,,) <$> cmd' <*> pid' <*> sid' + readMVar var >>= logResponse diff --git a/src/Pulsar/Internal/Core.hs b/src/Pulsar/Internal/Core.hs --- a/src/Pulsar/Internal/Core.hs +++ b/src/Pulsar/Internal/Core.hs @@ -1,33 +1,58 @@ -{-# LANGUAGE RankNTypes, GeneralizedNewtypeDeriving #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{- Defines a Pulsar Monad, which wraps a Managed resource -} +{- Defines a Pulsar Monad, which wraps a ReaderT and runs internal computations in the background -} module Pulsar.Internal.Core where +import Control.Concurrent.Async ( cancel ) +import Control.Concurrent.MVar import qualified Control.Logging as L -import Control.Monad.Catch +import Control.Monad.Catch ( MonadThrow + , finally + , throwM + ) import Control.Monad.Managed +import Control.Monad.Reader +import Data.Foldable ( traverse_ ) +import Data.IORef ( readIORef ) +import Pulsar.AppState ( AppState(..) ) +import Pulsar.Connection ( PulsarCtx(..) ) -{- | The main Pulsar monad, which abstracts over a 'Managed' monad. -} -newtype Pulsar a = Pulsar (Managed a) +{- | Pulsar connection monad, which abstracts over a 'Managed' monad. -} +newtype Connection a = Connection (Managed a) deriving (Functor, Applicative, Monad, MonadIO, MonadManaged) +instance MonadThrow Connection where + throwM = liftIO . throwM + +{- | Alias for Connection PulsarCtx. -} +type PulsarConnection = Connection PulsarCtx + +{- | The main Pulsar monad, which abstracts over a 'ReaderT' monad. -} +newtype Pulsar a = Pulsar (ReaderT PulsarCtx IO a) + deriving (Functor, Applicative, Monad, MonadIO, MonadReader PulsarCtx) + {- | Runs a Pulsar computation with default logging to standard output -} -runPulsar :: forall a b . Pulsar a -> (a -> IO b) -> IO b -runPulsar (Pulsar mgd) f = do - L.setLogTimeFormat "%H:%M:%S%Q" - L.withStdoutLogging $ with mgd f +runPulsar :: PulsarConnection -> Pulsar a -> IO () +runPulsar = runPulsar' (LogOptions Debug StdOut) {- | Runs a Pulsar computation with the supplied logging options -} -runPulsar' :: forall a b . LogOptions -> Pulsar a -> (a -> IO b) -> IO b -runPulsar' (LogOptions lvl out) (Pulsar mgd) f = do - L.setLogLevel $ convertLogLevel lvl +runPulsar' :: LogOptions -> PulsarConnection -> Pulsar a -> IO () +runPulsar' (LogOptions lvl out) (Connection mgd) (Pulsar mr) = do + L.setLogLevel $ fromLogLevel lvl L.setLogTimeFormat "%H:%M:%S%Q" case out of - StdOut -> L.withStdoutLogging $ with mgd f - File fp -> L.withFileLogging fp $ with mgd f - -instance MonadThrow Pulsar where - throwM = liftIO . throwM + StdOut -> L.withStdoutLogging runner + File fp -> L.withFileLogging fp runner + where + runner = runManaged $ do + ctx <- mgd + void . liftIO $ runReaderT mr ctx `finally` finalizers ctx + finalizers ctx = do + let (worker, connVar) = ctxConnWorker ctx + app <- readIORef (ctxState ctx) + traverse_ (\(a, v) -> putMVar v () >> cancel a) (_appWorkers app) + `finally` putMVar connVar () + cancel worker {- | Internal logging options. Can be used together with `runPulsar'`. -} data LogOptions = LogOptions @@ -41,8 +66,8 @@ {- | Internal logging output, part of 'LogOptions'. Can be used together with `runPulsar'`. -} data LogOutput = StdOut | File FilePath deriving Show -convertLogLevel :: LogLevel -> L.LogLevel -convertLogLevel Error = L.LevelError -convertLogLevel Warn = L.LevelWarn -convertLogLevel Info = L.LevelInfo -convertLogLevel Debug = L.LevelDebug +fromLogLevel :: LogLevel -> L.LogLevel +fromLogLevel Error = L.LevelError +fromLogLevel Warn = L.LevelWarn +fromLogLevel Info = L.LevelInfo +fromLogLevel Debug = L.LevelDebug diff --git a/src/Pulsar/Internal/TCPClient.hs b/src/Pulsar/Internal/TCPClient.hs --- a/src/Pulsar/Internal/TCPClient.hs +++ b/src/Pulsar/Internal/TCPClient.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE LambdaCase, OverloadedStrings #-} +{-# LANGUAGE LambdaCase, OverloadedStrings, RankNTypes #-} {- A simple TCP client, used to communicate with the Pulsar server -} module Pulsar.Internal.TCPClient @@ -6,26 +6,31 @@ ) where -import qualified Control.Exception as E -import Control.Monad.IO.Class -import Control.Monad.Managed +import Control.Monad.Catch ( bracket + , bracketOnError + ) +import Control.Monad.IO.Class ( MonadIO + , liftIO + ) +import Control.Monad.Managed ( MonadManaged + , managed + , using + ) import qualified Network.Socket as NS acquireSocket :: (MonadIO m, MonadManaged m) => NS.HostName -> NS.ServiceName -> m NS.Socket acquireSocket host port = do addr <- liftIO resolve - using $ managed - (E.bracket - (putStrLn "[ Establishing connection with Pulsar ]" >> open addr) - (\s -> putStrLn "[ Closing Pulsar connection ]" >> NS.close s) - ) + using $ managed (bracket (acquire addr) release) where + acquire = (putStrLn "[ Establishing connection with Pulsar ]" >>) . open + release = (putStrLn "[ Closing Pulsar connection ]" >>) . NS.close resolve = do let hints = NS.defaultHints { NS.addrSocketType = NS.Stream } NS.getAddrInfo (Just hints) (Just host) (Just port) >>= \case [addr] -> pure addr - _ -> E.ioError $ userError "Could not resolve socket address" - open addr = E.bracketOnError (NS.openSocket addr) NS.close $ \sock -> do + _ -> ioError $ userError "Could not resolve socket address" + open addr = bracketOnError (NS.openSocket addr) NS.close $ \sock -> do NS.connect sock $ NS.addrAddress addr return sock diff --git a/src/Pulsar/Producer.hs b/src/Pulsar/Producer.hs --- a/src/Pulsar/Producer.hs +++ b/src/Pulsar/Producer.hs @@ -1,18 +1,52 @@ -{- Defines a high-level Pulsar producer for the end user -} +{-# LANGUAGE FlexibleContexts #-} + +{- | +Module : Pulsar.Producer +Description : Apache Pulsar client +License : Apache-2.0 +Maintainer : gabriel.volpe@chatroulette.com +Stability : experimental + +The basic producer interaction looks as follows: http://pulsar.apache.org/docs/en/develop-binary-protocol/#producer + +>>> LOOKUP +<<< LOOKUP_RESPONSE +>>> PRODUCER +<<< SUCCESS +>>> SEND 1 +>>> SEND 2 +<<< SEND_RECEIPT 1 +<<< SEND_RECEIPT 2 + +When the program finishes, either succesfully or due to a failure, we close the producer. + +>>> CLOSE_PRODUCER +<<< SUCCESS +-} module Pulsar.Producer where -import qualified Control.Monad.Catch as E -import Control.Monad.Managed +import Control.Concurrent.Async ( async ) +import Control.Monad.Catch ( bracket_ ) +import Control.Concurrent.MVar +import Control.Monad.IO.Class ( MonadIO + , liftIO + ) +import Control.Monad.Managed ( managed_ + , runManaged + ) +import Control.Monad.Reader ( MonadReader + , ask + ) import Data.IORef import Data.Text ( Text ) +import Pulsar.AppState import qualified Pulsar.Core as C -import Pulsar.Connection +import Pulsar.Connection ( PulsarCtx(..) ) import Pulsar.Types -import UnliftIO.Chan -{- | An abstract 'Producer' able to 'produce' messages of type 'PulsarMessage'. -} +{- | An abstract 'Producer' able to 'send' messages of type 'PulsarMessage'. -} newtype Producer m = Producer - { produce :: PulsarMessage -> m () -- ^ Produces a single message. + { send :: PulsarMessage -> m () -- ^ Produces a single message. } data ProducerState = ProducerState @@ -27,23 +61,26 @@ {- | Create a new 'Producer' by supplying a 'PulsarCtx' (returned by 'Pulsar.connect') and a 'Topic'. -} newProducer - :: (MonadManaged m, MonadIO f) => PulsarCtx -> Topic -> m (Producer f) -newProducer (Ctx conn app) topic = do - chan <- newChan - pid <- mkProducerId chan app - pname <- liftIO $ mkProducer chan pid - pst <- liftIO $ newIORef (ProducerState 0 pname) - using $ managed - (E.bracket (pure $ Producer (dispatch chan pid pst)) - (const $ newReq >>= \r -> C.closeProducer conn chan r pid) - ) + :: (MonadIO m, MonadReader PulsarCtx m, MonadIO f) => Topic -> m (Producer f) +newProducer topic = do + (Ctx conn app _) <- ask + pid <- mkProducerId app + pname <- liftIO $ mkProducer conn pid app + pst <- liftIO $ newIORef (ProducerState 0 pname) + var <- liftIO newEmptyMVar + let release = newReq app >>= \(r, v) -> C.closeProducer conn v pid r + handler = managed_ (bracket_ (pure ()) release) >> liftIO (readMVar var) + worker <- liftIO $ async (runManaged handler) + addWorker app (worker, var) + return $ Producer (dispatch conn pid app pst) where - newReq = mkRequestId app - dispatch chan pid pst msg = do + newReq app = mkRequestId app + dispatch conn pid app pst msg = do sid <- mkSeqId pst - liftIO $ C.send conn chan pid sid msg - mkProducer chan pid = do - req1 <- newReq - C.lookup conn chan req1 topic - req2 <- newReq - C.newProducer conn chan req2 pid topic + var <- registerSeqId app pid sid + liftIO $ C.send conn var pid sid msg + mkProducer conn pid app = do + (req1, var1) <- newReq app + C.lookup conn var1 req1 topic + (req2, var2) <- newReq app + C.newProducer conn var2 req2 pid topic diff --git a/src/Pulsar/Protocol/CheckSum.hs b/src/Pulsar/Protocol/CheckSum.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Protocol/CheckSum.hs @@ -0,0 +1,16 @@ +module Pulsar.Protocol.CheckSum where + +import qualified Data.Binary as B +import Data.Bool ( bool ) +import qualified Data.ByteString.Lazy.Char8 as CL +import Data.Digest.CRC32C ( crc32c ) + +data CheckSumValidation = Valid | Invalid deriving Show + +newtype CheckSum = CheckSum B.Word32 deriving (Eq, Show) + +runCheckSum :: CL.ByteString -> CheckSum -> CheckSumValidation +runCheckSum t cs = bool Invalid Valid $ computeCheckSum t == cs + +computeCheckSum :: CL.ByteString -> CheckSum +computeCheckSum t = CheckSum $ crc32c (CL.toStrict t) diff --git a/src/Pulsar/Protocol/Commands.hs b/src/Pulsar/Protocol/Commands.hs --- a/src/Pulsar/Protocol/Commands.hs +++ b/src/Pulsar/Protocol/Commands.hs @@ -22,27 +22,33 @@ & F.clientVersion .~ "Pulsar-Client-Haskell-v" <> T.pack (showVersion version) & F.protocolVersion .~ 15 -subscribe :: B.Word64 -> B.Word64 -> Topic -> SubscriptionName -> BaseCommand -subscribe req cid topic (SubscriptionName sub) = defMessage +subType :: SubType -> CommandSubscribe'SubType +subType Exclusive = CommandSubscribe'Exclusive +subType Shared = CommandSubscribe'Shared +subType Failover = CommandSubscribe'Failover +subType KeyShared = CommandSubscribe'Key_Shared + +subscribe :: B.Word64 -> B.Word64 -> Topic -> SubType -> SubName -> BaseCommand +subscribe req cid topic stype (SubName sname) = defMessage & F.type' .~ BaseCommand'SUBSCRIBE & F.subscribe .~ subs where subs :: CommandSubscribe subs = defMessage & F.topic .~ T.pack (show topic) - & F.subscription .~ sub - & F.subType .~ CommandSubscribe'Shared + & F.subscription .~ sname + & F.subType .~ subType stype & F.consumerId .~ cid & F.requestId .~ req -flow :: B.Word64 -> BaseCommand -flow cid = defMessage +flow :: B.Word64 -> B.Word32 -> BaseCommand +flow cid permits = defMessage & F.type' .~ BaseCommand'FLOW & F.flow .~ flowCmd where flowCmd :: CommandFlow flowCmd = defMessage - & F.messagePermits .~ 100 + & F.messagePermits .~ permits & F.consumerId .~ cid ack :: B.Word64 -> MessageIdData -> BaseCommand diff --git a/src/Pulsar/Protocol/Decoder.hs b/src/Pulsar/Protocol/Decoder.hs --- a/src/Pulsar/Protocol/Decoder.hs +++ b/src/Pulsar/Protocol/Decoder.hs @@ -7,16 +7,19 @@ ) where -import Control.Monad ( unless ) +import Control.Monad ( guard ) +import qualified Data.Binary as B import qualified Data.Binary.Get as B import qualified Data.Binary.Put as B import qualified Data.ByteString.Lazy.Char8 as CL -import Data.Digest.CRC32C ( crc32c ) import Data.Bifunctor ( bimap ) import Data.Int ( Int32 ) import qualified Data.ProtoLens.Encoding as PL +import Pulsar.Protocol.CheckSum import Pulsar.Protocol.Frame +data ValidateCheckSum = Yes | No deriving Show + {- - These 5 bytes are part of a total of 8 bytes sent as the payload's prefix from the Java client. - Apparently that's how Google's FlatBuffers serialize data: https://google.github.io/flatbuffers/ @@ -29,52 +32,60 @@ dropPayloadGarbage bs = maybe bs (CL.drop 3) (CL.stripPrefix "\NUL\NUL\NUL\EOT\CAN" bs) +{- | Parse total size, command size and message. If done, return simple frame. Otherwise, try to parse a payload frame. -} parseFrame :: B.Get Frame parseFrame = do ts <- B.getInt32be cs <- B.getInt32be ms <- B.getLazyByteString (fromIntegral cs) - let simpleCmd = SimpleCommand ts cs ms + let simpleCmd = SimpleCommand ts cs ms + payloadRes = parsePayload ts cs simpleCmd B.isEmpty >>= \case - True -> return $ SimpleFrame simpleCmd - False -> parsePayload ts cs simpleCmd + True -> return $! SimpleFrame simpleCmd + False -> validateMagicNumber payloadRes +{- | The 2-bytes "magic number" is optional. If present, it indicates that a 4-bytes checksum follows. -} +validateMagicNumber :: (ValidateCheckSum -> B.Get Frame) -> B.Get Frame +validateMagicNumber payload = B.lookAheadM peekMagicNumber >>= \case + Just _ -> payload Yes + Nothing -> payload No + where + peekMagicNumber :: B.Get (Maybe ()) + peekMagicNumber = guard . (== frameMagicNumber) <$> B.getWord16be + +{- | If a checksum is given, validate it. Otherwise, return simple frame. -} validateCheckSum :: Frame -> B.Get Frame -validateCheckSum (PayloadFrame sc (PayloadCommand cs ms md pl)) = - let - metaSize = CL.toStrict (B.runPut $ B.putInt32be ms) - metadata = CL.toStrict md - payload = CL.toStrict pl - checksum = crc32c $ metaSize <> metadata <> payload - frame = PayloadFrame sc (PayloadCommand cs ms md (dropPayloadGarbage pl)) - in - if checksum == cs then return $! frame else fail "Invalid checksum" +validateCheckSum (PayloadFrame sc (PayloadCommand cs@(Just csm) ms md pl)) = + case runCheckSum (B.runPut (B.putInt32be ms) <> md <> pl) csm of + Valid -> return + $! PayloadFrame sc (PayloadCommand cs ms md (dropPayloadGarbage pl)) + Invalid -> fail "Invalid checksum" validateCheckSum x = return $! x -parsePayload :: Int32 -> Int32 -> SimpleCmd -> B.Get Frame -parsePayload ts cs simpleCmd = do - mn <- B.getWord16be - unless (mn == frameMagicNumber) $ fail ("Invalid magic number: " <> show mn) - cm <- B.getWord32be - ms <- B.getInt32be - md <- B.getLazyByteString . fromIntegral $ ms - -- 14 remaining bytes = 4 (command size field) + 2 (magic number) + 4 (checksum) + 4 (metadata size field) - pl <- payload $ ts - (14 + cs + ms) - let payloadCmd = PayloadCommand cm ms md pl - validateCheckSum (PayloadFrame simpleCmd payloadCmd) +{- | Take in a simple command and try to parse a payload command. -} +parsePayload :: Int32 -> Int32 -> SimpleCmd -> ValidateCheckSum -> B.Get Frame +parsePayload ts cs simpleCmd vcs = case vcs of + Yes -> parsePayload' . Just . CheckSum =<< B.getWord32be + No -> parsePayload' Nothing where - payload rms | rms > 0 = B.getLazyByteString $ fromIntegral rms - | otherwise = pure CL.empty + parsePayload' cm = do + ms <- B.getInt32be + md <- B.getLazyByteString . fromIntegral $ ms + pl <- B.getLazyByteString . fromIntegral $ ts - (remaining cm + cs + ms) + let frame = PayloadFrame simpleCmd (PayloadCommand cm ms md pl) + validateCheckSum frame + remaining (Just _) = 14 -- 4 (command size) + 2 (magic number) + 4 (checksum) + 4 (metadata size) + remaining Nothing = 8 -- no magic number and checksum decodeFrame :: CL.ByteString -> Either String Frame decodeFrame = bimap (\(_, _, e) -> e) (\(_, _, f) -> f) . B.runGetOrFail parseFrame +{- | Decode either a 'SimpleFrame' or a 'PayloadFrame'. -} decodeBaseCommand :: CL.ByteString -> Either String Response decodeBaseCommand bytes = decodeFrame bytes >>= \case - SimpleFrame s -> do - cmd <- PL.decodeMessage (CL.toStrict $ frameMessage s) - return $ SimpleResponse cmd + SimpleFrame s -> + SimpleResponse <$> PL.decodeMessage (CL.toStrict $ frameMessage s) PayloadFrame s (PayloadCommand _ _ md pl) -> do cmd <- PL.decodeMessage . CL.toStrict $ frameMessage s meta <- PL.decodeMessage . CL.toStrict $ md diff --git a/src/Pulsar/Protocol/Encoder.hs b/src/Pulsar/Protocol/Encoder.hs --- a/src/Pulsar/Protocol/Encoder.hs +++ b/src/Pulsar/Protocol/Encoder.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} - {- An encoder that understands the Pulsar protocol, as specified at: http://pulsar.apache.org/docs/en/develop-binary-protocol -} module Pulsar.Protocol.Encoder ( encodeBaseCommand @@ -8,13 +6,13 @@ import qualified Data.Binary.Put as B import qualified Data.ByteString.Lazy.Char8 as CL -import Data.Digest.CRC32C ( crc32c ) import Data.Int ( Int32 ) import Data.Maybe ( fromMaybe ) import qualified Data.ProtoLens.Encoding as PL import Proto.PulsarApi ( BaseCommand , MessageMetadata ) +import Pulsar.Protocol.CheckSum import Pulsar.Protocol.Frame mkSimpleCommand :: Int32 -> BaseCommand -> SimpleCmd @@ -34,13 +32,13 @@ -- payload fields metadata = PL.encodeMessage meta metaSize = fromIntegral . CL.length . CL.fromStrict $ metadata - metaSizeBS = B.runPut . B.putInt32be $ metaSize - checksum = crc32c $ CL.toStrict metaSizeBS <> metadata <> CL.toStrict pl - payloadSize = fromIntegral . CL.length $ pl + metaSizeBS = B.runPut $ B.putInt32be metaSize + checkSum = computeCheckSum $ metaSizeBS <> CL.fromStrict metadata <> pl + payloadSize = fromIntegral $ CL.length pl -- frame: extra 14 bytes = 2 (magic number) + 4 (checksum) + 4 (metadata size) + 4 (command size) extraBytes = fromIntegral (14 + metaSize) + payloadSize simpleCmd = mkSimpleCommand extraBytes cmd - payloadCmd = PayloadCommand { frameCheckSum = checksum + payloadCmd = PayloadCommand { frameCheckSum = Just checkSum , frameMetadataSize = metaSize , frameMetadata = CL.fromStrict metadata , framePayload = pl @@ -55,17 +53,23 @@ encodeFrame :: Frame -> CL.ByteString encodeFrame (SimpleFrame scmd) = encodeSimpleCmd scmd encodeFrame (PayloadFrame scmd (PayloadCommand cs mds md p)) = - let simpleCmd = encodeSimpleCmd scmd - metaSizeBS = B.runPut . B.putInt32be $ mds - magicNumber = B.runPut . B.putWord16be $ frameMagicNumber - crc32cSum = B.runPut . B.putWord32be $ cs - payloadCmd = magicNumber <> crc32cSum <> metaSizeBS <> md <> p + let simpleCmd = encodeSimpleCmd scmd + metaSizeBS = B.runPut $ B.putInt32be mds + payloadCmd = encodeOptionalFields cs <> metaSizeBS <> md <> p in simpleCmd <> payloadCmd +-- If a magic number is present, a CRC32-C checksum of everything that comes after it (4 bytes) should follow +encodeOptionalFields :: Maybe CheckSum -> CL.ByteString +encodeOptionalFields (Just (CheckSum cs)) = + let magicNumber = B.runPut $ B.putWord16be frameMagicNumber + crc32cSum = B.runPut $ B.putWord32be cs + in magicNumber <> crc32cSum +encodeOptionalFields Nothing = CL.empty + encodeBaseCommand :: Maybe MessageMetadata -> Maybe Payload -> BaseCommand -> CL.ByteString encodeBaseCommand (Just meta) p cmd = - let pl = fromMaybe (Payload "") p + let pl = fromMaybe (Payload CL.empty) p in encodeFrame . uncurry PayloadFrame $ mkPayloadCommand cmd meta pl encodeBaseCommand Nothing _ cmd = encodeFrame . SimpleFrame . mkSimpleCommand 4 $ cmd diff --git a/src/Pulsar/Protocol/Frame.hs b/src/Pulsar/Protocol/Frame.hs --- a/src/Pulsar/Protocol/Frame.hs +++ b/src/Pulsar/Protocol/Frame.hs @@ -7,6 +7,7 @@ import Proto.PulsarApi ( BaseCommand , MessageMetadata ) +import Pulsar.Protocol.CheckSum ( CheckSum ) -- The maximum allowable size of a single frame is 5 MB: http://pulsar.apache.org/docs/en/develop-binary-protocol/#framing frameMaxSize :: Int @@ -14,7 +15,7 @@ -- A 2-byte byte array (0x0e01) identifying the current format frameMagicNumber :: B.Word16 -frameMagicNumber = 0x0e01 +frameMagicNumber = 0x0e01 data Frame = SimpleFrame SimpleCmd | PayloadFrame SimpleCmd PayloadCmd @@ -27,7 +28,7 @@ -- Payload command: http://pulsar.apache.org/docs/en/develop-binary-protocol/#payload-commands data PayloadCmd = PayloadCommand - { frameCheckSum :: B.Word32 -- A CRC32-C checksum of everything that comes after it (4 bytes) + { frameCheckSum :: Maybe CheckSum -- A CRC32-C checksum of everything that comes after it (4 bytes) - OPTIONAL , frameMetadataSize :: Int32 -- The size of the message metadata (4 bytes) , frameMetadata :: CL.ByteString -- The message metadata stored as a binary protobuf message , framePayload :: CL.ByteString -- Anything left in the frame is considered the payload and can include any sequence of bytes diff --git a/src/Pulsar/Types.hs b/src/Pulsar/Types.hs --- a/src/Pulsar/Types.hs +++ b/src/Pulsar/Types.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE OverloadedStrings #-} + {-| Module : Pulsar.Types Description : End-user Pulsar API types. @@ -11,7 +13,9 @@ import qualified Data.ByteString.Lazy.Char8 as CL import Data.Char ( toLower ) -import Data.String +import Data.String ( IsString + , fromString + ) import qualified Data.Text as T import Proto.PulsarApi ( MessageIdData ) @@ -24,11 +28,11 @@ } {- | A default 'Topic': "non-persistent:\/\/public\/default\/my-topic". -} -defaultTopic :: String -> Topic +defaultTopic :: TopicName -> Topic defaultTopic n = Topic { type' = NonPersistent - , tenant = Tenant "public" - , namespace = NameSpace "default" - , name = TopicName n + , tenant = "public" + , namespace = "default" + , name = n } instance Show Topic where @@ -43,22 +47,31 @@ show NonPersistent = "non-persistent" {- | A tenant can be any string value. Default value is "public". -} -newtype Tenant = Tenant String +newtype Tenant = Tenant T.Text +instance IsString Tenant where + fromString = Tenant . T.pack + instance Show Tenant where - show (Tenant t) = t + show (Tenant t) = T.unpack t {- | A namespace can be any string value. Default value is "default". -} -newtype NameSpace = NameSpace String +newtype NameSpace = NameSpace T.Text +instance IsString NameSpace where + fromString = NameSpace . T.pack + instance Show NameSpace where - show (NameSpace t) = t + show (NameSpace t) = T.unpack t {- | A topic name can be any string value. -} -newtype TopicName = TopicName String +newtype TopicName = TopicName T.Text +instance IsString TopicName where + fromString = TopicName . T.pack + instance Show TopicName where - show (TopicName t) = t + show (TopicName t) = T.unpack t {- | A message id, needed for acknowledging messages. See 'Pulsar.Consumer.ack'. -} newtype MsgId = MsgId MessageIdData @@ -73,7 +86,13 @@ fromString = PulsarMessage . CL.pack {- | A subscription name can be any string value. -} -newtype SubscriptionName = SubscriptionName T.Text deriving Show +newtype SubName = SubName T.Text deriving Show -instance IsString SubscriptionName where - fromString = SubscriptionName . T.pack +instance IsString SubName where + fromString = SubName . T.pack + +{- | A subscription type. See to learn more. -} +data SubType = Exclusive | Failover | Shared | KeyShared deriving Show + +{- | A subscription with a type and a name. -} +data Subscription = Subscription SubType SubName deriving Show diff --git a/supernova.cabal b/supernova.cabal --- a/supernova.cabal +++ b/supernova.cabal @@ -1,7 +1,7 @@ cabal-version: 3.0 name: supernova -version: 0.0.2 +version: 0.0.3 synopsis: Apache Pulsar client for Haskell description: Supernova is an Apache Pulsar client that implements the specified TCP protocol. homepage: https://github.com/cr-org/supernova @@ -19,6 +19,7 @@ other-modules: Paths_supernova , Proto.PulsarApi , Proto.PulsarApi_Fields + , Pulsar.AppState , Pulsar.Core , Pulsar.Connection , Pulsar.Consumer @@ -26,26 +27,29 @@ , Pulsar.Internal.Logger , Pulsar.Internal.TCPClient , Pulsar.Producer + , Pulsar.Protocol.CheckSum , Pulsar.Protocol.Commands , Pulsar.Protocol.Decoder , Pulsar.Protocol.Encoder , Pulsar.Protocol.Frame , Pulsar.Types autogen-modules: Paths_supernova - build-depends: base >= 4.13.0 && < 4.14, + build-depends: async >= 2.2.2 && < 2.3, + base >= 4.13.0 && < 4.14, bifunctor >= 0.1.0 && < 0.2, binary >= 0.8.7 && < 0.9, bytestring >= 0.10.10 && < 0.11, crc32c >= 0.0.0 && < 0.1, exceptions >= 0.10.4 && < 0.11, lens-family-core >= 2.0.0 && < 2.1, + lens-family-th >= 0.5.1 && < 0.6, logging >= 3.0.5 && < 3.1, + mtl >= 2.2.2 && < 2.3, text >= 1.2.4 && < 1.3, managed >= 1.0.7 && < 1.1, network >= 3.1.2 && < 3.2, proto-lens >= 0.7.0 && < 0.8, proto-lens-runtime >= 0.7.0 && < 0.8, - unliftio >= 0.2.13 && < 0.3 hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -1,11 +1,11 @@ -{-# LANGUAGE DeriveAnyClass, DeriveGeneric #-} -{-# LANGUAGE OverloadedStrings, RecordWildCards #-} +{-# LANGUAGE DeriveAnyClass, DeriveGeneric, OverloadedStrings, RecordWildCards #-} module Main where import Control.Concurrent ( threadDelay ) import Control.Concurrent.Async ( concurrently_ ) import Control.Monad ( forever ) +import Control.Monad.IO.Class ( liftIO ) import Data.Aeson import qualified Data.ByteString.Lazy.Char8 as CL import Data.Foldable ( traverse_ ) @@ -38,19 +38,27 @@ topic :: Topic topic = defaultTopic "app" +sub :: Subscription +sub = Subscription Exclusive "test-sub" + demo :: IO () -demo = runPulsar resources $ \(Consumer {..}, Producer {..}) -> +demo = runPulsar conn pulsar + +conn :: PulsarConnection +conn = connect defaultConnectData + +pulsar :: Pulsar () +pulsar = do + c <- newConsumer topic sub + p <- newProducer topic + liftIO $ program c p + +program :: Consumer IO -> Producer IO -> IO () +program Consumer {..} Producer {..} = let c = forever $ fetch >>= \(Message i m) -> msgDecoder m >> ack i - p = forever $ sleep 5 >> traverse_ produce messages + p = forever $ sleep 5 >> traverse_ send messages in concurrently_ c p -resources :: Pulsar (Consumer IO, Producer IO) -resources = do - ctx <- connect defaultConnectData - consumer <- newConsumer ctx topic "test-sub" - producer <- newProducer ctx topic - return (consumer, producer) - sleep :: Int -> IO () sleep n = threadDelay (n * 1000000) @@ -58,7 +66,13 @@ logOpts = LogOptions Info StdOut streamDemo :: IO () -streamDemo = runPulsar' logOpts resources $ \(Consumer {..}, Producer {..}) -> +streamDemo = runPulsar' logOpts conn $ do + c <- newConsumer topic sub + p <- newProducer topic + liftIO $ streamProgram c p + +streamProgram :: Consumer IO -> Producer IO -> IO () +streamProgram (Consumer fetch ack) (Producer send) = let c = forever $ fetch >>= \(Message i m) -> msgDecoder m >> ack i - p = forever $ sleep 5 >> traverse_ produce messages + p = forever $ sleep 5 >> traverse_ send messages in S.drain . asyncly . maxThreads 10 $ S.yieldM c <> S.yieldM p