zeromq3-haskell-0.3.1: Bindings to ZeroMQ 3.x

Portabilitynon-portable
Stabilityexperimental
MaintainerToralf Wittner <tw@dtex.org>
Safe HaskellNone

System.ZMQ3.Monadic

Contents

Description

This modules exposes a monadic interface of ZMQ3. Actions run inside a ZMQ monad and Sockets are guaranteed not to leak outside their corresponding runZMQ scope. Running ZMQ computations asynchronously is directly supported through async.

Synopsis

Type Definitions

data ZMQ z a

The ZMQ monad is modeled after ST and encapsulates a Context. It uses the uninstantiated type variable z to distinguish different invoctions of runZMQ and to prevent unintented use of Sockets outside their scope. Cf. the paper of John Launchbury and Simon Peyton Jones Lazy Functional State Threads.

Instances

data Socket z t

The ZMQ socket, parameterised by SocketType and belonging to a particular ZMQ thread.

data Flag

Flags to apply on send operations (cf. man zmq_send)

Constructors

SendMore

ZMQ_SNDMORE

Instances

data Switch

Configuration switch

Constructors

Default

Use default setting

On

Activate setting

Off

De-activate setting

Instances

type Timeout = Int64

data Event

Socket events.

Constructors

In

ZMQ_POLLIN (incoming messages)

Out

ZMQ_POLLOUT (outgoing messages, i.e. at least 1 byte can be written)

Err

ZMQ_POLLERR

Instances

data EventType

Event types to monitor.

Constructors

ConnectedEvent 
ConnectDelayedEvent 
ConnectRetriedEvent 
ListeningEvent 
BindFailedEvent 
AcceptedEvent 
AcceptFailedEvent 
ClosedEvent 
CloseFailedEvent 
DisconnectedEvent 
AllEvents 

Instances

data EventMsg

Event Message to receive when monitoring socket events.

Constructors

Connected !ByteString !Fd 
ConnectDelayed !ByteString !Fd 
ConnectRetried !ByteString !Int 
Listening !ByteString !Fd 
BindFailed !ByteString !Fd 
Accepted !ByteString !Fd 
AcceptFailed !ByteString !Int 
Closed !ByteString !Fd 
CloseFailed !ByteString !Int 
Disconnected !ByteString !Int 

Instances

data Poll m where

Type representing a descriptor, poll is waiting for (either a 0MQ socket or a file descriptor) plus the type of event to wait for.

Constructors

Sock :: Socket s -> [Event] -> Maybe ([Event] -> m ()) -> Poll m 
File :: Fd -> [Event] -> Maybe ([Event] -> m ()) -> Poll m 

Type Classes

class SocketType a

Socket types.

Instances

class Sender a

Sockets which can send.

Instances

class Receiver a

Sockets which can receive.

Instances

class Subscriber a

Sockets which can subscribe.

Instances

Socket Types

data Pair

Socket to communicate with a single peer. Allows for only a single connect or a single bind. There's no message routing or message filtering involved. Compatible peer sockets: Pair.

Constructors

Pair 

Instances

data Pub

Socket to distribute data. receive function is not implemented for this socket type. Messages are distributed in fanout fashion to all the peers. Compatible peer sockets: Sub.

Constructors

Pub 

Instances

data Sub

Socket to subscribe for data. send function is not implemented for this socket type. Initially, socket is subscribed for no messages. Use subscribe to specify which messages to subscribe for. Compatible peer sockets: Pub.

Constructors

Sub 

Instances

data XPub

Same as Pub except that you can receive subscriptions from the peers in form of incoming messages. Subscription message is a byte 1 (for subscriptions) or byte 0 (for unsubscriptions) followed by the subscription body. Compatible peer sockets: Sub, XSub.

Constructors

XPub 

Instances

data XSub

Same as Sub except that you subscribe by sending subscription messages to the socket. Subscription message is a byte 1 (for subscriptions) or byte 0 (for unsubscriptions) followed by the subscription body. Compatible peer sockets: Pub, XPub.

Constructors

XSub 

Instances

data Req

Socket to send requests and receive replies. Requests are load-balanced among all the peers. This socket type allows only an alternated sequence of send's and recv's. Compatible peer sockets: Rep, Router.

