supernova (empty) → 0.0.1
raw patch · 19 files changed
+2339/−0 lines, 19 filesdep +aesondep +asyncdep +basebuild-type:Customsetup-changed
Dependencies added: aeson, async, base, bifunctor, binary, bytestring, crc32c, exceptions, lens-family-core, logging, managed, network, proto-lens, proto-lens-runtime, streamly, supernova, text, unliftio
Files
- LICENSE +201/−0
- README.md +75/−0
- Setup.hs +3/−0
- src/Pulsar.hs +61/−0
- src/Pulsar/Connection.hs +156/−0
- src/Pulsar/Consumer.hs +60/−0
- src/Pulsar/Core.hs +140/−0
- src/Pulsar/Internal/Core.hs +48/−0
- src/Pulsar/Internal/Logger.hs +14/−0
- src/Pulsar/Internal/TCPClient.hs +31/−0
- src/Pulsar/Producer.hs +49/−0
- src/Pulsar/Protocol/Commands.hs +124/−0
- src/Pulsar/Protocol/Decoder.hs +84/−0
- src/Pulsar/Protocol/Encoder.hs +71/−0
- src/Pulsar/Protocol/Frame.hs +43/−0
- src/Pulsar/Types.hs +79/−0
- src/pulsar_api.proto +963/−0
- supernova.cabal +73/−0
- test/Main.hs +64/−0
+ LICENSE view
@@ -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.
+ README.md view
@@ -0,0 +1,75 @@+supernova+=========++[](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.++[](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+```
+ Setup.hs view
@@ -0,0 +1,3 @@+import Data.ProtoLens.Setup++main = defaultMainGeneratingProtos "src"
+ src/Pulsar.hs view
@@ -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
+ src/Pulsar/Connection.hs view
@@ -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
+ src/Pulsar/Consumer.hs view
@@ -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
+ src/Pulsar/Core.hs view
@@ -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'
+ src/Pulsar/Internal/Core.hs view
@@ -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
+ src/Pulsar/Internal/Logger.hs view
@@ -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)
+ src/Pulsar/Internal/TCPClient.hs view
@@ -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
+ src/Pulsar/Producer.hs view
@@ -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
+ src/Pulsar/Protocol/Commands.hs view
@@ -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
+ src/Pulsar/Protocol/Decoder.hs view
@@ -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
+ src/Pulsar/Protocol/Encoder.hs view
@@ -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
+ src/Pulsar/Protocol/Frame.hs view
@@ -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
+ src/Pulsar/Types.hs view
@@ -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
+ src/pulsar_api.proto view
@@ -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;+}
+ supernova.cabal view
@@ -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
+ test/Main.hs view
@@ -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