hsnsq 0.1.1.0 → 0.1.2.0
raw patch · 6 files changed
+178/−204 lines, 6 filesdep ~aesondep ~asyncdep ~attoparsecPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependency ranges changed: aeson, async, attoparsec, attoparsec-binary, base, bytestring, containers, hostname, hslogger, mtl, network, pipes, pipes-attoparsec, pipes-network, stm, stm-chans, text
API changes (from Hackage documentation)
+ Network.NSQ: Unknown :: ByteString -> ErrorType
+ Network.NSQ.Types: Unknown :: ByteString -> ErrorType
- Network.NSQ: Error :: ByteString -> Message
+ Network.NSQ: Error :: ErrorType -> Message
- Network.NSQ.Types: Error :: ByteString -> Message
+ Network.NSQ.Types: Error :: ErrorType -> Message
Files
- hsnsq.cabal +45/−62
- src/Example.hs +0/−54
- src/Network/NSQ/Connection.hs +38/−40
- src/Network/NSQ/Identify.hs +19/−2
- src/Network/NSQ/Parser.hs +35/−11
- src/Network/NSQ/Types.hs +41/−35
hsnsq.cabal view
@@ -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
− src/Example.hs
@@ -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"
src/Network/NSQ/Connection.hs view
@@ -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
src/Network/NSQ/Identify.hs view
@@ -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
src/Network/NSQ/Parser.hs view
@@ -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"
src/Network/NSQ/Types.hs view
@@ -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