Constructors

Req 

Instances

data Rep

Socket to receive requests and send replies. This socket type allows only an alternated sequence of receive's and send's. Each send is routed to the peer that issued the last received request. Compatible peer sockets: Req, Dealer.

Constructors

Rep 

Instances

data Dealer

Each message sent is round-robined among all connected peers, and each message received is fair-queued from all connected peers. Compatible peer sockets: Router, Req, Rep.

Constructors

Dealer 

Instances

data Router

When receiving messages a Router socket shall prepend a message part containing the identity of the originating peer to the message before passing it to the application. Messages received are fair-queued from among all connected peers. When sending messages a Router socket shall remove the first part of the message and use it to determine the identity of the peer the message shall be routed to. If the peer does not exist anymore the message shall be silently discarded. Compatible peer sockets: Dealer, Req, Rep.

Constructors

Router 

Instances

data Pull

A socket of type Pull is used by a pipeline node to receive messages from upstream pipeline nodes. Messages are fair-queued from among all connected upstream nodes. The zmq_send() function is not implemented for this socket type.

Constructors

Pull 

Instances

data Push

A socket of type Push is used by a pipeline node to send messages to downstream pipeline nodes. Messages are load-balanced to all connected downstream nodes. The zmq_recv() function is not implemented for this socket type.

When a Push socket enters an exceptional state due to having reached the high water mark for all downstream nodes, or if there are no downstream nodes at all, then any zmq_send(3) operations on the socket shall block until the exceptional state ends or at least one downstream node becomes available for sending; messages are not discarded.

Constructors

Push 

Instances

General Operations

version :: ZMQ z (Int, Int, Int)

runZMQ :: MonadIO m => (forall z. ZMQ z a) -> m a

Return the value computed by the given ZMQ monad. Rank-2 polymorphism is used to prevent leaking of z. An invocation of runZMQ will internally create a Context and all actions are executed relative to this context. On finish the context will be disposed, but see async.

async :: ZMQ z a -> ZMQ z ()

Run the given ZMQ computation asynchronously, i.e. this function runs the computation in a new thread using forkIO. N.B. reference counting is used to prolong the lifetime of the Context encapsulated in ZMQ as necessary, e.g.: 

 runZMQ $ do
     s <- socket Pair
     async $ do
         liftIO (threadDelay 10000000)
         identity s >>= liftIO . print

Here, runZMQ will finish before the code section in async, but due to reference counting, the Context will only be disposed after async finishes as well.

socket :: SocketType t => t -> ZMQ z (Socket z t)

ZMQ Options (Read)

ioThreads :: ZMQ z Word

maxSockets :: ZMQ z Word

ZMQ Options (Write)

setIoThreads :: Word -> ZMQ z ()

setMaxSockets :: Word -> ZMQ z ()

Socket operations

close :: Socket z t -> ZMQ z ()

bind :: Socket z t -> String -> ZMQ z ()

unbind :: Socket z t -> String -> ZMQ z ()

connect :: Socket z t -> String -> ZMQ z ()

send :: Sender t => Socket z t -> [Flag] -> ByteString -> ZMQ z ()

send' :: Sender t => Socket z t -> [Flag] -> ByteString -> ZMQ z ()

sendMulti :: Sender t => Socket z t -> [ByteString] -> ZMQ z ()

receive :: Receiver t => Socket z t -> ZMQ z ByteString

receiveMulti :: Receiver t => Socket z t -> ZMQ z [ByteString]

subscribe :: Subscriber t => Socket z t -> ByteString -> ZMQ z ()

unsubscribe :: Subscriber t => Socket z t -> ByteString -> ZMQ z ()

proxy :: Socket z a -> Socket z b -> Maybe (Socket z c) -> ZMQ z ()

monitor :: [EventType] -> Socket z t -> ZMQ z (Bool -> IO (Maybe EventMsg))

poll :: MonadIO m => Timeout -> [Poll m] -> m [[Event]]

Polls for events on the given Poll descriptors. Returns the same list of Poll descriptors with an updated PollEvent field (cf. zmq_poll). Sockets which have seen no activity have None in their PollEvent field.

