Pulsar

In the following example, we will create a quick example showcasing a consumer and producer running concurrently, step by step.

Consider the following imports (needs the async library).

import           Control.Concurrent             ( threadDelay )
import           Control.Concurrent.Async       ( concurrently_ )
import           Pulsar

Firstly, we create a connection to Pulsar, defined as PulsarConnection.

conn :: PulsarConnection
conn = connect defaultConnectData

Then a consumer and a producer, which operate in the Pulsar monad.

pulsar :: Pulsar ()
pulsar = do
  c <- newConsumer topic sub
  p <- newProducer topic
  liftIO $ program c p
 where
  topic = defaultTopic "app"
  sub   = Subscription Exclusive "test-sub"

And the main user program that consume and produce messages concurrently, running in IO.

program :: Consumer IO -> Producer IO -> IO ()
program Consumer {..} Producer {..} =
  let c = fetch >>= (Message i m) -> print m >> ack i >> c
      p = threadDelay (3 * 1000000) >> send "Hello World!" >> p
  in  concurrently_ c p

We have a delay of 3 seconds before publishing to make sure the consumer is already running. Otherwise, it might miss some messages.

Finally, we put it all together and call runPulsar with the connection and the program in the Pulsar monad.

main :: IO ()
main = runPulsar conn pulsar

Since a Pulsar connection, consumers, and producers are long-lived resources, Supernova manages them accordingly for you. Once the program exits, the resources will be released in the respective order (always opposite to the order of acquisition).

Documentation

connect :: (MonadIO m, MonadThrow m, MonadManaged m) => ConnectData -> m PulsarCtx

Starts a Pulsar connection with the supplied ConnectData

defaultConnectData :: ConnectData

Default connection data: "127.0.0.1:6650"

newConsumer :: (MonadIO m, MonadIO f, MonadReader PulsarCtx m) => Topic -> Subscription -> m (Consumer f)

Create a new Consumer by supplying a PulsarCtx (returned by connect), a Topic and a SubscriptionName.

newProducer :: (MonadIO m, MonadReader PulsarCtx m, MonadIO f) => Topic -> m (Producer f)

Create a new Producer by supplying a PulsarCtx (returned by connect) and a Topic.

runPulsar :: PulsarConnection -> Pulsar a -> IO ()

Runs a Pulsar computation with default logging to standard output

runPulsar' :: LogOptions -> PulsarConnection -> Pulsar a -> IO ()

Runs a Pulsar computation with the supplied logging options

data Consumer m

An abstract Consumer able to fetch messages and acknowledge them.

Constructors

Consumer

Fields

• fetch :: m Message

  Fetches a single message. Blocks if no messages are available.

• ack :: MsgId -> m ()

  Acknowledges a single message.

newtype Producer m

An abstract Producer able to send messages of type PulsarMessage.

Constructors

Producer

Fields

• send :: PulsarMessage -> m ()

  Produces a single message.

data Pulsar a

The main Pulsar monad, which abstracts over a ReaderT monad.

Instances

Monad Pulsar

Defined in Pulsar.Internal.Core

Methods

(>>=) :: Pulsar a -> (a -> Pulsar b) -> Pulsar b

(>>) :: Pulsar a -> Pulsar b -> Pulsar b

return :: a -> Pulsar a

Functor Pulsar

Defined in Pulsar.Internal.Core

Methods

fmap :: (a -> b) -> Pulsar a -> Pulsar b

(<$) :: a -> Pulsar b -> Pulsar a

Applicative Pulsar

Defined in Pulsar.Internal.Core

Methods

pure :: a -> Pulsar a

(<*>) :: Pulsar (a -> b) -> Pulsar a -> Pulsar b

liftA2 :: (a -> b -> c) -> Pulsar a -> Pulsar b -> Pulsar c

(*>) :: Pulsar a -> Pulsar b -> Pulsar b

(<*) :: Pulsar a -> Pulsar b -> Pulsar a

MonadIO Pulsar

Defined in Pulsar.Internal.Core

Methods

liftIO :: IO a -> Pulsar a

type PulsarConnection = Connection PulsarCtx

Alias for Connection PulsarCtx.

data ConnectData

Connection details: host and port.

Constructors

ConnData

Fields

• connHost :: HostName
• connPort :: ServiceName

Instances

Show ConnectData

Defined in Pulsar.Connection

Methods

showsPrec :: Int -> ConnectData -> ShowS

show :: ConnectData -> String

showList :: [ConnectData] -> ShowS

data LogLevel

