System.CMQ

CMQ, a UDP-based inherently asynchronous message queue to orchestrate messages, events and processes in the cloud. It trades guarantees, consistency mechanisms, (shared) state and transactions for robustness, scalability and performance. CMQ fares especially well in modern Layer 2 switches in data center networks, as well as in the presence of errors. A Message is pushed to the queue together with a queue identifier (Cmq) and a KEY that specifies the recipient. Messages can be pushed in logarithmic time and the next message can be retrieved in constant time.

This implementation is based on

• J. Fritsch, C. Walker, CMQ - A lightweight, asynchronous high-performance messaging queue for the cloud (2012).

The queue identifier (Token)

data Cmq a

General purpose finite queue.

IPv4 address

Use read "192.0.2.1" :: IPv4, for example. Also, "192.0.2.1" can be used as literal with OverloadedStrings.

data IPv4

The abstract data type to express an IPv4 address. To create this, use toIPv4. Or use read "192.0.2.1" :: IPv4, for example. Also, "192.0.2.1" can be used as literal with OverloadedStrings.

>>> read "192.0.2.1" :: IPv4
192.0.2.1

Instances

Eq IPv4

Ord IPv4

Read IPv4

Show IPv4

IsString IPv4

Destination identifier (KEY)

type KEY

Arguments

 = (IPv4, Integer)

The KEY identifies the message destination in the format IPv4 address, Integer. The integer is reserved for future use e.g. as unique process identifier

Construction

newRq

Arguments

:: Serialize a

=> Socket

Socket does not need to be in connected state.

-> Int

Maximum Queue length in bytes.

-> Rational

Maximum Queue age in ms.

-> IO (Cmq a)

Token returned to identify the Queue.

Builds and returns a new instance of Cmq.

  (token) <- newRq soc 512 200

Insertion (Push Message)

cwPush :: Serialize a => Socket -> KEY -> a -> Cmq a -> IO ()

O(log n). Push a message to the queue.

  cwPush soc ("192.168.35.69", 0) ("ping" :: String) token 

Query (Pop a Message)

cwPop :: Cmq a -> IO (Maybe a)

O(1). A message is popped of CMQ. The next value is read from the queue. Use for example

   msg <- cwPop token :: IO (Maybe String)

or with ScopedTypeVariables

   (msg :: Maybe String) <- cwPop token