diff --git a/Web/WindowsAzure/ServiceBus.hs b/Web/WindowsAzure/ServiceBus.hs --- a/Web/WindowsAzure/ServiceBus.hs +++ b/Web/WindowsAzure/ServiceBus.hs @@ -7,6 +7,11 @@ BrokerProperties (..), -- * Initialization simpleSBInfo, - sbContext ) where + sbContext, + -- * Locked Messages + unlockMessage, + renewLock, + deleteMessage + ) where import Web.WindowsAzure.ServiceBus.SBTypes diff --git a/Web/WindowsAzure/ServiceBus/Queue.hs b/Web/WindowsAzure/ServiceBus/Queue.hs --- a/Web/WindowsAzure/ServiceBus/Queue.hs +++ b/Web/WindowsAzure/ServiceBus/Queue.hs @@ -43,10 +43,7 @@ enQueueBodySrc, -- * Reading data from Queue deQueue, - peekLockQueue, - unlockMessage, - deleteMessage, - renewLock + peekLockQueue )where import qualified Data.ByteString.Char8 as C @@ -66,7 +63,6 @@ import qualified Data.CaseInsensitive as CI import Network.Connection (TLSSettings (..)) -import Data.Aeson import Network(withSocketsDo) @@ -149,7 +145,7 @@ -- other receivers (on the same subscription) during the duration of the lock. -- -- Refer for semantics of the underlying REST API. -peekLockQueue :: String -> Int -> SBContext -> IO (QLockedMsgInfo,L.ByteString) +peekLockQueue :: String -> Int -> SBContext -> IO (LockedMsgInfo,L.ByteString) peekLockQueue qName timeout (SBContext baseUrl manager aContext) = do token <- acsToken manager aContext reqInit <- parseUrl (baseUrl ++ "/" ++ qName ++ "/messages/head?timeout=" ++ (show timeout)) @@ -158,58 +154,3 @@ }) manager return $ (getQLI res,responseBody res) -getQLI :: Response L.ByteString -> QLockedMsgInfo -getQLI res = QLockedMsgInfo loc bp - where - loc = case lookup hLocation (responseHeaders res) of - Nothing -> error "Expected Location Header in the response!" - Just x -> C.unpack x - bp = case lookup (CI.mk . C.pack $ "BrokerProperties") (responseHeaders res) of - Nothing -> emptyBP - Just bs -> case decode $ L.fromChunks [bs] of - Nothing -> emptyBP - Just b -> b - --- | Unlock a messages that has been locked earlier. --- --- Given a queueName and the broker properties of the message that has been locked before, --- 'unlockMessage' removes the lock so that the message can be consumed by other consumers. --- --- see 'peekLockQueue' and for details on the underlying API. -unlockMessage :: String -> QLockedMsgInfo -> SBContext -> IO () -unlockMessage queueName (QLockedMsgInfo url brokerProps) (SBContext baseUrl manager acsContext) = do - token <- acsToken manager acsContext - reqInit <- parseUrl url - res <-withSocketsDo $ httpLbs (reqInit { method = methodPut, - requestHeaders = [token] - }) manager - return () - --- | Delete a message that has been locked earlier. --- --- Given a queueName and the locked message info, 'deleteMessage' deletes the message from the queue --- --- see 'peekLockQueue' and for details on the underlying API. -deleteMessage :: String -> QLockedMsgInfo -> SBContext -> IO () -deleteMessage queueName (QLockedMsgInfo url brokerProps) (SBContext baseUrl manager acsContext) = do - token <- acsToken manager acsContext - reqInit <- parseUrl url - res <-withSocketsDo $ httpLbs (reqInit { method = methodDelete, - requestHeaders = [token] - }) manager - return () - --- | Renews lock on a locked message --- --- Given a queueName and the locked-message info, 'renewLock' renews the lock --- 'deleteMessage' deletes the message from the queue --- --- see 'peekLockQueue' and for details on the underlying API. -renewLock :: String -> QLockedMsgInfo -> SBContext -> IO () -renewLock queueName (QLockedMsgInfo url brokerProps) (SBContext baseUrl manager acsContext) = do - token <- acsToken manager acsContext - reqInit <- parseUrl url - res <-withSocketsDo $ httpLbs (reqInit { method = methodPost, - requestHeaders = [token] - }) manager - return () diff --git a/Web/WindowsAzure/ServiceBus/SBTypes.hs b/Web/WindowsAzure/ServiceBus/SBTypes.hs --- a/Web/WindowsAzure/ServiceBus/SBTypes.hs +++ b/Web/WindowsAzure/ServiceBus/SBTypes.hs @@ -22,13 +22,15 @@ import qualified Data.ByteString.Lazy as L import Network.HTTP.Conduit hiding (requestBodySource) import Network.HTTP.Client.Conduit hiding (httpLbs) -import Network.HTTP.Types.Method (methodDelete, methodPost) +import Network.HTTP.Types.Method (methodDelete, methodPost,methodPut) import Network.HTTP.Types.Header +import Network(withSocketsDo) import Network.Connection (TLSSettings (..)) import Data.Aeson import Data.Monoid import Control.Applicative +import qualified Data.CaseInsensitive as CI -- | 'SBInfo' is encapsulation of Connection Information needed to connect to a Service Bus Namespace. @@ -97,3 +99,65 @@ ] emptyBP = BrokerProperties 0 0 "" "" "" "" 0 "" 0 + +-- | 'LockedMsgInfo' provides Information of the locked message. +-- +data LockedMsgInfo = LockedMsgInfo String BrokerProperties + deriving (Show) + + +-- | Unlock a messages that has been locked earlier. +-- +-- see 'Web.WindowsAzure.ServiceBus.Topic.peekLockTopic' and 'Web.WindowsAzure.ServiceBus.Queue.peekLockQueue'. +-- +-- Also consult and for details on the underlying REST API. +unlockMessage :: LockedMsgInfo -> SBContext -> IO () +unlockMessage (LockedMsgInfo url brokerProps) (SBContext baseUrl manager acsContext) = do + token <- acsToken manager acsContext + reqInit <- parseUrl url + res <-withSocketsDo $ httpLbs (reqInit { method = methodPut, + requestHeaders = [token] + }) manager + return () + +-- | Delete a message that has been locked earlier. +-- +-- see 'Web.WindowsAzure.ServiceBus.Topic.peekLockTopic' and 'Web.WindowsAzure.ServiceBus.Queue.peekLockQueue'. +-- +-- Also consult and for details on the underlying REST API. +deleteMessage :: LockedMsgInfo -> SBContext -> IO () +deleteMessage (LockedMsgInfo url brokerProps) (SBContext baseUrl manager acsContext) = do + token <- acsToken manager acsContext + reqInit <- parseUrl url + res <-withSocketsDo $ httpLbs (reqInit { method = methodDelete, + requestHeaders = [token] + }) manager + return () + +-- | Renews lock on a locked message +-- +-- see 'Web.WindowsAzure.ServiceBus.Topic.peekLockTopic' and 'Web.WindowsAzure.ServiceBus.Queue.peekLockQueue'. +-- +-- Also consult and for details on the underlying REST API. +renewLock :: LockedMsgInfo -> SBContext -> IO () +renewLock (LockedMsgInfo url brokerProps) (SBContext baseUrl manager acsContext) = do + token <- acsToken manager acsContext + reqInit <- parseUrl url + res <-withSocketsDo $ httpLbs (reqInit { method = methodPost, + requestHeaders = [token] + }) manager + return () + + + +getQLI :: Response L.ByteString -> LockedMsgInfo +getQLI res = LockedMsgInfo loc bp + where + loc = case lookup hLocation (responseHeaders res) of + Nothing -> error "Expected Location Header in the response!" + Just x -> C.unpack x + bp = case lookup (CI.mk . C.pack $ "BrokerProperties") (responseHeaders res) of + Nothing -> emptyBP + Just bs -> case decode $ L.fromChunks [bs] of + Nothing -> emptyBP + Just b -> b diff --git a/Web/WindowsAzure/ServiceBus/Topic.hs b/Web/WindowsAzure/ServiceBus/Topic.hs new file mode 100644 --- /dev/null +++ b/Web/WindowsAzure/ServiceBus/Topic.hs @@ -0,0 +1,149 @@ +-- | +-- Module : Web.WindowsAzure.ServiceBus.Topic +-- Description : API for reading from and writing to ServiceBus Topic +-- Copyright : (c) Hemanth Kapila, 2014 +-- License : BSD3 +-- Maintainer : saihemanth@gmail.com +-- Stability : Experimental +-- +-- Provides API to pull from and push to ServiceBus topic +-- Please refer to for information on the API provided by +-- Microsoft Service bus. +-- +-- Simple example for how to use this library is as below +-- +-- @ +-- import Web.WindowsAzure.ServiceBus.Topic +-- import Web.WindowsAzure.ServiceBus +-- import qualified Data.ByteString.Char8 as C +-- +-- topicName = "topicName" +-- subscriptionName = "subscriptionName" +-- sbNamespace = "namespace" +-- sbIssuerKey = C.pack "1287361251262as=" +-- sbIssuerName = C.pack "owner" +-- +-- sbinfo = SBInfo sbNamespace sbIssuerName sbIssuerKey +-- message = C.pack "Hello from Haskell" +-- +-- main = do +-- sbContext <- sbContext sbinfo +-- sendTopicBS topicName message sbContext +-- res <- destructiveRead topicName subscriptionName 30 sbContext +-- print res +-- @ +-- +-- see examples (available as a part of distribution) for a more detailed example. + +module Web.WindowsAzure.ServiceBus.Topic( + -- * Pushing data to Topic + sendTopicBS, + sendTopicLBS, + sendTopicBodySrc, + -- * Reading data from Topic + destructiveRead, + peekLockTopic + )where + +import qualified Data.ByteString.Char8 as C +import qualified Data.ByteString.Lazy as L +import Data.Conduit +import Data.Int + +import Web.WindowsAzure.ACS +import Web.WindowsAzure.ServiceBus.SBTypes +import qualified Data.ByteString.Char8 as C +import qualified Data.ByteString.Lazy as L +import Network.HTTP.Conduit hiding (requestBodySource) +import Network.HTTP.Client.Conduit hiding (httpLbs) +import Network.HTTP.Types.Method (methodDelete, methodPost,methodPut) +import Network.HTTP.Types.Header +import Network.HTTP.Types.Method + +import qualified Data.CaseInsensitive as CI +import Network.Connection (TLSSettings (..)) +import Data.Aeson +import Network(withSocketsDo) + +-- | Internal low-level method for performing HTTP calls. +-- +-- Not exposed to the user +sendTopicRequest :: String -> RequestBody -> SBContext -> IO () +sendTopicRequest topicName body (SBContext baseUrl manager aContext) = do + token <- acsToken manager aContext + reqInit <- parseUrl (baseUrl ++ "/" ++ topicName ++ "/messages") + withSocketsDo $ httpLbs (reqInit { method = methodPost, + requestHeaders = [token], + requestBody = body + }) manager + return () + +-- | Internal low-level method for creating the HTTP calls. For internal use. +-- +-- Not exposed to the user +destructiveReadRequest :: String -> String -> Int -> SBContext -> IO L.ByteString +destructiveReadRequest topic subsc timeout (SBContext baseUrl manager aContext) = do + token <- acsToken manager aContext + reqInit <- parseUrl (baseUrl ++ "/" ++ topic ++ "/Subscriptions/" ++ subsc ++ "/messages/head?timeout=" ++ (show timeout)) + res <-withSocketsDo $ httpLbs ( reqInit { method = methodDelete, + requestHeaders = [token] + }) manager + return $ responseBody res + +-- | publish a message containing 'C.ByteString' to a topic. +-- +-- The following publishes a strict bytestring \bs\ to topic \t\ +-- +-- @ +-- sendTopicBS t bs ctx +-- @ +sendTopicBS :: String -> C.ByteString -> SBContext -> IO () +sendTopicBS topicName content context = + sendTopicRequest topicName (RequestBodyBS content) context + + +-- | publish a message containing 'L.ByteString' to a topic +-- +-- The following publishes a lazy bytestring ,\lbs\, to topic \t\, +-- +-- @ +-- sendTopicLBS t lbs ctx +-- @ +-- +sendTopicLBS :: String -> L.ByteString -> SBContext -> IO () +sendTopicLBS topicName content context = + sendTopicRequest topicName (RequestBodyLBS content) context + + +-- | publish from a 'Source' (refer to 'requestBodySource') +sendTopicBodySrc :: String -> Int64 -> Source IO C.ByteString -> SBContext -> IO () +sendTopicBodySrc topicName len bodysrc context = sendTopicRequest topicName (requestBodySource len bodysrc) context + + +-- | Reads and deletes the message from a topic at a given subscription. +-- +-- +-- In order to destructively read the latest message from the subscription /subsc/ on topic /t/ (with a time out of n seconds), +-- +-- +-- @ +-- destructiveRead t subsc n context +-- @ +-- Note that the timeout can be at the most 55 seconds. +destructiveRead :: String -> String -> Int -> SBContext -> IO (L.ByteString) +destructiveRead topic subsc timeout context = destructiveReadRequest topic subsc (timeout `mod` 55) context + +-- | Peek Lock Message from a Topic. Non-Destructive Read. +-- +-- Atomically retrieves the message from a topic (on a given subscription) without deleting it. The message is locked for a duration so that it is not visible to other +-- receivers. +-- +-- Refer for semantics of the underlying REST API. +peekLockTopic :: String -> String -> Int -> SBContext -> IO (LockedMsgInfo,L.ByteString) +peekLockTopic topic subscr timeout (SBContext baseUrl manager aContext) = do + token <- acsToken manager aContext + reqInit <- parseUrl (baseUrl ++ "/" ++ topic ++ "/Subscriptions/" ++ subscr ++ "/messages/head?timeout=" ++ (show timeout)) + res <-withSocketsDo $ httpLbs (reqInit { method = methodPost, + requestHeaders = [token] + }) manager + return $ (getQLI res,responseBody res) diff --git a/azure-servicebus.cabal b/azure-servicebus.cabal --- a/azure-servicebus.cabal +++ b/azure-servicebus.cabal @@ -2,12 +2,32 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: azure-servicebus -version: 0.1.1.0 +version: 0.1.5.0 synopsis: Haskell wrapper over Microsoft Azure ServiceBus REST API -description: +Description: + /Overview/ + . This library provides haskell wrappers over Microsoft Azure ServiceBus REST API. - The current version provides for queues. + The current version provides Runtime API for . + + . + The following functionality is covered: + . + * Send bytestring (lazy as well as strict) to a Queue. + . + * Send bytestring (lazy as well as strict) to a topic. + . + * Destructively read data from a queue or topic into a lazy bytestring + . + * acquire peek-locks on messages on Queues and Subscriptions + . + * renew lock + . + * delete a locked message + . + * unlock a locked message + homepage: https://github.com/kapilash/hs-azure license: BSD3 license-file: LICENSE @@ -21,6 +41,7 @@ library exposed-modules: Web.WindowsAzure.ServiceBus.Queue, + Web.WindowsAzure.ServiceBus.Topic, Web.WindowsAzure.ServiceBus other-modules: Web.WindowsAzure.ServiceBus.SBTypes -- other-extensions: diff --git a/examples/Example.hs b/examples/Example.hs deleted file mode 100644 --- a/examples/Example.hs +++ /dev/null @@ -1,37 +0,0 @@ -module Main where - -import Web.WindowsAzure.ServiceBus.Queue -import Web.WindowsAzure.ServiceBus -import qualified Data.ByteString.Char8 as C -import Network(withSocketsDo) - -queueName = "QUEUE-NAME" -sbNamespace = "sb-namespace" -sbIssuerKey = C.pack "somekey23823872=" -sbIssuerName = C.pack "owner" - -sbinfo = SBInfo sbNamespace sbIssuerName sbIssuerKey -message1 = C.pack "Hello from Haskell" -message2 = C.pack "Haskell Rocks" - -main = do - sbContext <- sbContext sbinfo - enQueueBS queueName message1 sbContext - -- Enqueued message 1 to queue - res <- deQueue queueName 30 sbContext - -- Dequeued it - print res - -- Enqueue it again - enQueueBS queueName message1 sbContext - -- EnQueue Another Message - enQueueBS queueName message2 sbContext - -- Peek Lock on the first message - (lockInfo,msgFromQueue) <- peekLockQueue queueName 50 sbContext - -- Now a deQueue would result in us getting the second message (while the first is still in the queue) - res3 <- deQueue queueName 10 sbContext - print res3 - -- renew lock - renewLock queueName lockInfo sbContext - -- delete the message - deleteMessage queueName lockInfo sbContext - diff --git a/examples/QueueExample.hs b/examples/QueueExample.hs new file mode 100644 --- /dev/null +++ b/examples/QueueExample.hs @@ -0,0 +1,37 @@ +module Main where + +import Web.WindowsAzure.ServiceBus.Queue +import Web.WindowsAzure.ServiceBus +import qualified Data.ByteString.Char8 as C +import Network(withSocketsDo) + +queueName = "QUEUE-NAME" +sbNamespace = "sb-namespace" +sbIssuerKey = C.pack "somekey23823872=" +sbIssuerName = C.pack "owner" + +sbinfo = SBInfo sbNamespace sbIssuerName sbIssuerKey +message1 = C.pack "Hello from Haskell" +message2 = C.pack "Haskell Rocks" + +main = do + sbContext <- sbContext sbinfo + enQueueBS queueName message1 sbContext + -- Enqueued message 1 to queue + res <- deQueue queueName 30 sbContext + -- Dequeued it + print res + -- Enqueue it again + enQueueBS queueName message1 sbContext + -- EnQueue Another Message + enQueueBS queueName message2 sbContext + -- Peek Lock on the first message + (lockInfo,msgFromQueue) <- peekLockQueue queueName 50 sbContext + -- Now a deQueue would result in us getting the second message (while the first is still in the queue) + res3 <- deQueue queueName 10 sbContext + print res3 + -- renew lock + renewLock lockInfo sbContext + -- delete the message + deleteMessage lockInfo sbContext + diff --git a/examples/TopicExample.hs b/examples/TopicExample.hs new file mode 100644 --- /dev/null +++ b/examples/TopicExample.hs @@ -0,0 +1,38 @@ +module Main where + +import Web.WindowsAzure.ServiceBus.Topic +import Web.WindowsAzure.ServiceBus +import qualified Data.ByteString.Char8 as C +import Network(withSocketsDo) + +topic = "TopicName" +subs = "subsName" +sbNamespace = "SBNAMESPACE" +sbIssuerKey = C.pack "someSubscriptionKey=" +sbIssuerName = C.pack "owner" + +sbinfo = SBInfo sbNamespace sbIssuerName sbIssuerKey +message1 = C.pack "Hello from Haskell" +message2 = C.pack "Haskell Rocks" + +main = do + sbContext <- sbContext sbinfo + sendTopicBS topic message1 sbContext + -- Enqueued message 1 to queue + res <- destructiveRead topic subs 30 sbContext + -- Dequeued it + print res + -- Enqueue it again + sendTopicBS topic message1 sbContext + -- EnQueue Another Message + sendTopicBS topic message2 sbContext + -- Peek Lock on the first message + (lockInfo,msgFromQueue) <- peekLockTopic topic subs 50 sbContext + -- Now a destructiveRead would result in us getting the second message (while the first is still in the topic) + res3 <- destructiveRead topic subs 10 sbContext + print res3 + -- renew lock + renewLock lockInfo sbContext + -- delete the message + deleteMessage lockInfo sbContext +