Network.NSQ

This is a haskell client for the NSQ message queue service.

TODO:

• Start implementing a more through client
• Write some form of tests
• More through docs, the initial docs sucks.

Documentation

type MsgId = ByteString

Message Id, it is a 16-byte hexdecimal string encoded as ASCII

type Topic = Text

NSQ Topic, the only allowed character in a topic is \.a-zA-Z0-9_-

type Channel = Text

NSQ Channel, the only allowed character in a channel is \.a-zA-Z0-9_- A channel can be marked as ephemeral by toggling the Bool value to true in Sub

data Message

The message and replies back from the server.

Constructors

OK

Everything is allright.

Heartbeat

Heartbeat, reply with the NOP Command.

CloseWait

Server has closed the connection.

Error ByteString

The server sent back an error.

Message Int64 Word16 MsgId ByteString

A message to be processed. The values are: Nanosecond Timestamp, number of attempts, Message Id, and the content of the message to be processed.

CatchAllMessage FrameType ByteString

Catch-all message for future expansion. This currently includes the reply from Identify if feature negotiation is set.

Instances

Show Message

data Command

NSQ Command

Constructors

Protocol

The protocol version

NOP

No-op, usually used in reply to a Heartbeat request from the server.

Identify IdentifyMetadata

Client Identification + possible features negotiation.

Sub Topic Channel Bool

Subscribe to a specified 'Topic'/'Channel', use True if its an ephemeral channel.

Pub Topic ByteString

Publish a message to the specified Topic.

MPub Topic [ByteString]

Publish multiple messages to a specified Topic.

Rdy Word64

Update RDY state (ready to recieve messages). Number of message you can process at once.

Fin MsgId

Finish a message.

Req MsgId Word64

Re-queue a message (failure to process), Timeout is in milliseconds.

Touch MsgId

Reset the timeout for an in-flight message.

Cls

Cleanly close the connection to the NSQ daemon.

Command ByteString

Catch-all command for future expansion/custom commands.

Instances

Show Command

data FrameType

Frame Type of the incoming data from the NSQ daemon.

Constructors

FTResponse

Response to a Command from the server.

FTError

An error in response to a Command.

FTMessage

Messages.

FTUnknown Int32

For future extension for handling new Frame Types.

Instances

Show FrameType

data ErrorType

Types of error that the server can return in response to an Command

Constructors

Invalid

BadBody

BadTopic

BadChannel

BadMessage

PubFailed

MPubFailed

FinFailed

ReqFailed

TouchFailed

Instances

Show ErrorType

data OptionalSetting

Constructors

Disabled

Custom Word64

Instances

Show OptionalSetting

data TLS

Constructors

NoTLS

TLSV1

Instances

Show TLS

data Compression

Constructors

NoCompression

Snappy

Deflate Word8

Instances

Show Compression

data Identification

Constructors

Identification

Fields

clientId :: Text

hostname :: Text

shortId :: Maybe Text

longId :: Maybe Text

userAgent :: Maybe Text

Instances

Show Identification

data IdentifyMetadata

Constructors

IdentifyMetadata

Fields

ident :: Identification

tls :: Maybe TLS

compression :: Maybe Compression

heartbeatInterval :: Maybe OptionalSetting

outputBufferSize :: Maybe OptionalSetting

outputBufferTimeout :: Maybe OptionalSetting

sampleRate :: Maybe OptionalSetting

custom :: Maybe (Map Text Text)

customNegotiation :: Bool

Instances

Show IdentifyMetadata

ToJSON IdentifyMetadata

data NSQConnection

Per Connection configuration * Per nsqd (connection) state (rdy, load balance, etc) * Per topic state (channel related info and which nsqd connection) * Global? state (do we have any atm? maybe configuration?) TODO: consider using monad-journal logger for pure code tracing

Constructors

NSQConnection

Fields

server :: String

port :: PortNumber

logName :: LogName

identConf :: IdentifyMetadata

message :: Parser Message

decode :: ByteString -> Maybe Message

encode :: Command -> ByteString

defaultIdentify :: Text -> Text -> IdentifyMetadata

defaultUserAgent :: Text

encodeMetadata :: IdentifyMetadata -> ByteString

defaultConfig :: String -> IO NSQConnection

establish :: NSQConnection -> TQueue Message -> TQueue Command -> IO ()