Socket Options (Read)

affinity :: Socket z t -> ZMQ z Word64

backlog :: Socket z t -> ZMQ z Int

delayAttachOnConnect :: Socket z t -> ZMQ z Bool

events :: Socket z t -> ZMQ z [Event]

fileDescriptor :: Socket z t -> ZMQ z Fd

identity :: Socket z t -> ZMQ z ByteString

ipv4Only :: Socket z t -> ZMQ z Bool

lastEndpoint :: Socket z t -> ZMQ z String

linger :: Socket z t -> ZMQ z Int

maxMessageSize :: Socket z t -> ZMQ z Int64

mcastHops :: Socket z t -> ZMQ z Int

moreToReceive :: Socket z t -> ZMQ z Bool

rate :: Socket z t -> ZMQ z Int

receiveBuffer :: Socket z t -> ZMQ z Int

receiveHighWM :: Socket z t -> ZMQ z Int

receiveTimeout :: Socket z t -> ZMQ z Int

reconnectInterval :: Socket z t -> ZMQ z Int

reconnectIntervalMax :: Socket z t -> ZMQ z Int

recoveryInterval :: Socket z t -> ZMQ z Int

sendBuffer :: Socket z t -> ZMQ z Int

sendHighWM :: Socket z t -> ZMQ z Int

sendTimeout :: Socket z t -> ZMQ z Int

tcpKeepAlive :: Socket z t -> ZMQ z Switch

tcpKeepAliveCount :: Socket z t -> ZMQ z Int

tcpKeepAliveIdle :: Socket z t -> ZMQ z Int

tcpKeepAliveInterval :: Socket z t -> ZMQ z Int

Socket Options (Write)

setAffinity :: Word64 -> Socket z t -> ZMQ z ()

setBacklog :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setDelayAttachOnConnect :: Bool -> Socket z t -> ZMQ z ()

setIdentity :: Restricted N1 N254 ByteString -> Socket z t -> ZMQ z ()

setIpv4Only :: Bool -> Socket z t -> ZMQ z ()

setLinger :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()

setMaxMessageSize :: Integral i => Restricted Nneg1 Int64 i -> Socket z t -> ZMQ z ()

setMcastHops :: Integral i => Restricted N1 Int32 i -> Socket z t -> ZMQ z ()

setRate :: Integral i => Restricted N1 Int32 i -> Socket z t -> ZMQ z ()

setReceiveBuffer :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setReceiveHighWM :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setReceiveTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()

setReconnectInterval :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setReconnectIntervalMax :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setRecoveryInterval :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setRouterMandatory :: Bool -> Socket z Router -> ZMQ z ()

setSendBuffer :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setSendHighWM :: Integral i => Restricted N0 Int32 i -> Socket z t -> ZMQ z ()

setSendTimeout :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()

setTcpAcceptFilter :: Maybe ByteString -> Socket z t -> ZMQ z ()

setTcpKeepAlive :: Switch -> Socket z t -> ZMQ z ()

setTcpKeepAliveCount :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()

setTcpKeepAliveIdle :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()

setTcpKeepAliveInterval :: Integral i => Restricted Nneg1 Int32 i -> Socket z t -> ZMQ z ()

setXPubVerbose :: Bool -> Socket z XPub -> ZMQ z ()

Error Handling

data ZMQError

ZMQError encapsulates information about errors, which occur when using the native 0MQ API, such as error number and message.

Instances

errno :: ZMQError -> Int

Error number value.

source :: ZMQError -> String

Source where this error originates from.

message :: ZMQError -> String

Actual error message.

Re-exports

liftIO :: MonadIO m => forall a. IO a -> m a

Lift a computation from the IO monad.

restrict :: Restriction l u v => v -> Restricted l u v

Create a restricted value. If the given value does not satisfy the restrictions, a modified variant is used instead, e.g. if an integer is larger than the upper bound, the upper bound value is used.

toRestricted :: Restriction l u v => v -> Maybe (Restricted l u v)

Create a restricted value. Returns Nothing if the given value does not satisfy all restrictions.

Low-level Functions

waitRead :: Socket z t -> ZMQ z ()

waitWrite :: Socket z t -> ZMQ z ()