Copyright	(c) Paul Berens, 2014
License	Apache 2.0
Maintainer	berens.paul@gmail.com
Stability	experimental
Portability	unknown
Safe Haskell	None
Language	Haskell2010

Network.NSQ

Description

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.

Synopsis

Documentation

type MsgId = ByteString Source

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

type Topic = Text Source

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

type Channel = Text Source

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 Source

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

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 Source

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 Source

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

Constructors

Invalid	Something went wrong with the command (IDENTIFY, SUB, PUB, MPUB, RDY, FIN, REQ, TOUCH, CLS)
BadBody	Bad Body (IDENTIFY, MPUB)
BadTopic	Bad Topic (most likely used disallowed characters) (SUB, PUB, MPUB)
BadChannel	Bad channel (Like `BadTopic` probably used an disallowed character) (SUB)
BadMessage	Bad Message (PUB, MPUB)
PubFailed	Publishing a message failed (PUB)
MPubFailed	Same as `PubFailed` (MPUB)
FinFailed	Finish failed (Probably already finished or non-existant message-id) (FIN)
ReqFailed	Requeue failed (REQ)
TouchFailed	Touch failed (TOUCH)
Unknown ByteString	New unknown type of error (ANY)

Instances

Show ErrorType

data OptionalSetting Source

Optional settings, if Disabled then this setting will be put in the json as disabled specifically vs "not being listed".

Constructors

Disabled
Custom Word64

Instances

Show OptionalSetting

data TLS Source

TLS version supported

Constructors

NoTLS
TLSV1

Instances

Show TLS

data Compression Source

For Deflate its the compression level from 0-9

Constructors

NoCompression
Snappy
Deflate Word8

Instances

Show Compression

data Identification Source

The client identification

Constructors

Identification
Fields clientId :: Text An identifier of this consumer, something specific to this consumer hostname :: Text Hostname of the machine the client is running on shortId :: Maybe Text Deprecated in favor of client_id longId :: Maybe Text Deprecated in favor of hostname userAgent :: Maybe Text Default (client_library_name/version)

Instances

Show Identification

data IdentifyMetadata Source

Metadata for feature negotiation, if any of the values are set it will be sent to the server otherwise they will be omitted. If the setting is set to Nothing it will not be sent to the server, and if its set to Just Disabled it will be sent to the server as disabled explicitly.

Constructors

IdentifyMetadata

Fields

ident :: Identification: Client identification
tls :: Maybe TLS: TLS
compression :: Maybe Compression: Compression
heartbeatInterval :: Maybe OptionalSetting: The time between each heartbeat (disabled = -1)
outputBufferSize :: Maybe OptionalSetting: The size of the buffer (disabled = -1)
outputBufferTimeout :: Maybe OptionalSetting: The timeout for the buffer (disabled = -1)
sampleRate :: Maybe OptionalSetting: Sampling of the message will be sent to the client (disabled = 0)
custom :: Maybe (Map Text Text): Map of possible key -> value for future protocol expansion
customNegotiation :: Bool: Set if there are any custom values to send

Instances

Show IdentifyMetadata
ToJSON IdentifyMetadata

data NSQConnection Source

Per Connection configuration such as: per nsqd state (rdy, load balance), per topic state (channel)

Constructors

NSQConnection
Fields server :: String port :: PortNumber logName :: LogName identConf :: IdentifyMetadata

message :: Parser Message Source

Parse the low level message frames into a Message type.

decode :: ByteString -> Maybe Message Source

Decode the ByteString to an Message

encode :: Command -> ByteString Source

Encode a Command into raw ByteString to send to the network to the nsqd daemon. There are a few gotchas here; You can only have one Sub (topic/channel) per nsqld connection, any other will yield Invalid. Also you can publish to any number of topic without limitation.

defaultIdentify :: Text -> Text -> IdentifyMetadata Source

Build a default IdentifyMetadata that makes sense which is basically just setting the client Identification and leaving the rest of the settings up to the server to determine.

defaultUserAgent :: Text Source

The default user agent to send, for identifying what client library is connecting to the nsqd.

encodeMetadata :: IdentifyMetadata -> ByteString Source

Encode the metadata from IdentifyMetadata into a ByteString for feeding the Identify Command for sending the metadata to the nsq daemon as part of the feature negotiation.

defaultConfig :: String -> IO NSQConnection Source

Attempt to come up with an intelligent default NSQConnection default by discovering your client's hostname and reusing it for the client id for the IdentifyMetadata

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

Establish a session with the specified nsqd using the provided TQueue so that the actual data processing can be done in a decoupled manner from the hsnsq stack.

This supports connecting to a specific nsqd, it is however recommended in the future when the feature comes out to user a higher layer that handles the load balancing between multiple nsqd.