License	Apache-2.0
Maintainer	gabriel.volpe@chatroulette.com
Stability	experimental
Safe Haskell	None
Language	Haskell2010

Pulsar

Description

Consider the following imports (needs the 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

Synopsis

connect :: (MonadThrow m, MonadIO m, MonadManaged m) => ConnectData -> m PulsarCtx
defaultConnectData :: ConnectData
newConsumer :: (MonadManaged m, MonadIO f) => PulsarCtx -> Topic -> SubscriptionName -> m (Consumer f)
newProducer :: (MonadManaged m, MonadIO f) => PulsarCtx -> Topic -> m (Producer f)
runPulsar :: forall a b. Pulsar a -> (a -> IO b) -> IO b
runPulsar' :: forall a b. LogOptions -> Pulsar a -> (a -> IO b) -> IO b
data Consumer m = Consumer {
- fetch :: m Message
- ack :: MsgId -> m ()
}
newtype Producer m = Producer {
- produce :: PulsarMessage -> m ()
}
data Pulsar a
data PulsarCtx
data ConnectData
data LogLevel
- = Error
- | Warn
- | Info
- | Debug
data LogOptions = LogOptions {
- logLevel :: LogLevel
- logOutput :: LogOutput
}
data LogOutput
- = StdOut
- | File FilePath
data Topic = Topic {
- type' :: TopicType
- tenant :: Tenant
- namespace :: NameSpace
- name :: TopicName
}
defaultTopic :: String -> Topic
data TopicType
- = Persistent
- | NonPersistent
newtype Tenant = Tenant String
newtype NameSpace = NameSpace String
newtype TopicName = TopicName String
newtype MsgId = MsgId MessageIdData
data Message = Message MsgId ByteString
newtype PulsarMessage = PulsarMessage ByteString
newtype SubscriptionName = SubscriptionName Text

Documentation

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

Starts a Pulsar connection with the supplied ConnectData

defaultConnectData :: ConnectData Source #

Default connection data: "127.0.0.1:6650"

newConsumer :: (MonadManaged m, MonadIO f) => PulsarCtx -> Topic -> SubscriptionName -> m (Consumer f) Source #

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

newProducer :: (MonadManaged m, MonadIO f) => PulsarCtx -> Topic -> m (Producer f) Source #

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

runPulsar :: forall a b. Pulsar a -> (a -> IO b) -> IO b Source #

Runs a Pulsar computation with default logging to standard output

runPulsar' :: forall a b. LogOptions -> Pulsar a -> (a -> IO b) -> IO b Source #

Runs a Pulsar computation with the supplied logging options

data Consumer m Source #

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 Source #

An abstract Producer able to produce messages of type PulsarMessage.

Constructors

Producer
Fields produce :: PulsarMessage -> m () Produces a single message.

data Pulsar a Source #

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

Instances

Instances details

Monad Pulsar Source #
Instance details 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 Source #
Instance details Defined in Pulsar.Internal.Core Methods fmap :: (a -> b) -> Pulsar a -> Pulsar b # (<$) :: a -> Pulsar b -> Pulsar a #
Applicative Pulsar Source #
Instance details 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 Source #
Instance details Defined in Pulsar.Internal.Core Methods liftIO :: IO a -> Pulsar a #
MonadThrow Pulsar Source #
Instance details Defined in Pulsar.Internal.Core Methods throwM :: Exception e => e -> Pulsar a #
MonadManaged Pulsar Source #
Instance details Defined in Pulsar.Internal.Core Methods using :: Managed a -> Pulsar a #

data PulsarCtx Source #

Internal Pulsar context. You will never need to access its content (not exported) but might need to take it as argument.

data ConnectData Source #

Connection details: host and port.

Instances

Instances details

Show ConnectData Source #
Instance details Defined in Pulsar.Connection Methods showsPrec :: Int -> ConnectData -> ShowS # show :: ConnectData -> String # showList :: [ConnectData] -> ShowS #

data LogLevel Source #

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

Constructors

Error
Warn
Info
Debug

Instances

Instances details

Show LogLevel Source #
Instance details Defined in Pulsar.Internal.Core Methods showsPrec :: Int -> LogLevel -> ShowS # show :: LogLevel -> String # showList :: [LogLevel] -> ShowS #

data LogOptions Source #

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

Constructors

LogOptions
Fields logLevel :: LogLevel logOutput :: LogOutput

Instances

Instances details

Show LogOptions Source #
Instance details Defined in Pulsar.Internal.Core Methods showsPrec :: Int -> LogOptions -> ShowS # show :: LogOptions -> String # showList :: [LogOptions] -> ShowS #

data LogOutput Source #

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

Constructors

StdOut
File FilePath

Instances

Instances details

Show LogOutput Source #
Instance details Defined in Pulsar.Internal.Core Methods showsPrec :: Int -> LogOutput -> ShowS # show :: LogOutput -> String # showList :: [LogOutput] -> ShowS #

data Topic Source #

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

Instances details

Show Topic Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> Topic -> ShowS # show :: Topic -> String # showList :: [Topic] -> ShowS #

defaultTopic :: String -> Topic Source #

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

data TopicType Source #

A topic can be either Persistent or NonPersistent.

Constructors

Persistent
NonPersistent

Instances

Instances details

Show TopicType Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> TopicType -> ShowS # show :: TopicType -> String # showList :: [TopicType] -> ShowS #

newtype Tenant Source #

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

Constructors

Tenant String

Instances

Instances details

Show Tenant Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> Tenant -> ShowS # show :: Tenant -> String # showList :: [Tenant] -> ShowS #

newtype NameSpace Source #

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

Constructors

NameSpace String

Instances

Instances details

Show NameSpace Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> NameSpace -> ShowS # show :: NameSpace -> String # showList :: [NameSpace] -> ShowS #

newtype TopicName Source #

A topic name can be any string value.

Constructors

TopicName String

Instances

Instances details

Show TopicName Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> TopicName -> ShowS # show :: TopicName -> String # showList :: [TopicName] -> ShowS #

newtype MsgId Source #

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

Constructors

MsgId MessageIdData

data Message Source #

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

Constructors

Message MsgId ByteString

newtype PulsarMessage Source #

A produced message, containing just a payload as bytestring.

Constructors

PulsarMessage ByteString

Instances

Instances details

Show PulsarMessage Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> PulsarMessage -> ShowS # show :: PulsarMessage -> String # showList :: [PulsarMessage] -> ShowS #
IsString PulsarMessage Source #
Instance details Defined in Pulsar.Types Methods fromString :: String -> PulsarMessage #

newtype SubscriptionName Source #

A subscription name can be any string value.

Constructors

SubscriptionName Text

Instances

Instances details

Show SubscriptionName Source #
Instance details Defined in Pulsar.Types Methods showsPrec :: Int -> SubscriptionName -> ShowS # show :: SubscriptionName -> String # showList :: [SubscriptionName] -> ShowS #
IsString SubscriptionName Source #
Instance details Defined in Pulsar.Types Methods fromString :: String -> SubscriptionName #