diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,75 @@ +supernova +========= + +[![CI Status](https://github.com/cr-org/supernova/workflows/Haskell%20CI/badge.svg)](https://github.com/cr-org/supernova/actions) + +⚠️ **it is still very much under development 🚧 so use it at your own risk** ⚠️ + +A *supernova* is a powerful and luminous stellar explosion. This transient astronomical event occurs during the last evolutionary stages of a massive star or when a white dwarf is triggered into runaway nuclear fusion. The original object, called the progenitor, either collapses to a [neutron](https://github.com/cr-org/neutron) star or black hole, or is completely destroyed. The peak optical luminosity of a supernova can be comparable to that of an entire galaxy before fading over several weeks or months. + +[![supernova](https://www.jpl.nasa.gov/spaceimages/images/largesize/PIA22352_hires.jpg "Kepler Beyond Planets: Finding Exploding Stars (Type Ia Supernova from a White Dwarf Stealing Matter)")](https://www.jpl.nasa.gov/spaceimages/details.php?id=PIA22352) + +### Quick Start + +The example located in `test/Main.hs` showcases a simple consumer and producer running concurrently (needs the `async` library). + +```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 + +resources :: Pulsar (Consumer IO, Producer IO) +resources = do + ctx <- connect defaultConnectData + consumer <- newConsumer ctx topic "test-sub" + producer <- newProducer ctx topic + return (consumer, producer) +``` + +A `Message` contains a `MessageID` you need for `ack`ing and a payload defined as a lazy `ByteString`. + +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`. + +```haskell +logOpts :: LogOptions +logOpts = LogOptions Info StdOut + +runPulsar' logOpts resources +``` + +### Streaming + +Since both consumers and producers operate on any `MonadIO m`, we could leverage some streaming libraries. Here's the same example using [streamly](https://hackage.haskell.org/package/streamly). + +```haskell +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 + in S.drain . asyncly . maxThreads 10 $ S.yieldM c <> S.yieldM p +``` + +### Development + +It is recommended to use [Cachix](https://app.cachix.org/cache/hpulsar) to reduce the compilation time. + +```shell +nix-build +``` + +Or within a Nix shell (run `nix-shell` at the project's root). + +```shell +cabal new-build +``` diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,3 @@ +import Data.ProtoLens.Setup + +main = defaultMainGeneratingProtos "src" diff --git a/src/Pulsar.hs b/src/Pulsar.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar.hs @@ -0,0 +1,61 @@ +{-| +Module : Pulsar +Description : Apache Pulsar client +License : Apache-2.0 +Maintainer : gabriel.volpe@chatroulette.com +Stability : experimental + +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. + +@ +resources :: Pulsar (Consumer IO, Producer IO) +resources = do + ctx <- connect defaultConnectData + consumer <- newConsumer ctx topic "test-sub" + producer <- newProducer ctx topic + return (consumer, producer) +@ + +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). + +@ +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" + in concurrently_ c p +@ +-} +module Pulsar + ( connect + , defaultConnectData + , newConsumer + , newProducer + , runPulsar + , runPulsar' + , Consumer(..) + , Producer(..) + , Pulsar + , PulsarCtx + , ConnectData + , LogLevel(..) + , LogOptions(..) + , LogOutput(..) + , module Pulsar.Types + ) +where + +import Pulsar.Connection +import Pulsar.Consumer +import Pulsar.Internal.Core +import Pulsar.Producer +import Pulsar.Types diff --git a/src/Pulsar/Connection.hs b/src/Pulsar/Connection.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Connection.hs @@ -0,0 +1,156 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving, 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 Data.Foldable ( traverse_ ) +import Data.IORef +import Lens.Family +import qualified Network.Socket as NS +import qualified Network.Socket.ByteString.Lazy + as SBL +import Proto.PulsarApi ( BaseCommand + , MessageMetadata + ) +import qualified Proto.PulsarApi_Fields as F +import Pulsar.Internal.Logger +import Pulsar.Internal.TCPClient ( acquireSocket ) +import qualified Pulsar.Protocol.Commands as P +import Pulsar.Protocol.Decoder ( decodeBaseCommand ) +import Pulsar.Protocol.Encoder ( encodeBaseCommand ) +import Pulsar.Protocol.Frame ( Payload + , Response(..) + , frameMaxSize + , 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 + , connPort :: NS.ServiceName + } deriving Show + +{- | Internal Pulsar context. You will never need to access its content (not exported) but might need to take it as argument. -} +data PulsarCtx = Ctx + { ctxConn :: Connection + , ctxState :: IORef AppState + } + +{- | Default connection data: "127.0.0.1:6650" -} +defaultConnectData :: ConnectData +defaultConnectData = ConnData { connHost = "127.0.0.1", connPort = "6650" } + +{- | Starts a Pulsar connection with the supplied 'ConnectData' -} +connect + :: (MonadThrow m, MonadIO m, MonadManaged m) => ConnectData -> m PulsarCtx +connect (ConnData h p) = do + socket <- acquireSocket h p + liftIO $ sendSimpleCmd socket P.connect + 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) + +initAppState :: MonadIO m => m (IORef AppState) +initAppState = liftIO . newIORef $ AppState [] 0 [] 0 0 + +recvDispatch + :: MonadIO m => NS.Socket -> IORef AppState -> Chan BaseCommand -> m () +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)) + +{- 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 s chan = forever $ do + threadDelay (29 * 1000000) + logRequest P.ping + sendSimpleCmd s P.ping + liftIO $ timeout (2 * 1000000) (readChan chan) >>= \case + Just cmd -> logResponse cmd + Nothing -> throwIO (userError "Keep Alive interruption") + +sendSimpleCmd :: MonadIO m => NS.Socket -> BaseCommand -> m () +sendSimpleCmd s cmd = + liftIO . SBL.sendAll s $ encodeBaseCommand Nothing Nothing cmd + +sendPayloadCmd + :: MonadIO m + => NS.Socket + -> BaseCommand + -> MessageMetadata + -> Maybe Payload + -> m () +sendPayloadCmd s cmd meta payload = + liftIO . SBL.sendAll s $ encodeBaseCommand (Just meta) payload cmd + +receive :: MonadIO m => NS.Socket -> m Response +receive s = liftIO $ do + msg <- SBL.recv s $ fromIntegral frameMaxSize + case decodeBaseCommand msg of + Left e -> fail $ "Decoding error: " <> e + Right resp -> case getCommand resp ^. F.maybe'ping of + Just _ -> do + logResponse $ getCommand resp + logRequest P.pong + sendSimpleCmd s P.pong + return resp + Nothing -> return resp diff --git a/src/Pulsar/Consumer.hs b/src/Pulsar/Consumer.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Consumer.hs @@ -0,0 +1,60 @@ +{-# LANGUAGE LambdaCase, OverloadedStrings #-} + +module Pulsar.Consumer where + +import Control.Monad ( forever ) +import qualified Control.Monad.Catch as E +import Control.Monad.Managed +import Lens.Family +import qualified Proto.PulsarApi_Fields as F +import qualified Pulsar.Core as C +import Pulsar.Connection +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 + { fetch :: m Message -- ^ Fetches a single message. Blocks if no messages are available. + , ack :: MsgId -> m () -- ^ Acknowledges a single message. + } + +{- | 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 + -> 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) + ) + 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 diff --git a/src/Pulsar/Core.hs b/src/Pulsar/Core.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Core.hs @@ -0,0 +1,140 @@ +{-# LANGUAGE DataKinds, FlexibleContexts, LambdaCase, OverloadedStrings #-} + +{- Defines a set of transactional commands, communicating via internal channels -} +module Pulsar.Core where + +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 qualified Proto.PulsarApi_Fields as F +import Pulsar.Connection +import Pulsar.Internal.Logger +import qualified Pulsar.Protocol.Commands as P +import Pulsar.Protocol.Frame ( Payload(..) + , Response(..) + , 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 + 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 + +newProducer + :: Connection -> Chan Response -> ReqId -> ProducerId -> Topic -> IO Text +newProducer (Conn s) chan r@(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 + 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 + logRequest $ P.closeProducer req pid + sendSimpleCmd s $ P.closeProducer req pid + void $ verifyResponse r chan F.maybe'success + +newSubscriber + :: Connection + -> Chan Response + -> ReqId + -> ConsumerId + -> Topic + -> SubscriptionName + -> 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 + +flow :: Connection -> ConsumerId -> IO () +flow (Conn s) (CId cid) = do + logRequest $ P.flow cid + sendSimpleCmd s $ P.flow cid + +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 + 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 () + +------ Keep Alive ------- + +ping :: (MonadThrow m, MonadIO m) => Connection -> Chan Response -> m () +ping (Conn s) chan = do + logRequest P.ping + sendSimpleCmd s P.ping + cmd <- getCommand <$> readChan chan + case cmd ^. F.maybe'pong of + Just p -> logResponse p + Nothing -> liftIO . throwIO $ userError "Failed to get PONG" + +pong :: MonadIO m => Connection -> m () +pong (Conn s) = do + logRequest P.pong + sendSimpleCmd s P.pong + +------ Payload commands ------ + +send + :: Connection + -> Chan Response + -> ProducerId + -> SeqId + -> PulsarMessage + -> IO () +send (Conn s) chan (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' diff --git a/src/Pulsar/Internal/Core.hs b/src/Pulsar/Internal/Core.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Internal/Core.hs @@ -0,0 +1,48 @@ +{-# LANGUAGE RankNTypes, GeneralizedNewtypeDeriving #-} + +{- Defines a Pulsar Monad, which wraps a Managed resource -} +module Pulsar.Internal.Core where + +import qualified Control.Logging as L +import Control.Monad.Catch +import Control.Monad.Managed + +{- | The main Pulsar monad, which abstracts over a 'Managed' monad. -} +newtype Pulsar a = Pulsar (Managed a) + deriving (Functor, Applicative, Monad, MonadIO, MonadManaged) + +{- | 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 + +{- | 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 + 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 + +{- | Internal logging options. Can be used together with `runPulsar'`. -} +data LogOptions = LogOptions + { logLevel :: LogLevel + , logOutput :: LogOutput + } deriving Show + +{- | Internal logging level, part of 'LogOptions'. Can be used together with `runPulsar'`. -} +data LogLevel = Error | Warn | Info | Debug deriving Show + +{- | 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 diff --git a/src/Pulsar/Internal/Logger.hs b/src/Pulsar/Internal/Logger.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Internal/Logger.hs @@ -0,0 +1,14 @@ +{-# LANGUAGE OverloadedStrings #-} + +{- Defines a few internal functions for logging requests and responses in debug mode -} +module Pulsar.Internal.Logger where + +import Control.Logging ( debug' ) +import Control.Monad.IO.Class ( MonadIO ) +import qualified Data.Text as T + +logRequest :: (MonadIO m, Show a) => a -> m () +logRequest cmd = debug' $ ">>> " <> T.pack (show cmd) + +logResponse :: (MonadIO m, Show a) => a -> m () +logResponse cmd = debug' $ "<<< " <> T.pack (show cmd) diff --git a/src/Pulsar/Internal/TCPClient.hs b/src/Pulsar/Internal/TCPClient.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Internal/TCPClient.hs @@ -0,0 +1,31 @@ +{-# LANGUAGE LambdaCase, OverloadedStrings #-} + +{- A simple TCP client, used to communicate with the Pulsar server -} +module Pulsar.Internal.TCPClient + ( acquireSocket + ) +where + +import qualified Control.Exception as E +import Control.Monad.IO.Class +import Control.Monad.Managed +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) + ) + where + 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 + NS.connect sock $ NS.addrAddress addr + return sock diff --git a/src/Pulsar/Producer.hs b/src/Pulsar/Producer.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Producer.hs @@ -0,0 +1,49 @@ +{- Defines a high-level Pulsar producer for the end user -} +module Pulsar.Producer where + +import qualified Control.Monad.Catch as E +import Control.Monad.Managed +import Data.IORef +import Data.Text ( Text ) +import qualified Pulsar.Core as C +import Pulsar.Connection +import Pulsar.Types +import UnliftIO.Chan + +{- | An abstract 'Producer' able to 'produce' messages of type 'PulsarMessage'. -} +newtype Producer m = Producer + { produce :: PulsarMessage -> m () -- ^ Produces a single message. + } + +data ProducerState = ProducerState + { stSeqId :: SeqId -- an incremental message sequence counter + , stName :: Text -- a unique name + } + +mkSeqId :: MonadIO m => IORef ProducerState -> m SeqId +mkSeqId ref = liftIO $ atomicModifyIORef + ref + (\(ProducerState s n) -> let s' = s + 1 in (ProducerState s' n, s)) + +{- | 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) + ) + where + newReq = mkRequestId app + dispatch chan pid 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 diff --git a/src/Pulsar/Protocol/Commands.hs b/src/Pulsar/Protocol/Commands.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Protocol/Commands.hs @@ -0,0 +1,124 @@ +{-# LANGUAGE OverloadedStrings #-} + +{- Defines commands using the generated protobuf datatypes -} +module Pulsar.Protocol.Commands where + +import qualified Data.Binary as B +import Data.ProtoLens ( defMessage ) +import qualified Data.Text as T +import Data.Version ( showVersion ) +import Lens.Family +import Paths_supernova ( version ) +import Proto.PulsarApi +import qualified Proto.PulsarApi_Fields as F +import Pulsar.Types + +connect :: BaseCommand +connect = defMessage + & F.connect .~ conn + where + conn :: CommandConnect + conn = defMessage + & 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 + & 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.consumerId .~ cid + & F.requestId .~ req + +flow :: B.Word64 -> BaseCommand +flow cid = defMessage + & F.type' .~ BaseCommand'FLOW + & F.flow .~ flowCmd + where + flowCmd :: CommandFlow + flowCmd = defMessage + & F.messagePermits .~ 100 + & F.consumerId .~ cid + +ack :: B.Word64 -> MessageIdData -> BaseCommand +ack cid msgId = defMessage + & F.type' .~ BaseCommand'ACK + & F.ack .~ ackCmd + where + ackCmd :: CommandAck + ackCmd = defMessage + & F.messageId .~ [ msgId ] + & F.consumerId .~ cid + +closeConsumer :: B.Word64 -> B.Word64 -> BaseCommand +closeConsumer req cid = defMessage + & F.type' .~ BaseCommand'CLOSE_CONSUMER + & F.closeConsumer .~ close + where + close :: CommandCloseConsumer + close = defMessage + & F.consumerId .~ cid + & F.requestId .~ req + +producer :: B.Word64 -> B.Word64 -> Topic -> BaseCommand +producer req pid topic = defMessage + & F.type' .~ BaseCommand'PRODUCER + & F.producer .~ prod + where + prod :: CommandProducer + prod = defMessage + & F.topic .~ T.pack (show topic) + & F.producerId .~ pid + & F.requestId .~ req + +closeProducer :: B.Word64 -> B.Word64 -> BaseCommand +closeProducer req pid = defMessage + & F.type' .~ BaseCommand'CLOSE_PRODUCER + & F.closeProducer .~ prod + where + prod :: CommandCloseProducer + prod = defMessage + & F.producerId .~ pid + & F.requestId .~ req + +send :: B.Word64 -> B.Word64 -> BaseCommand +send pid sid = defMessage + & F.type' .~ BaseCommand'SEND + & F.send .~ sendCmd + where + sendCmd :: CommandSend + sendCmd = defMessage + & F.producerId .~ pid + & F.sequenceId .~ sid + +lookup :: B.Word64 -> Topic -> BaseCommand +lookup req topic = defMessage + & F.type' .~ BaseCommand'LOOKUP + & F.lookupTopic .~ lut + where + lut :: CommandLookupTopic + lut = defMessage + & F.topic .~ T.pack (show topic) + & F.requestId .~ req + +------- Keep Alive -------- + +ping :: BaseCommand +ping = defMessage + & F.type' .~ BaseCommand'PING + & F.ping .~ defMessage + +pong :: BaseCommand +pong = defMessage + & F.type' .~ BaseCommand'PONG + & F.pong .~ defMessage + +------- Metadata for Payload commands -------- + +messageMetadata :: MessageMetadata +messageMetadata = defMessage diff --git a/src/Pulsar/Protocol/Decoder.hs b/src/Pulsar/Protocol/Decoder.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Protocol/Decoder.hs @@ -0,0 +1,84 @@ +{-# LANGUAGE FlexibleInstances, LambdaCase, OverloadedStrings #-} + +{- A decoder that understands the Pulsar protocol, as specified at: http://pulsar.apache.org/docs/en/develop-binary-protocol -} +module Pulsar.Protocol.Decoder + ( decodeBaseCommand + , dropPayloadGarbage + ) +where + +import Control.Monad ( unless ) +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.Frame + +{- + - 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/ + - + - Source: https://github.com/apache/pulsar/blob/master/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/fbs/Message.java#L22 + - + - More info on the Ascii spec: https://www.december.com/html/spec/ascii.html. Maybe this could be helpful: https://hackage.haskell.org/package/flatbuffers + -} +dropPayloadGarbage :: CL.ByteString -> CL.ByteString +dropPayloadGarbage bs = + maybe bs (CL.drop 3) (CL.stripPrefix "\NUL\NUL\NUL\EOT\CAN" bs) + +parseFrame :: B.Get Frame +parseFrame = do + ts <- B.getInt32be + cs <- B.getInt32be + ms <- B.getLazyByteString (fromIntegral cs) + let simpleCmd = SimpleCommand ts cs ms + B.isEmpty >>= \case + True -> return $ SimpleFrame simpleCmd + False -> parsePayload ts cs simpleCmd + +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 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) + where + payload rms | rms > 0 = B.getLazyByteString $ fromIntegral rms + | otherwise = pure CL.empty + +decodeFrame :: CL.ByteString -> Either String Frame +decodeFrame = + bimap (\(_, _, e) -> e) (\(_, _, f) -> f) . B.runGetOrFail parseFrame + +decodeBaseCommand :: CL.ByteString -> Either String Response +decodeBaseCommand bytes = decodeFrame bytes >>= \case + SimpleFrame s -> do + cmd <- PL.decodeMessage (CL.toStrict $ frameMessage s) + return $ SimpleResponse cmd + PayloadFrame s (PayloadCommand _ _ md pl) -> do + cmd <- PL.decodeMessage . CL.toStrict $ frameMessage s + meta <- PL.decodeMessage . CL.toStrict $ md + return $ PayloadResponse cmd meta (payload pl) + where + payload p | CL.null p = Nothing + | otherwise = Just $ Payload p diff --git a/src/Pulsar/Protocol/Encoder.hs b/src/Pulsar/Protocol/Encoder.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Protocol/Encoder.hs @@ -0,0 +1,71 @@ +{-# 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 + ) +where + +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.Frame + +mkSimpleCommand :: Int32 -> BaseCommand -> SimpleCmd +mkSimpleCommand extraBytes cmd = SimpleCommand + { frameTotalSize = cmdSize + extraBytes + , frameCommandSize = cmdSize + , frameMessage = msg + } + where + msg = CL.fromStrict $ PL.encodeMessage cmd + cmdSize = fromIntegral $ CL.length msg + +mkPayloadCommand + :: BaseCommand -> MessageMetadata -> Payload -> (SimpleCmd, PayloadCmd) +mkPayloadCommand cmd meta (Payload pl) = (simpleCmd, payloadCmd) + where + -- 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 + -- 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 + , frameMetadataSize = metaSize + , frameMetadata = CL.fromStrict metadata + , framePayload = pl + } + +encodeSimpleCmd :: SimpleCmd -> CL.ByteString +encodeSimpleCmd (SimpleCommand ts cs msg) = + let totalSize = B.runPut $ B.putInt32be ts + commandSize = B.runPut $ B.putInt32be cs + in totalSize <> commandSize <> msg + +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 + in simpleCmd <> payloadCmd + +encodeBaseCommand + :: Maybe MessageMetadata -> Maybe Payload -> BaseCommand -> CL.ByteString +encodeBaseCommand (Just meta) p cmd = + let pl = fromMaybe (Payload "") 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 new file mode 100644 --- /dev/null +++ b/src/Pulsar/Protocol/Frame.hs @@ -0,0 +1,43 @@ +{- An internal representation of a frame, as specified by the Pulsar protocol: http://pulsar.apache.org/docs/en/develop-binary-protocol -} +module Pulsar.Protocol.Frame where + +import qualified Data.Binary as B +import qualified Data.ByteString.Lazy.Char8 as CL +import Data.Int ( Int32 ) +import Proto.PulsarApi ( BaseCommand + , MessageMetadata + ) + +-- The maximum allowable size of a single frame is 5 MB: http://pulsar.apache.org/docs/en/develop-binary-protocol/#framing +frameMaxSize :: Int +frameMaxSize = 5 * 1024 * 1024 -- 5mb + +-- A 2-byte byte array (0x0e01) identifying the current format +frameMagicNumber :: B.Word16 +frameMagicNumber = 0x0e01 + +data Frame = SimpleFrame SimpleCmd | PayloadFrame SimpleCmd PayloadCmd + +-- Simple command: http://pulsar.apache.org/docs/en/develop-binary-protocol/#simple-commands +data SimpleCmd = SimpleCommand + { frameTotalSize :: Int32 -- The size of the frame, counting everything that comes after it (4 bytes) + , frameCommandSize :: Int32 -- The size of the protobuf-serialized command (4 bytes) + , frameMessage :: CL.ByteString -- The protobuf message serialized in a raw binary format (rather than in protobuf format) + } + +-- 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) + , 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 + } + +newtype Payload = Payload CL.ByteString deriving Show + +data Response = SimpleResponse BaseCommand | PayloadResponse BaseCommand MessageMetadata (Maybe Payload) deriving Show + +getCommand :: Response -> BaseCommand +getCommand response = case response of + (SimpleResponse cmd ) -> cmd + (PayloadResponse cmd _ _) -> cmd diff --git a/src/Pulsar/Types.hs b/src/Pulsar/Types.hs new file mode 100644 --- /dev/null +++ b/src/Pulsar/Types.hs @@ -0,0 +1,79 @@ +{-| +Module : Pulsar.Types +Description : End-user Pulsar API types. +License : Apache-2.0 +Maintainer : gabriel.volpe@chatroulette.com +Stability : experimental + +End-user types to configure consumers and producers. +-} +module Pulsar.Types where + +import qualified Data.ByteString.Lazy.Char8 as CL +import Data.Char ( toLower ) +import Data.String +import qualified Data.Text as T +import Proto.PulsarApi ( MessageIdData ) + +{- | A Topic is in the form "type:\/\/tenant\/namespace\/topic-name", which is what the 'Show' instance does. -} +data Topic = Topic + { type' :: TopicType + , tenant :: Tenant + , namespace :: NameSpace + , name :: TopicName + } + +{- | A default 'Topic': "non-persistent:\/\/public\/default\/my-topic". -} +defaultTopic :: String -> Topic +defaultTopic n = Topic { type' = NonPersistent + , tenant = Tenant "public" + , namespace = NameSpace "default" + , name = TopicName n + } + +instance Show Topic where + show (Topic typ tn ns n) = + toLower <$> show typ <> "://" <> show tn <> "/" <> show ns <> "/" <> show n + +{- | A topic can be either 'Persistent' or 'NonPersistent'. -} +data TopicType = Persistent | NonPersistent + +instance Show TopicType where + show Persistent = "persistent" + show NonPersistent = "non-persistent" + +{- | A tenant can be any string value. Default value is "public". -} +newtype Tenant = Tenant String + +instance Show Tenant where + show (Tenant t) = t + +{- | A namespace can be any string value. Default value is "default". -} +newtype NameSpace = NameSpace String + +instance Show NameSpace where + show (NameSpace t) = t + +{- | A topic name can be any string value. -} +newtype TopicName = TopicName String + +instance Show TopicName where + show (TopicName t) = t + +{- | A message id, needed for acknowledging messages. See 'Pulsar.Consumer.ack'. -} +newtype MsgId = MsgId MessageIdData + +{- | A consumed message, containing both 'MsgId' and payload as bytestring. -} +data Message = Message MsgId CL.ByteString + +{- | A produced message, containing just a payload as bytestring. -} +newtype PulsarMessage = PulsarMessage CL.ByteString deriving Show + +instance IsString PulsarMessage where + fromString = PulsarMessage . CL.pack + +{- | A subscription name can be any string value. -} +newtype SubscriptionName = SubscriptionName T.Text deriving Show + +instance IsString SubscriptionName where + fromString = SubscriptionName . T.pack diff --git a/src/pulsar_api.proto b/src/pulsar_api.proto new file mode 100644 --- /dev/null +++ b/src/pulsar_api.proto @@ -0,0 +1,963 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +// Source: https://raw.githubusercontent.com/apache/pulsar/master/pulsar-common/src/main/proto/PulsarApi.proto +syntax = "proto2"; + +package pulsar.proto; +option java_package = "org.apache.pulsar.common.api.proto"; +option optimize_for = LITE_RUNTIME; + +message Schema { + enum Type { + None = 0; + String = 1; + Json = 2; + Protobuf = 3; + Avro = 4; + Bool = 5; + Int8 = 6; + Int16 = 7; + Int32 = 8; + Int64 = 9; + Float = 10; + Double = 11; + Date = 12; + Time = 13; + Timestamp = 14; + KeyValue = 15; + } + + required string name = 1; + required bytes schema_data = 3; + required Type type = 4; + repeated KeyValue properties = 5; + +} + +message MessageIdData { + required uint64 ledgerId = 1; + required uint64 entryId = 2; + optional int32 partition = 3 [default = -1]; + optional int32 batch_index = 4 [default = -1]; + repeated int64 ack_set = 5; +} + +message KeyValue { + required string key = 1; + required string value = 2; +} + +message KeyLongValue { + required string key = 1; + required uint64 value = 2; +} + +message IntRange { + required int32 start = 1; + required int32 end = 2; +} + +message EncryptionKeys { + required string key = 1; + required bytes value = 2; + repeated KeyValue metadata = 3; +} + +enum CompressionType { + NONE = 0; + LZ4 = 1; + ZLIB = 2; + ZSTD = 3; + SNAPPY = 4; +} + +message MessageMetadata { + required string producer_name = 1; + required uint64 sequence_id = 2; + required uint64 publish_time = 3; + repeated KeyValue properties = 4; + + // Property set on replicated message, + // includes the source cluster name + optional string replicated_from = 5; + //key to decide partition for the msg + optional string partition_key = 6; + // Override namespace's replication + repeated string replicate_to = 7; + optional CompressionType compression = 8 [default = NONE]; + optional uint32 uncompressed_size = 9 [default = 0]; + // Removed below checksum field from Metadata as + // it should be part of send-command which keeps checksum of header + payload + //optional sfixed64 checksum = 10; + // differentiate single and batch message metadata + optional int32 num_messages_in_batch = 11 [default = 1]; + + // the timestamp that this event occurs. it is typically set by applications. + // if this field is omitted, `publish_time` can be used for the purpose of `event_time`. + optional uint64 event_time = 12 [default = 0]; + // Contains encryption key name, encrypted key and metadata to describe the key + repeated EncryptionKeys encryption_keys = 13; + // Algorithm used to encrypt data key + optional string encryption_algo = 14; + // Additional parameters required by encryption + optional bytes encryption_param = 15; + optional bytes schema_version = 16; + + optional bool partition_key_b64_encoded = 17 [ default = false ]; + // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode. + optional bytes ordering_key = 18; + + // Mark the message to be delivered at or after the specified timestamp + optional int64 deliver_at_time = 19; + + // Identify whether a message is a "marker" message used for + // internal metadata instead of application published data. + // Markers will generally not be propagated back to clients + optional int32 marker_type = 20; + + // transaction related message info + optional uint64 txnid_least_bits = 22; + optional uint64 txnid_most_bits = 23; + + /// Add highest sequence id to support batch message with external sequence id + optional uint64 highest_sequence_id = 24 [default = 0]; + + // Indicate if the message payload value is set + optional bool null_value = 25 [ default = false ]; + optional string uuid = 26; + optional int32 num_chunks_from_msg = 27; + optional int32 total_chunk_msg_size = 28; + optional int32 chunk_id = 29; + + // Indicate if the message partition key is set + optional bool null_partition_key = 30 [default = false]; +} + +message SingleMessageMetadata { + repeated KeyValue properties = 1; + optional string partition_key = 2; + required int32 payload_size = 3; + optional bool compacted_out = 4 [default = false]; + + // the timestamp that this event occurs. it is typically set by applications. + // if this field is omitted, `publish_time` can be used for the purpose of `event_time`. + optional uint64 event_time = 5 [default = 0]; + optional bool partition_key_b64_encoded = 6 [ default = false ]; + // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode. + optional bytes ordering_key = 7; + // Allows consumer retrieve the sequence id that the producer set. + optional uint64 sequence_id = 8; + // Indicate if the message payload value is set + optional bool null_value = 9 [ default = false ]; + // Indicate if the message partition key is set + optional bool null_partition_key = 10 [ default = false]; +} + +enum ServerError { + UnknownError = 0; + MetadataError = 1; // Error with ZK/metadata + PersistenceError = 2; // Error writing reading from BK + AuthenticationError = 3; // Non valid authentication + AuthorizationError = 4; // Not authorized to use resource + + ConsumerBusy = 5; // Unable to subscribe/unsubscribe because + // other consumers are connected + ServiceNotReady = 6; // Any error that requires client retry operation with a fresh lookup + ProducerBlockedQuotaExceededError = 7; // Unable to create producer because backlog quota exceeded + ProducerBlockedQuotaExceededException = 8; // Exception while creating producer because quota exceeded + ChecksumError = 9; // Error while verifying message checksum + UnsupportedVersionError = 10; // Error when an older client/version doesn't support a required feature + TopicNotFound = 11; // Topic not found + SubscriptionNotFound = 12; // Subscription not found + ConsumerNotFound = 13; // Consumer not found + TooManyRequests = 14; // Error with too many simultaneously request + TopicTerminatedError = 15; // The topic has been terminated + + ProducerBusy = 16; // Producer with same name is already connected + InvalidTopicName = 17; // The topic name is not valid + + IncompatibleSchema = 18; // Specified schema was incompatible with topic schema + ConsumerAssignError = 19; // Dispatcher assign consumer error + + TransactionCoordinatorNotFound = 20; // Transaction coordinator not found error + InvalidTxnStatus = 21; // Invalid txn status error + NotAllowedError = 22; // Not allowed error +} + +enum AuthMethod { + AuthMethodNone = 0; + AuthMethodYcaV1 = 1; + AuthMethodAthens = 2; +} + +// Each protocol version identify new features that are +// incrementally added to the protocol +enum ProtocolVersion { + v0 = 0; // Initial versioning + v1 = 1; // Added application keep-alive + v2 = 2; // Added RedeliverUnacknowledgedMessages Command + v3 = 3; // Added compression with LZ4 and ZLib + v4 = 4; // Added batch message support + v5 = 5; // Added disconnect client w/o closing connection + v6 = 6; // Added checksum computation for metadata + payload + v7 = 7; // Added CommandLookupTopic - Binary Lookup + v8 = 8; // Added CommandConsumerStats - Client fetches broker side consumer stats + v9 = 9; // Added end of topic notification + v10 = 10;// Added proxy to broker + v11 = 11;// C++ consumers before this version are not correctly handling the checksum field + v12 = 12;// Added get topic's last messageId from broker + // Added CommandActiveConsumerChange + // Added CommandGetTopicsOfNamespace + v13 = 13; // Schema-registry : added avro schema format for json + v14 = 14; // Add CommandAuthChallenge and CommandAuthResponse for mutual auth + // Added Key_Shared subscription + v15 = 15; // Add CommandGetOrCreateSchema and CommandGetOrCreateSchemaResponse +} + +message CommandConnect { + required string client_version = 1; + optional AuthMethod auth_method = 2; // Deprecated. Use "auth_method_name" instead. + optional string auth_method_name = 5; + optional bytes auth_data = 3; + optional int32 protocol_version = 4 [default = 0]; + + // Client can ask to be proxyied to a specific broker + // This is only honored by a Pulsar proxy + optional string proxy_to_broker_url = 6; + + // Original principal that was verified by + // a Pulsar proxy. In this case the auth info above + // will be the auth of the proxy itself + optional string original_principal = 7; + + // Original auth role and auth Method that was passed + // to the proxy. In this case the auth info above + // will be the auth of the proxy itself + optional string original_auth_data = 8; + optional string original_auth_method = 9; + + // Feature flags + optional FeatureFlags feature_flags = 10; +} + +message FeatureFlags { + optional bool supports_auth_refresh = 1 [default = false]; +} + +message CommandConnected { + required string server_version = 1; + optional int32 protocol_version = 2 [default = 0]; + optional int32 max_message_size = 3; +} + +message CommandAuthResponse { + optional string client_version = 1; + optional AuthData response = 2; + optional int32 protocol_version = 3 [default = 0]; +} + +message CommandAuthChallenge { + optional string server_version = 1; + optional AuthData challenge = 2; + optional int32 protocol_version = 3 [default = 0]; +} + +// To support mutual authentication type, such as Sasl, reuse this command to mutual auth. +message AuthData { + optional string auth_method_name = 1; + optional bytes auth_data = 2; +} + +enum KeySharedMode { + AUTO_SPLIT = 0; + STICKY = 1; +} + +message KeySharedMeta { + required KeySharedMode keySharedMode = 1; + repeated IntRange hashRanges = 3; + optional bool allowOutOfOrderDelivery = 4 [default = false]; +} + +message CommandSubscribe { + enum SubType { + Exclusive = 0; + Shared = 1; + Failover = 2; + Key_Shared = 3; + } + required string topic = 1; + required string subscription = 2; + required SubType subType = 3; + + required uint64 consumer_id = 4; + required uint64 request_id = 5; + optional string consumer_name = 6; + optional int32 priority_level = 7; + + // Signal wether the subscription should be backed by a + // durable cursor or not + optional bool durable = 8 [default = true]; + + // If specified, the subscription will position the cursor + // markd-delete position on the particular message id and + // will send messages from that point + optional MessageIdData start_message_id = 9; + + /// Add optional metadata key=value to this consumer + repeated KeyValue metadata = 10; + + optional bool read_compacted = 11; + + optional Schema schema = 12; + enum InitialPosition { + Latest = 0; + Earliest = 1; + } + // Signal whether the subscription will initialize on latest + // or not -- earliest + optional InitialPosition initialPosition = 13 [default = Latest]; + + // Mark the subscription as "replicated". Pulsar will make sure + // to periodically sync the state of replicated subscriptions + // across different clusters (when using geo-replication). + optional bool replicate_subscription_state = 14; + + // If true, the subscribe operation will cause a topic to be + // created if it does not exist already (and if topic auto-creation + // is allowed by broker. + // If false, the subscribe operation will fail if the topic + // does not exist. + optional bool force_topic_creation = 15 [default = true]; + + // If specified, the subscription will reset cursor's position back + // to specified seconds and will send messages from that point + optional uint64 start_message_rollback_duration_sec = 16 [default = 0]; + + optional KeySharedMeta keySharedMeta = 17; +} + +message CommandPartitionedTopicMetadata { + required string topic = 1; + required uint64 request_id = 2; + // TODO - Remove original_principal, original_auth_data, original_auth_method + // Original principal that was verified by + // a Pulsar proxy. + optional string original_principal = 3; + + // Original auth role and auth Method that was passed + // to the proxy. + optional string original_auth_data = 4; + optional string original_auth_method = 5; +} + +message CommandPartitionedTopicMetadataResponse { + enum LookupType { + Success = 0; + Failed = 1; + } + optional uint32 partitions = 1; // Optional in case of error + required uint64 request_id = 2; + optional LookupType response = 3; + optional ServerError error = 4; + optional string message = 5; +} + +message CommandLookupTopic { + required string topic = 1; + required uint64 request_id = 2; + optional bool authoritative = 3 [default = false]; + + // TODO - Remove original_principal, original_auth_data, original_auth_method + // Original principal that was verified by + // a Pulsar proxy. + optional string original_principal = 4; + + // Original auth role and auth Method that was passed + // to the proxy. + optional string original_auth_data = 5; + optional string original_auth_method = 6; + // + optional string advertised_listener_name = 7; +} + +message CommandLookupTopicResponse { + enum LookupType { + Redirect = 0; + Connect = 1; + Failed = 2; + } + + optional string brokerServiceUrl = 1; // Optional in case of error + optional string brokerServiceUrlTls = 2; + optional LookupType response = 3; + required uint64 request_id = 4; + optional bool authoritative = 5 [default = false]; + optional ServerError error = 6; + optional string message = 7; + + // If it's true, indicates to the client that it must + // always connect through the service url after the + // lookup has been completed. + optional bool proxy_through_service_url = 8 [default = false]; +} + +/// Create a new Producer on a topic, assigning the given producer_id, +/// all messages sent with this producer_id will be persisted on the topic +message CommandProducer { + required string topic = 1; + required uint64 producer_id = 2; + required uint64 request_id = 3; + + /// If a producer name is specified, the name will be used, + /// otherwise the broker will generate a unique name + optional string producer_name = 4; + + optional bool encrypted = 5 [default = false]; + + /// Add optional metadata key=value to this producer + repeated KeyValue metadata = 6; + + optional Schema schema = 7; + + // If producer reconnect to broker, the epoch of this producer will +1 + optional uint64 epoch = 8 [default = 0]; + + // Indicate the name of the producer is generated or user provided + // Use default true here is in order to be forward compatible with the client + optional bool user_provided_producer_name = 9 [default = true]; +} + +message CommandSend { + required uint64 producer_id = 1; + required uint64 sequence_id = 2; + optional int32 num_messages = 3 [default = 1]; + optional uint64 txnid_least_bits = 4 [default = 0]; + optional uint64 txnid_most_bits = 5 [default = 0]; + + /// Add highest sequence id to support batch message with external sequence id + optional uint64 highest_sequence_id = 6 [default = 0]; + optional bool is_chunk =7 [default = false]; +} + +message CommandSendReceipt { + required uint64 producer_id = 1; + required uint64 sequence_id = 2; + optional MessageIdData message_id = 3; + optional uint64 highest_sequence_id = 4 [default = 0]; +} + +message CommandSendError { + required uint64 producer_id = 1; + required uint64 sequence_id = 2; + required ServerError error = 3; + required string message = 4; +} + +message CommandMessage { + required uint64 consumer_id = 1; + required MessageIdData message_id = 2; + optional uint32 redelivery_count = 3 [default = 0]; + repeated int64 ack_set = 4; +} + +message CommandAck { + enum AckType { + Individual = 0; + Cumulative = 1; + } + + required uint64 consumer_id = 1; + required AckType ack_type = 2; + + // In case of individual acks, the client can pass a list of message ids + repeated MessageIdData message_id = 3; + + // Acks can contain a flag to indicate the consumer + // received an invalid message that got discarded + // before being passed on to the application. + enum ValidationError { + UncompressedSizeCorruption = 0; + DecompressionError = 1; + ChecksumMismatch = 2; + BatchDeSerializeError = 3; + DecryptionError = 4; + } + + optional ValidationError validation_error = 4; + repeated KeyLongValue properties = 5; + + optional uint64 txnid_least_bits = 6 [default = 0]; + optional uint64 txnid_most_bits = 7 [default = 0]; +} + +message CommandAckResponse { + required uint64 consumer_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +// changes on active consumer +message CommandActiveConsumerChange { + required uint64 consumer_id = 1; + optional bool is_active = 2 [default = false]; +} + +message CommandFlow { + required uint64 consumer_id = 1; + + // Max number of messages to prefetch, in addition + // of any number previously specified + required uint32 messagePermits = 2; +} + +message CommandUnsubscribe { + required uint64 consumer_id = 1; + required uint64 request_id = 2; +} + +// Reset an existing consumer to a particular message id +message CommandSeek { + required uint64 consumer_id = 1; + required uint64 request_id = 2; + + optional MessageIdData message_id = 3; + optional uint64 message_publish_time = 4; +} + +// Message sent by broker to client when a topic +// has been forcefully terminated and there are no more +// messages left to consume +message CommandReachedEndOfTopic { + required uint64 consumer_id = 1; +} + +message CommandCloseProducer { + required uint64 producer_id = 1; + required uint64 request_id = 2; +} + +message CommandCloseConsumer { + required uint64 consumer_id = 1; + required uint64 request_id = 2; +} + +message CommandRedeliverUnacknowledgedMessages { + required uint64 consumer_id = 1; + repeated MessageIdData message_ids = 2; +} + +message CommandSuccess { + required uint64 request_id = 1; + optional Schema schema = 2; +} + +/// Response from CommandProducer +message CommandProducerSuccess { + required uint64 request_id = 1; + required string producer_name = 2; + + // The last sequence id that was stored by this producer in the previous session + // This will only be meaningful if deduplication has been enabled. + optional int64 last_sequence_id = 3 [default = -1]; + optional bytes schema_version = 4; +} + +message CommandError { + required uint64 request_id = 1; + required ServerError error = 2; + required string message = 3; +} + +// Commands to probe the state of connection. +// When either client or broker doesn't receive commands for certain +// amount of time, they will send a Ping probe. +message CommandPing { +} +message CommandPong { +} + +message CommandConsumerStats { + required uint64 request_id = 1; + // required string topic_name = 2; + // required string subscription_name = 3; + required uint64 consumer_id = 4; +} + +message CommandConsumerStatsResponse { + required uint64 request_id = 1; + optional ServerError error_code = 2; + optional string error_message = 3; + + /// Total rate of messages delivered to the consumer. msg/s + optional double msgRateOut = 4; + + /// Total throughput delivered to the consumer. bytes/s + optional double msgThroughputOut = 5; + + /// Total rate of messages redelivered by this consumer. msg/s + optional double msgRateRedeliver = 6; + + /// Name of the consumer + optional string consumerName = 7; + + /// Number of available message permits for the consumer + optional uint64 availablePermits = 8; + + /// Number of unacknowledged messages for the consumer + optional uint64 unackedMessages = 9; + + /// Flag to verify if consumer is blocked due to reaching threshold of unacked messages + optional bool blockedConsumerOnUnackedMsgs = 10; + + /// Address of this consumer + optional string address = 11; + + /// Timestamp of connection + optional string connectedSince = 12; + + /// Whether this subscription is Exclusive or Shared or Failover + optional string type = 13; + + /// Total rate of messages expired on this subscription. msg/s + optional double msgRateExpired = 14; + + /// Number of messages in the subscription backlog + optional uint64 msgBacklog = 15; +} + +message CommandGetLastMessageId { + required uint64 consumer_id = 1; + required uint64 request_id = 2; +} + +message CommandGetLastMessageIdResponse { + required MessageIdData last_message_id = 1; + required uint64 request_id = 2; +} + +message CommandGetTopicsOfNamespace { + enum Mode { + PERSISTENT = 0; + NON_PERSISTENT = 1; + ALL = 2; + } + required uint64 request_id = 1; + required string namespace = 2; + optional Mode mode = 3 [default = PERSISTENT]; +} + +message CommandGetTopicsOfNamespaceResponse { + required uint64 request_id = 1; + repeated string topics = 2; +} + +message CommandGetSchema { + required uint64 request_id = 1; + required string topic = 2; + + optional bytes schema_version = 3; +} + +message CommandGetSchemaResponse { + required uint64 request_id = 1; + optional ServerError error_code = 2; + optional string error_message = 3; + + optional Schema schema = 4; + optional bytes schema_version = 5; +} + +message CommandGetOrCreateSchema { + required uint64 request_id = 1; + required string topic = 2; + required Schema schema = 3; +} + +message CommandGetOrCreateSchemaResponse { + required uint64 request_id = 1; + optional ServerError error_code = 2; + optional string error_message = 3; + + optional bytes schema_version = 4; +} + +/// --- transaction related --- + +enum TxnAction { + COMMIT = 0; + ABORT = 1; +} + +message CommandNewTxn { + required uint64 request_id = 1; + optional uint64 txn_ttl_seconds = 2 [default = 0]; + optional uint64 tc_id = 3 [default = 0]; +} + +message CommandNewTxnResponse { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +message CommandAddPartitionToTxn { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + repeated string partitions = 4; +} + +message CommandAddPartitionToTxnResponse { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +message Subscription { + required string topic = 1; + required string subscription = 2; +} +message CommandAddSubscriptionToTxn { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + repeated Subscription subscription = 4; +} + +message CommandAddSubscriptionToTxnResponse { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +message CommandEndTxn { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional TxnAction txn_action = 4; +} + +message CommandEndTxnResponse { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +message CommandEndTxnOnPartition { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional string topic = 4; + optional TxnAction txn_action = 5; +} + +message CommandEndTxnOnPartitionResponse { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +message CommandEndTxnOnSubscription { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional Subscription subscription= 4; + optional TxnAction txn_action = 5; +} + +message CommandEndTxnOnSubscriptionResponse { + required uint64 request_id = 1; + optional uint64 txnid_least_bits = 2 [default = 0]; + optional uint64 txnid_most_bits = 3 [default = 0]; + optional ServerError error = 4; + optional string message = 5; +} + +message BaseCommand { + enum Type { + CONNECT = 2; + CONNECTED = 3; + SUBSCRIBE = 4; + + PRODUCER = 5; + + SEND = 6; + SEND_RECEIPT= 7; + SEND_ERROR = 8; + + MESSAGE = 9; + ACK = 10; + FLOW = 11; + + UNSUBSCRIBE = 12; + + SUCCESS = 13; + ERROR = 14; + + CLOSE_PRODUCER = 15; + CLOSE_CONSUMER = 16; + + PRODUCER_SUCCESS = 17; + + PING = 18; + PONG = 19; + + REDELIVER_UNACKNOWLEDGED_MESSAGES = 20; + + PARTITIONED_METADATA = 21; + PARTITIONED_METADATA_RESPONSE = 22; + + LOOKUP = 23; + LOOKUP_RESPONSE = 24; + + CONSUMER_STATS = 25; + CONSUMER_STATS_RESPONSE = 26; + + REACHED_END_OF_TOPIC = 27; + + SEEK = 28; + + GET_LAST_MESSAGE_ID = 29; + GET_LAST_MESSAGE_ID_RESPONSE = 30; + + ACTIVE_CONSUMER_CHANGE = 31; + + + GET_TOPICS_OF_NAMESPACE = 32; + GET_TOPICS_OF_NAMESPACE_RESPONSE = 33; + + GET_SCHEMA = 34; + GET_SCHEMA_RESPONSE = 35; + + AUTH_CHALLENGE = 36; + AUTH_RESPONSE = 37; + + ACK_RESPONSE = 38; + + GET_OR_CREATE_SCHEMA = 39; + GET_OR_CREATE_SCHEMA_RESPONSE = 40; + + // transaction related + NEW_TXN = 50; + NEW_TXN_RESPONSE = 51; + + ADD_PARTITION_TO_TXN = 52; + ADD_PARTITION_TO_TXN_RESPONSE = 53; + + ADD_SUBSCRIPTION_TO_TXN = 54; + ADD_SUBSCRIPTION_TO_TXN_RESPONSE = 55; + + END_TXN = 56; + END_TXN_RESPONSE = 57; + + END_TXN_ON_PARTITION = 58; + END_TXN_ON_PARTITION_RESPONSE = 59; + + END_TXN_ON_SUBSCRIPTION = 60; + END_TXN_ON_SUBSCRIPTION_RESPONSE = 61; + + } + + + required Type type = 1; + + optional CommandConnect connect = 2; + optional CommandConnected connected = 3; + + optional CommandSubscribe subscribe = 4; + optional CommandProducer producer = 5; + optional CommandSend send = 6; + optional CommandSendReceipt send_receipt = 7; + optional CommandSendError send_error = 8; + optional CommandMessage message = 9; + optional CommandAck ack = 10; + optional CommandFlow flow = 11; + optional CommandUnsubscribe unsubscribe = 12; + + optional CommandSuccess success = 13; + optional CommandError error = 14; + + optional CommandCloseProducer close_producer = 15; + optional CommandCloseConsumer close_consumer = 16; + + optional CommandProducerSuccess producer_success = 17; + optional CommandPing ping = 18; + optional CommandPong pong = 19; + optional CommandRedeliverUnacknowledgedMessages redeliverUnacknowledgedMessages = 20; + + optional CommandPartitionedTopicMetadata partitionMetadata = 21; + optional CommandPartitionedTopicMetadataResponse partitionMetadataResponse = 22; + + optional CommandLookupTopic lookupTopic = 23; + optional CommandLookupTopicResponse lookupTopicResponse = 24; + + optional CommandConsumerStats consumerStats = 25; + optional CommandConsumerStatsResponse consumerStatsResponse = 26; + + optional CommandReachedEndOfTopic reachedEndOfTopic = 27; + + optional CommandSeek seek = 28; + + optional CommandGetLastMessageId getLastMessageId = 29; + optional CommandGetLastMessageIdResponse getLastMessageIdResponse = 30; + + optional CommandActiveConsumerChange active_consumer_change = 31; + + optional CommandGetTopicsOfNamespace getTopicsOfNamespace = 32; + optional CommandGetTopicsOfNamespaceResponse getTopicsOfNamespaceResponse = 33; + + optional CommandGetSchema getSchema = 34; + optional CommandGetSchemaResponse getSchemaResponse = 35; + + optional CommandAuthChallenge authChallenge = 36; + optional CommandAuthResponse authResponse = 37; + + optional CommandAckResponse ackResponse = 38; + + optional CommandGetOrCreateSchema getOrCreateSchema = 39; + optional CommandGetOrCreateSchemaResponse getOrCreateSchemaResponse = 40; + + // transaction related + optional CommandNewTxn newTxn = 50; + optional CommandNewTxnResponse newTxnResponse = 51; + optional CommandAddPartitionToTxn addPartitionToTxn= 52; + optional CommandAddPartitionToTxnResponse addPartitionToTxnResponse = 53; + optional CommandAddSubscriptionToTxn addSubscriptionToTxn = 54; + optional CommandAddSubscriptionToTxnResponse addSubscriptionToTxnResponse = 55; + optional CommandEndTxn endTxn = 56; + optional CommandEndTxnResponse endTxnResponse = 57; + optional CommandEndTxnOnPartition endTxnOnPartition = 58; + optional CommandEndTxnOnPartitionResponse endTxnOnPartitionResponse = 59; + optional CommandEndTxnOnSubscription endTxnOnSubscription = 60; + optional CommandEndTxnOnSubscriptionResponse endTxnOnSubscriptionResponse = 61; +} diff --git a/supernova.cabal b/supernova.cabal new file mode 100644 --- /dev/null +++ b/supernova.cabal @@ -0,0 +1,73 @@ +cabal-version: 3.0 + +name: supernova +version: 0.0.1 +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 +bug-reports: https://github.com/cr-org/supernova/issues +license: Apache-2.0 +license-file: LICENSE +author: Gabriel Volpe +maintainer: gabriel.volpe@chatroulette.com +category: Network +build-type: Custom +extra-source-files: README.md + , src/pulsar_api.proto + +custom-setup + setup-depends: base + , Cabal + , proto-lens-setup + +library + exposed-modules: Pulsar + other-modules: Paths_supernova + , Proto.PulsarApi + , Proto.PulsarApi_Fields + , Pulsar.Core + , Pulsar.Connection + , Pulsar.Consumer + , Pulsar.Internal.Core + , Pulsar.Internal.Logger + , Pulsar.Internal.TCPClient + , Pulsar.Producer + , Pulsar.Protocol.Commands + , Pulsar.Protocol.Decoder + , Pulsar.Protocol.Encoder + , Pulsar.Protocol.Frame + , Pulsar.Types + autogen-modules: Paths_supernova + , Proto.PulsarApi + , Proto.PulsarApi_Fields + build-depends: 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, + logging >= 3.0.5 && < 3.1, + 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-Suite supernova-tests + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs + build-depends: base + , aeson + , async + , bytestring + , supernova + , streamly + , text + default-language: Haskell2010 + ghc-options: -Wall -threaded -with-rtsopts -N diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,64 @@ +{-# LANGUAGE DeriveAnyClass, DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings, RecordWildCards #-} + +module Main where + +import Control.Concurrent ( threadDelay ) +import Control.Concurrent.Async ( concurrently_ ) +import Control.Monad ( forever ) +import Data.Aeson +import qualified Data.ByteString.Lazy.Char8 as CL +import Data.Foldable ( traverse_ ) +import Data.Text ( Text ) +import GHC.Generics ( Generic ) +import Pulsar +import Streamly ( asyncly + , maxThreads + ) +import qualified Streamly.Prelude as S + +main :: IO () +main = demo + +data Msg = Msg + { name :: Text + , amount :: Int + } deriving (Generic, FromJSON, ToJSON, Show) + +messages :: [PulsarMessage] +messages = + let msg = [Msg "foo" 2, Msg "bar" 5, Msg "taz" 1] + in PulsarMessage . encode <$> msg + +msgDecoder :: CL.ByteString -> IO () +msgDecoder bs = + let msg = decode bs :: Maybe Msg + in putStrLn $ "-----------------> " <> show msg + +topic :: Topic +topic = defaultTopic "app" + +demo :: IO () +demo = 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 + +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) + +logOpts :: LogOptions +logOpts = LogOptions Info StdOut + +streamDemo :: IO () +streamDemo = runPulsar' logOpts resources $ \(Consumer {..}, Producer {..}) -> + let c = forever $ fetch >>= \(Message i m) -> msgDecoder m >> ack i + p = forever $ sleep 5 >> traverse_ produce messages + in S.drain . asyncly . maxThreads 10 $ S.yieldM c <> S.yieldM p