Port tutorial 5 to Haskell
This commit is contained in:
parent
474b10d8ff
commit
d9ffc13cef
37
haskell/emitLogTopic.hs
Normal file
37
haskell/emitLogTopic.hs
Normal file
@ -0,0 +1,37 @@
|
||||
{-# OPTIONS -XOverloadedStrings #-}
|
||||
|
||||
import Network.AMQP
|
||||
import qualified Data.ByteString.Lazy.Char8 as BL
|
||||
import qualified Data.Text as DT
|
||||
import System.Environment (getArgs)
|
||||
import Text.Printf
|
||||
|
||||
logsExchange = "topic_logs"
|
||||
|
||||
main :: IO ()
|
||||
main = do
|
||||
args <- getArgs
|
||||
let body = bodyFor args
|
||||
severity = severityFor args
|
||||
conn <- openConnection "127.0.0.1" "/" "guest" "guest"
|
||||
ch <- openChannel conn
|
||||
|
||||
declareExchange ch newExchange {exchangeName = logsExchange,
|
||||
exchangeType = "topic",
|
||||
exchangeDurable = False}
|
||||
publishMsg ch logsExchange (DT.pack severity)
|
||||
(newMsg {msgBody = (BL.pack body),
|
||||
msgDeliveryMode = Just NonPersistent})
|
||||
|
||||
putStrLn $ printf " [x] Sent '%s'" (body)
|
||||
closeConnection conn
|
||||
|
||||
|
||||
bodyFor :: [String] -> String
|
||||
bodyFor [] = "Hello, world!"
|
||||
bodyFor xs = unwords $ tail xs
|
||||
|
||||
|
||||
severityFor :: [String] -> String
|
||||
severityFor [] = "anonymous.info"
|
||||
severityFor xs = head xs
|
39
haskell/receiveLogsTopic.hs
Normal file
39
haskell/receiveLogsTopic.hs
Normal file
@ -0,0 +1,39 @@
|
||||
{-# OPTIONS -XOverloadedStrings #-}
|
||||
|
||||
import Network.AMQP
|
||||
import qualified Data.ByteString.Lazy.Char8 as BL
|
||||
import qualified Data.Text as DT
|
||||
import System.Environment (getArgs)
|
||||
import Text.Printf (printf)
|
||||
import Control.Monad (forM)
|
||||
|
||||
logsExchange = "topic_logs"
|
||||
|
||||
main :: IO ()
|
||||
main = do
|
||||
conn <- openConnection "127.0.0.1" "/" "guest" "guest"
|
||||
ch <- openChannel conn
|
||||
severities <- getArgs
|
||||
|
||||
declareExchange ch newExchange {exchangeName = logsExchange,
|
||||
exchangeType = "topic",
|
||||
exchangeDurable = False}
|
||||
(q, _, _) <- declareQueue ch newQueue {queueName = "",
|
||||
queueAutoDelete = True,
|
||||
queueDurable = False}
|
||||
forM severities (\s -> bindQueue ch q logsExchange (DT.pack s))
|
||||
|
||||
putStrLn " [*] Waiting for messages. to Exit press CTRL+C"
|
||||
consumeMsgs ch q Ack deliveryHandler
|
||||
|
||||
-- waits for keypresses
|
||||
getLine
|
||||
closeConnection conn
|
||||
|
||||
deliveryHandler :: (Message, Envelope) -> IO ()
|
||||
deliveryHandler (msg, metadata) = do
|
||||
putStrLn $ printf " [x] %s:%s" (DT.unpack $ envRoutingKey metadata) body
|
||||
putStrLn $ " [x] Done"
|
||||
ackEnv metadata
|
||||
where
|
||||
body = (BL.unpack $ msgBody msg)
|
Loading…
Reference in New Issue
Block a user