Internal logging level, part of LogOptions. Can be used together with runPulsar`.

Constructors

Error

Warn

Info

Debug

Instances

Show LogLevel

Defined in Pulsar.Internal.Core

Methods

showsPrec :: Int -> LogLevel -> ShowS

show :: LogLevel -> String

showList :: [LogLevel] -> ShowS

data LogOptions

Internal logging options. Can be used together with runPulsar`.

Constructors

LogOptions

Fields

• logLevel :: LogLevel
• logOutput :: LogOutput

Instances

Show LogOptions

Defined in Pulsar.Internal.Core

Methods

showsPrec :: Int -> LogOptions -> ShowS

show :: LogOptions -> String

showList :: [LogOptions] -> ShowS

data LogOutput

Internal logging output, part of LogOptions. Can be used together with runPulsar`.

Constructors

StdOut

File FilePath

Instances

Show LogOutput

Defined in Pulsar.Internal.Core

Methods

showsPrec :: Int -> LogOutput -> ShowS

show :: LogOutput -> String

showList :: [LogOutput] -> ShowS

data Topic

A Topic is in the form "type://tenant/namespace/topic-name", which is what the Show instance does.

Constructors

Topic

Fields

• type' :: TopicType
• tenant :: Tenant
• namespace :: NameSpace
• name :: TopicName

Instances

Show Topic

Defined in Pulsar.Types

Methods

showsPrec :: Int -> Topic -> ShowS

show :: Topic -> String

showList :: [Topic] -> ShowS

defaultTopic :: TopicName -> Topic

A default Topic: "non-persistent://public/default/my-topic".

data TopicType

A topic can be either Persistent or NonPersistent.

Constructors

Persistent

NonPersistent

Instances

Show TopicType

Defined in Pulsar.Types

Methods

showsPrec :: Int -> TopicType -> ShowS

show :: TopicType -> String

showList :: [TopicType] -> ShowS

newtype Tenant

A tenant can be any string value. Default value is "public".

Constructors

Tenant Text

Instances

Show Tenant

Defined in Pulsar.Types

Methods

showsPrec :: Int -> Tenant -> ShowS

show :: Tenant -> String

showList :: [Tenant] -> ShowS

IsString Tenant

Defined in Pulsar.Types

Methods

fromString :: String -> Tenant

newtype NameSpace

A namespace can be any string value. Default value is "default".

Constructors

NameSpace Text

Instances

Show NameSpace

Defined in Pulsar.Types

Methods

showsPrec :: Int -> NameSpace -> ShowS

show :: NameSpace -> String

showList :: [NameSpace] -> ShowS

IsString NameSpace

Defined in Pulsar.Types

Methods

fromString :: String -> NameSpace

newtype TopicName

A topic name can be any string value.

Constructors

TopicName Text

Instances

Show TopicName

Defined in Pulsar.Types

Methods

showsPrec :: Int -> TopicName -> ShowS

show :: TopicName -> String

showList :: [TopicName] -> ShowS

IsString TopicName

Defined in Pulsar.Types

Methods

fromString :: String -> TopicName

newtype MsgId

A message id, needed for acknowledging messages. See ack.

Constructors

MsgId MessageIdData

data Message

A consumed message, containing both MsgId and payload as bytestring.

Constructors

Message MsgId ByteString

newtype PulsarMessage

A produced message, containing just a payload as bytestring.

Constructors

PulsarMessage ByteString

Instances

Show PulsarMessage

Defined in Pulsar.Types

Methods

showsPrec :: Int -> PulsarMessage -> ShowS

show :: PulsarMessage -> String

showList :: [PulsarMessage] -> ShowS

IsString PulsarMessage

Defined in Pulsar.Types

Methods

fromString :: String -> PulsarMessage

newtype SubName

A subscription name can be any string value.

Constructors

SubName Text

Instances

Show SubName

Defined in Pulsar.Types

Methods

showsPrec :: Int -> SubName -> ShowS

show :: SubName -> String

showList :: [SubName] -> ShowS

IsString SubName

Defined in Pulsar.Types

Methods

fromString :: String -> SubName

data SubType

A subscription type. See https://pulsar.apache.org/docs/en/concepts-messaging/#subscriptions to learn more.

Constructors

Exclusive

Failover

Shared

KeyShared

Instances

Show SubType

Defined in Pulsar.Types

Methods

showsPrec :: Int -> SubType -> ShowS

show :: SubType -> String

showList :: [SubType] -> ShowS

data Subscription

A subscription with a type and a name.

Constructors

Subscription SubType SubName

Instances

Show Subscription

Defined in Pulsar.Types

Methods

showsPrec :: Int -> Subscription -> ShowS

show :: Subscription -> String

showList :: [Subscription] -> ShowS