azure-servicebus 0.1.1.0 → 0.1.5.0

raw patch · 8 files changed

+321/−103 lines, 8 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Web/WindowsAzure/ServiceBus.hs +6/−1
• Web/WindowsAzure/ServiceBus/Queue.hs +2/−61
• Web/WindowsAzure/ServiceBus/SBTypes.hs +65/−1
• Web/WindowsAzure/ServiceBus/Topic.hs +149/−0
• azure-servicebus.cabal +24/−3
• examples/Example.hs +0/−37
• examples/QueueExample.hs +37/−0
• examples/TopicExample.hs +38/−0

@@ -7,6 +7,11 @@     BrokerProperties (..),     -- * Initialization     simpleSBInfo,-    sbContext ) where+    sbContext,+    -- * Locked Messages+    unlockMessage,+    renewLock,+    deleteMessage+    ) where  import Web.WindowsAzure.ServiceBus.SBTypes

@@ -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 <http://msdn.microsoft.com/en-us/library/hh780735.aspx ServiceBus documentation> 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 <http://msdn.microsoft.com/en-us/library/hh780723.aspx ServiceBus documentation> 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 <http://msdn.microsoft.com/en-us/library/hh780767.aspx ServiceBus documentation> 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 <http://msdn.microsoft.com/en-us/library/jj839741.aspx ServiceBus documentation> 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 ()

@@ -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 <http://msdn.microsoft.com/en-us/library/hh780723.aspx Unlock message from Queue> and <http://msdn.microsoft.com/en-us/library/hh780737.aspx Unlock message from Subscription> 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 <http://msdn.microsoft.com/en-us/library/hh780767.aspx Delete Message from a Queue> and <http://msdn.microsoft.com/en-us/library/hh780768.aspx Delete Message from Subscription> 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 <http://msdn.microsoft.com/en-us/library/jj839741.aspx Renew Lock for message from Queue> and <http://msdn.microsoft.com/en-us/library/jj839746.aspx Renew-Lock for a message from subscription> 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

@@ -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 <http://msdn.microsoft.com/en-us/library/hh780752.aspx Service Bus Rest API> 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 <http://msdn.microsoft.com/en-us/library/hh780722.aspx ServiceBus documentation> 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)

@@ -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 <http://msdn.microsoft.com/en-us/library/hh780762.aspx Runtime API> for queues.+        The current version provides Runtime API for  <http://msdn.microsoft.com/en-us/library/hh780762.aspx queues and topics>.+        +        .+        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:    

@@ -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-

@@ -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+

@@ -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+