supernova 0.0.2 → 0.0.3
raw patch · 17 files changed
+692/−346 lines, 17 filesdep +lens-family-thdep +mtldep −unliftiodep ~async
Dependencies added: lens-family-th, mtl
Dependencies removed: unliftio
Dependency ranges changed: async
Files
- README.md +39/−19
- src/Pulsar.hs +36/−15
- src/Pulsar/AppState.hs +99/−0
- src/Pulsar/Connection.hs +85/−67
- src/Pulsar/Consumer.hs +117/−40
- src/Pulsar/Core.hs +34/−65
- src/Pulsar/Internal/Core.hs +47/−22
- src/Pulsar/Internal/TCPClient.hs +16/−11
- src/Pulsar/Producer.hs +62/−25
- src/Pulsar/Protocol/CheckSum.hs +16/−0
- src/Pulsar/Protocol/Commands.hs +13/−7
- src/Pulsar/Protocol/Decoder.hs +41/−30
- src/Pulsar/Protocol/Encoder.hs +17/−13
- src/Pulsar/Protocol/Frame.hs +3/−2
- src/Pulsar/Types.hs +33/−14
- supernova.cabal +7/−3
- test/Main.hs +27/−13
README.md view
@@ -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 ```
src/Pulsar.hs view
@@ -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(..)
+ src/Pulsar/AppState.hs view
@@ -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
src/Pulsar/Connection.hs view
@@ -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 =
src/Pulsar/Consumer.hs view
@@ -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 ()
src/Pulsar/Core.hs view
@@ -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
src/Pulsar/Internal/Core.hs view
@@ -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
src/Pulsar/Internal/TCPClient.hs view
@@ -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
src/Pulsar/Producer.hs view
@@ -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
+ src/Pulsar/Protocol/CheckSum.hs view
@@ -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)
src/Pulsar/Protocol/Commands.hs view
@@ -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
src/Pulsar/Protocol/Decoder.hs view
@@ -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
src/Pulsar/Protocol/Encoder.hs view
@@ -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
src/Pulsar/Protocol/Frame.hs view
@@ -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
src/Pulsar/Types.hs view
@@ -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 <https://pulsar.apache.org/docs/en/concepts-messaging/#subscriptions> 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
supernova.cabal view
@@ -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
test/Main.hs view
@@ -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