diff --git a/hsnsq.cabal b/hsnsq.cabal --- a/hsnsq.cabal +++ b/hsnsq.cabal @@ -1,64 +1,47 @@ -name: hsnsq -version: 0.1.1.0 -synopsis: Haskell NSQ client. -description: Currently a primitive NSQ client, it implements the very basics of an NSQ client for communicating with single NSQ servers. -homepage: https://github.com/gamelost/hsnsq -license: Apache-2.0 -license-file: LICENSE -author: Paul Berens -maintainer: berens.paul@gmail.com --- copyright: -category: Network -build-type: Simple -extra-source-files: README.md -cabal-version: >=1.10 +name: hsnsq +version: 0.1.2.0 +cabal-version: >=1.10 +build-type: Simple +license: Apache-2.0 +license-file: LICENSE +copyright: (c) Paul Berens +maintainer: berens.paul@gmail.com +homepage: https://github.com/gamelost/hsnsq +synopsis: Haskell NSQ client. +description: + Currently a primitive NSQ client, it implements the very basics of an NSQ client for communicating with single NSQ servers. For example usage refer to https://github.com/gamelost/nsq-benchmarks/haskell/benchmark.hs +category: Network +author: Paul Berens +extra-source-files: + README.md library - exposed-modules: Network.NSQ - , Network.NSQ.Connection - , Network.NSQ.Identify - , Network.NSQ.Parser - , Network.NSQ.Types - default-language: Haskell2010 - ghc-options: -Wall -fno-warn-missing-signatures -O2 - hs-source-dirs: src - build-depends: base >=4.5 && <4.7 - , bytestring - , attoparsec - , attoparsec-binary - , aeson - , pipes - , pipes-attoparsec - , pipes-network - , network - , text - , mtl - , containers - , stm-chans - , stm - , async - , hslogger - , hostname - -executable nsq-example - main-is: Example.hs - default-language: Haskell2010 - ghc-options: -Wall -fno-warn-missing-signatures -O2 -rtsopts -with-rtsopts=-T -threaded -eventlog - hs-source-dirs: src - build-depends: base >=4.5 && <4.7 - , bytestring - , attoparsec - , attoparsec-binary - , aeson - , pipes - , pipes-attoparsec - , pipes-network - , network - , text - , mtl - , containers - , stm-chans - , stm - , async - , hslogger - , hostname + build-depends: + base >=4.5 && <4.8, + bytestring <0.11, + attoparsec <0.13, + attoparsec-binary <0.3, + aeson <0.9, + pipes <4.2, + pipes-attoparsec <0.6, + pipes-network <0.7, + network <2.6, + text <1.2, + mtl <2.3, + containers <0.6, + stm-chans <3.1, + stm <2.5, + async <2.1, + hslogger <1.3, + hostname <1.1 + exposed-modules: + Network.NSQ + Network.NSQ.Connection + Network.NSQ.Identify + Network.NSQ.Parser + Network.NSQ.Types + exposed: True + buildable: True + default-language: Haskell2010 + hs-source-dirs: src + ghc-options: -Wall -fno-warn-missing-signatures -O2 diff --git a/src/Example.hs b/src/Example.hs deleted file mode 100644 --- a/src/Example.hs +++ /dev/null @@ -1,54 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -import Network.NSQ.Types -import Network.NSQ.Connection - -import Control.Concurrent.STM -import Control.Monad -import Control.Concurrent.Async -import Control.Applicative - --- Logger -import System.IO (stderr, Handle) -import System.Log.Logger (rootLoggerName, setHandlers, updateGlobalLogger, Priority(DEBUG), setLevel, infoM) -import System.Log.Handler.Simple (streamHandler, GenericHandler) -import System.Log.Handler (setFormatter) -import System.Log.Formatter - - -main = do - -- Logger stuff - stream <- withFormatter <$> streamHandler stderr DEBUG - let loggerName = rootLoggerName - - updateGlobalLogger loggerName (setLevel DEBUG) - updateGlobalLogger loggerName (setHandlers [stream]) - - -- Create a channel to pump data into - conf <- defaultConfig "66.175.216.197" - topicQueue <- newTQueueIO - replyQueue <- newTQueueIO - - -- Connect - race_ - (establish conf topicQueue replyQueue) - (consumeMessages topicQueue replyQueue) - - -consumeMessages :: TQueue Message -> TQueue Command -> IO () -consumeMessages q r = forever $ do - msg <- atomically (do - m <- readTQueue q - -- Process data here - - -- TODO: Unsafe, assumes it only get Messages (true as of current implementation, but still unsafe) - writeTQueue r $ Fin $ mId m - return m) - infoM "Client.Consume" (show msg) - - where - mId (Message _ _ mesgId _) = mesgId - - -withFormatter :: GenericHandler Handle -> GenericHandler Handle -withFormatter handler = setFormatter handler formatter - where formatter = simpleLogFormatter "[$time $loggername $prio] $msg" diff --git a/src/Network/NSQ/Connection.hs b/src/Network/NSQ/Connection.hs --- a/src/Network/NSQ/Connection.hs +++ b/src/Network/NSQ/Connection.hs @@ -1,4 +1,11 @@ {-# LANGUAGE OverloadedStrings #-} +{-| +Module : Network.NSQ.Connection +Description : Protocol/parser layer for the NSQ client library. + +This is the low level client connection to the nsqd. It is recommended to use the +higher level library when those come out. +-} module Network.NSQ.Connection ( defaultConfig , establish @@ -32,13 +39,13 @@ import qualified Network.NSQ.Parser as NSQ --- --- Default Config --- -- TODO: standardize on some sort of logger hierchary (nsq server/topic?) -- NSQ.[subsystem].[topic].[connection] - message -- NSQ.[subsystem].[custom] .... -- +-- | 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' defaultConfig :: String -> IO NSQConnection defaultConfig serverHost = do localHost <- T.pack <$> getHostName @@ -47,14 +54,13 @@ return $ NSQConnection serverHost 4150 "NSQ.Connection." localIdent --- --- Establish a session with this server --- --- Detail: --- * Support connecting to a particular nsqd and doing the needful to establish identification and so forth --- * Auto-handle heartbeat and all related stuff --- * A higher layer will handle the message reading/balancing between multiplex nsqd connection for a particular topic/channel +-- | 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. establish :: NSQConnection -> TQueue Message -> TQueue Command -> IO () establish conn topicQueue reply = PNT.withSocketsDo $ -- Establish stocket @@ -63,36 +69,40 @@ -- TODO: maybe consider PNT.fromSocketN so that we can adjust fetch size if needed downstream let send = (log "send" $ logName conn) >-> PNT.toSocket sock let recv = PNT.fromSocket sock 8192 >-> (log "recv" $ logName conn) + + -- Establish NSQ first then go into normal handle mode + establishNSQ conn recv send + race_ (handleNSQ conn recv send topicQueue) (runEffect $ handleReply reply >-> showCommand >-> send) -- Handles user replies ) --- --- NSQ Reply handler --- +-- | Pump a 'Command' into the network from the 'TQueue' handleReply :: (Monad m, MonadIO m) => TQueue Command -> Producer NSQ.Command m () handleReply queue = forever $ do cmd <- liftIO $ atomically $ readTQueue queue yield cmd --- --- The NSQ handler and protocol --- +-- | Handles the parsing, error reporting, and logging of the NSQ +-- connection, then eventually pumping the 'Message' into the 'TQueue' +-- for the consumer to process. handleNSQ :: (Monad m, MonadIO m) => NSQConnection -> Producer BS.ByteString m () -> Consumer BS.ByteString m () -> TQueue Message -> m () handleNSQ sc recv send topicQueue = do + runEffect $ (nsqParserErrorLogging (logName sc) recv) >-> (command (logName sc) topicQueue) >-> showCommand >-> send + return () + +-- | Establish the connection to the nsqd and run the initial handshake +-- upto completion then hand it off to the 'handleNSQ' for handling the +-- regular protocol/message. +establishNSQ :: (Monad m, MonadIO m) => NSQConnection -> Producer BS.ByteString m () -> Consumer BS.ByteString m () -> m () +establishNSQ sc recv send = do -- Initial handshake to kick off the handshake runEffect $ (initialHandshake $ identConf sc) >-> showCommand >-> send -- Rest of the handshake process (parsing and dealing with identification) runEffect $ (nsqParserErrorLogging (logName sc) recv) >-> identReply - -- Setup the topic/channel/rdy - runEffect $ setupTopic >-> showCommand >-> send - - -- Regular nsq streaming - runEffect $ (nsqParserErrorLogging (logName sc) recv) >-> (command (logName sc) topicQueue) >-> showCommand >-> send - return () where @@ -102,12 +112,6 @@ yield $ NSQ.Identify im return () - -- TODO; replace it with actual logic - setupTopic = do - yield $ NSQ.Sub "glc-gamestate" "netheril.elder.lan." False - yield $ NSQ.Rdy 1 - return () - -- Process the ident reply identReply = do identR <- await @@ -117,9 +121,7 @@ return () --- --- Parses incoming nsq messages and emits any errors to a log and keep going --- +-- | Parses incoming nsq messages and emits any errors to a log and keep going nsqParserErrorLogging :: MonadIO m => LogName -> Producer BS.ByteString m () -> Producer NSQ.Message m () nsqParserErrorLogging l producer = do (result, rest) <- lift $ runStateT (PA.parse NSQ.message) producer @@ -132,24 +134,20 @@ Left x -> liftIO $ errorM l (show x) nsqParserErrorLogging l rest --- --- Format outbound NSQ Commands --- +-- | Format outbound NSQ Commands showCommand :: Monad m => Pipe NSQ.Command BS.ByteString m () showCommand = PP.map NSQ.encode --- --- Log anything that passes through this stream to a logfile --- +-- | Log anything that passes through this stream to a logfile log :: MonadIO m => String -> LogName -> Pipe BS.ByteString BS.ByteString m r log w l = forever $ do x <- await liftIO $ debugM l (w ++ ": " ++ show x) -- TODO: need a better way to log raw protocol messages yield x --- --- Do something with the inbound message --- +-- | Generic processor for processing various messages automatically such +-- as 'Heartbeat' and just passing upstream (to 'TQueue') the actual +-- 'Message' that needs processing. command :: (Monad m, MonadIO m) => LogName -> TQueue Message -> Pipe NSQ.Message NSQ.Command m () command l topicQueue = forever $ do msg <- await diff --git a/src/Network/NSQ/Identify.hs b/src/Network/NSQ/Identify.hs --- a/src/Network/NSQ/Identify.hs +++ b/src/Network/NSQ/Identify.hs @@ -1,4 +1,8 @@ {-# LANGUAGE OverloadedStrings #-} +{-| +Module : Network.NSQ.Identify +Description : The metadata component for formatting and parsing the metadata sent to nsqd as part of the feature negotiation done upon connection establish. +-} module Network.NSQ.Identify ( defaultIdentify , defaultUserAgent @@ -21,7 +25,9 @@ import Network.NSQ.Types - +-- | 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. defaultIdentify :: T.Text -> T.Text -> IdentifyMetadata defaultIdentify cid host = IdentifyMetadata { ident = Identification cid host Nothing Nothing Nothing @@ -35,9 +41,13 @@ , customNegotiation = False } +-- | The default user agent to send, for identifying what client library is +-- connecting to the nsqd. defaultUserAgent :: T.Text -defaultUserAgent = "hsnsq/0.1.0.0" +defaultUserAgent = "hsnsq/0.1.2.0" -- TODO: find out how to identify this in the build step +-- | Generate a collection of Aeson pairs to insert into the json +-- that is being sent to the server as part of the metadata negotiation. featureNegotiation :: IdentifyMetadata -> [A.Pair] featureNegotiation im = catMaybes ( @@ -52,21 +62,25 @@ optionalCompression (compression im) ) +-- | Take an optional setting and render an Aeson pair. optionalSettings :: T.Text -> Int -> Maybe OptionalSetting -> Maybe A.Pair optionalSettings _ _ Nothing = Nothing optionalSettings name def (Just Disabled) = Just (name, A.toJSON def) optionalSettings name _ (Just (Custom val)) = Just (name, A.toJSON val) +-- | Render the Aeson pairs for optional compression optionalCompression :: Maybe Compression -> [Maybe A.Pair] optionalCompression Nothing = [] optionalCompression (Just NoCompression) = Just `fmap` [ "snappy" .= False, "deflate" .= False ] optionalCompression (Just Snappy) = Just `fmap` [ "snappy" .= True, "deflate" .= False ] optionalCompression (Just (Deflate l)) = Just `fmap` [ "snappy" .= False, "deflate" .= True, "deflate_level" .= l ] +-- | Take the custom settings out of the custom map and render Aeson pairs customMetadata :: Maybe (Map.Map T.Text T.Text) -> [A.Pair] customMetadata Nothing = [] customMetadata (Just val) = Map.foldrWithKey (\k v xs -> (k .= v):xs) [] val +-- | Tls settings tlsSettings :: Maybe TLS -> [Maybe A.Pair] tlsSettings Nothing = [] tlsSettings (Just NoTLS) = [Just $ "tls_v1" .= False] @@ -92,5 +106,8 @@ customMetadata (custom im) ) +-- | 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. encodeMetadata :: IdentifyMetadata -> BL.ByteString encodeMetadata = A.encode diff --git a/src/Network/NSQ/Parser.hs b/src/Network/NSQ/Parser.hs --- a/src/Network/NSQ/Parser.hs +++ b/src/Network/NSQ/Parser.hs @@ -1,4 +1,8 @@ {-# LANGUAGE OverloadedStrings #-} +{-| +Module : Network.NSQ.Parser +Description : Protocol Parser layer for the NSQ client library. +-} module Network.NSQ.Parser ( message , decode @@ -26,18 +30,34 @@ import Network.NSQ.Identify +-- | Decode the 'ByteString' to an 'Message' decode :: BS.ByteString -> Maybe Message decode str = case parseOnly message str of Left _ -> Nothing Right r -> Just r --- TODO convert "E_*" into Error + +-- | Convert various nsq messages into useful 'Message' types command :: BS.ByteString -> Message -command "_heartbeat_" = Heartbeat -command "OK" = OK -command "CLOSE_WAIT" = CloseWait -command x = CatchAllMessage (FTUnknown 99) x -- TODO: Better FrameType +command "_heartbeat_" = Heartbeat +command "OK" = OK +command "CLOSE_WAIT" = CloseWait +command x = CatchAllMessage (FTUnknown 99) x -- TODO: Better FrameType +-- | Convert various Error strings to 'ErrorType' +errorCasting :: BS.ByteString -> ErrorType +errorCasting "E_INVALID" = Invalid +errorCasting "E_BAD_BODY" = BadBody +errorCasting "E_BAD_TOPIC" = BadTopic +errorCasting "E_BAD_MESSAGE" = BadMessage +errorCasting "E_PUB_FAILED" = PubFailed +errorCasting "E_MPUB_FAILED" = MPubFailed +errorCasting "E_FIN_FAILED" = FinFailed +errorCasting "E_REQ_FAILED" = ReqFailed +errorCasting "E_TOUCH_FAILED" = TouchFailed +errorCasting x = Unknown x + +-- | Frame types into 'FrameType' frameType :: Int32 -> FrameType frameType 0 = FTResponse frameType 1 = FTError @@ -46,14 +66,17 @@ -- TODO: do sanity check such as checking that the size is of a minimal -- size, then parsing the frameType, then the remainder (fail "messg") +-- | Parse the low level message frames into a 'Message' type. message :: Parser Message message = do size <- fromIntegral <$> anyWord32be ft <- frameType <$> fromIntegral <$> anyWord32be frame ft (size - 4) -- Taking in accord the frameType +-- | Parse in the frame (remaining portion) of the message in accordance of +-- the 'Frametype' frame :: FrameType -> Int -> Parser Message -frame FTError size = Error <$> take size +frame FTError size = Error <$> (errorCasting `fmap` take size) frame FTResponse size = command <$> take size frame FTMessage size = Message <$> (fromIntegral <$> anyWord64be) @@ -68,10 +91,11 @@ -- TODO: this won't work for streaming the data... -- Should provide two api, one for in memory (ie where we count up the length of the data manualy -- And a "streaming" version in which we know the actual size before streaming (ie streaming from a file for ex) +-- | Primitive version for encoding the size of the data into the frame +-- content then encoding the remaining. sizedData :: BS.ByteString -> BL.Builder sizedData dat = BL.word32BE (fromIntegral $ BS.length dat) <> BL.byteString dat - -- Body of a foldl to build up a sequence of concat sized data concatSizedData :: (Word32, Word32, BL.Builder) -> BS.ByteString -> (Word32, Word32, BL.Builder) concatSizedData (totalSize, count, xs) dat = ( @@ -80,10 +104,10 @@ xs <> sizedData dat ) --- Reply --- Note: --- * One sub (topic/channel) per nsqd connection max, any more will get an E_INVALID --- * Seems to be able to publish to any topic/channel without limitation +-- | 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. encode :: Command -> BS.ByteString encode Protocol = " V2" encode NOP = "NOP\n" diff --git a/src/Network/NSQ/Types.hs b/src/Network/NSQ/Types.hs --- a/src/Network/NSQ/Types.hs +++ b/src/Network/NSQ/Types.hs @@ -1,3 +1,7 @@ +{-| +Module : Network.NSQ.Types +Description : Not much to see here, just the types for the library. +-} module Network.NSQ.Types ( MsgId , Topic @@ -37,11 +41,8 @@ -- * Can probably later on provide helpers for consuming the queue --- | 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 +-- | Per Connection configuration such as: per nsqd state (rdy, load balance), per topic state (channel) data NSQConnection = NSQConnection { server :: String , port :: PortNumber @@ -87,25 +88,24 @@ deriving Show -- | Types of error that the server can return in response to an 'Command' -data ErrorType = Invalid - | BadBody - | BadTopic - | BadChannel - | BadMessage - | PubFailed - | MPubFailed - | FinFailed - | ReqFailed - | TouchFailed +data ErrorType = 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 BS.ByteString -- ^ New unknown type of error (ANY) deriving Show -- | The message and replies back from the server. data Message = OK -- ^ Everything is allright. | Heartbeat -- ^ Heartbeat, reply with the 'NOP' 'Command'. | CloseWait -- ^ Server has closed the connection. - - -- TODO: BS.ByteString -> ErrorType - | Error BS.ByteString -- ^ The server sent back an error. + | Error ErrorType -- ^ The server sent back an error. | Message Int64 Word16 MsgId BS.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. @@ -113,37 +113,43 @@ deriving Show --- Feature and Identification +-- | Optional settings, if 'Disabled' then this setting will be put in the +-- json as disabled specifically vs "not being listed". data OptionalSetting = Disabled | Custom Word64 deriving Show +-- | TLS version supported data TLS = NoTLS | TLSV1 deriving Show +-- | For 'Deflate' its the compression level from 0-9 data Compression = NoCompression | Snappy | Deflate Word8 deriving Show +-- | The client identification data Identification = Identification - { clientId :: T.Text - , hostname :: T.Text - , shortId :: Maybe T.Text -- Deprecated in favor of client_id - , longId :: Maybe T.Text -- Deprecated in favor of hostname - , userAgent :: Maybe T.Text -- Default (client_library_name/version) + { clientId :: T.Text -- ^ An identifier of this consumer, something specific to this consumer + , hostname :: T.Text -- ^ Hostname of the machine the client is running on + , shortId :: Maybe T.Text -- ^ Deprecated in favor of client_id + , longId :: Maybe T.Text -- ^ Deprecated in favor of hostname + , userAgent :: Maybe T.Text -- ^ Default (client_library_name/version) } deriving Show --- feature_negotiation - set automatically if anything is set +-- | 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. data IdentifyMetadata = IdentifyMetadata - { ident :: Identification - , tls :: Maybe TLS - , compression :: Maybe Compression - , heartbeatInterval :: Maybe OptionalSetting -- disabled = -1 - , outputBufferSize :: Maybe OptionalSetting -- disabled = -1 - , outputBufferTimeout :: Maybe OptionalSetting -- disabled = -1 - , sampleRate :: Maybe OptionalSetting -- disabled = 0 - - -- Map of possible json value for future compat - , custom :: Maybe (Map.Map T.Text T.Text) - , customNegotiation :: Bool + { 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.Map T.Text T.Text) -- ^ Map of possible key -> value for future protocol expansion + , customNegotiation :: Bool -- ^ Set if there are any 'custom' values to send } deriving Show