packages feed

google-cloud-pubsub (empty) → 1.1.0.0

raw patch · 9 files changed

+842/−0 lines, 9 filesdep +aesondep +basedep +base64-bytestringsetup-changed

Dependencies added: aeson, base, base64-bytestring, bytestring, containers, google-cloud-common, google-cloud-pubsub, http-conduit, http-types, text

Files

+ CHANGELOG.md view
@@ -0,0 +1,27 @@+# Changelog++All notable changes to this project will be documented in this file.++The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).++## [0.1.0.0] - 2025-01-27++## 1.1.0.0 - 2025-09-11++- Synchronized version bump for v1.1.0.0 release.++### Added+- Initial release of google-cloud-pubsub+- Topic operations:+  - Create topics+  - Delete topics+  - Publish messages to topics+- Subscription operations:+  - Create subscriptions+  - Delete subscriptions+  - Acknowledge messages+  - Modify acknowledgment deadlines+  - Pull messages from subscriptions+- Comprehensive data types for Pub/Sub entities+- Integration with google-cloud-common for authentication
+ LICENSE view
@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2025 Tushar Adhatrao++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ README.md view
@@ -0,0 +1,115 @@+# google-cloud-pubsub++A thin wrapper around Google Cloud Pub/Sub for Haskell.++## Features++### Topic Operations+- Create topics+- Delete topics  +- Publish messages to topics++### Subscription Operations+- Create subscriptions+- Delete subscriptions+- Acknowledge messages+- Modify acknowledgment deadlines+- Pull messages from subscriptions++## Installation++- Cabal: add to your `.cabal`+  - `build-depends: google-cloud-pubsub == 1.1.0.0`+- Stack: add to your `package.yaml`+  - `dependencies: - google-cloud-pubsub == 1.1.0.0`++This package depends on `google-cloud-common` for authentication and HTTP helpers.++## Usage++### Topic Operations++```haskell+import Google.Cloud.PubSub.Topic++-- Create a topic+result <- createTopic "my-project" "my-topic"+case result of+  Left err -> putStrLn $ "Error: " ++ err+  Right topic -> print topic++-- Publish a message+let message = Message "Hello World" Nothing Nothing+publishResult <- publishMessage "my-project" "my-topic" [message]+case publishResult of+  Left err -> putStrLn $ "Error: " ++ err+  Right response -> print $ messageIds response++-- Delete a topic+deleteResult <- deleteTopic "my-project" "my-topic"+case deleteResult of+  Left err -> putStrLn $ "Error: " ++ err+  Right _ -> putStrLn "Topic deleted successfully"+```++### Subscription Operations++```haskell+import Google.Cloud.PubSub.Subscription++-- Create a subscription+let config = SubscriptionConfig +  { topic = "projects/my-project/topics/my-topic"+  , pushConfig = Nothing+  , ackDeadlineSeconds = 600+  , retainAckedMessages = False+  , messageRetentionDuration = Nothing+  , labels = Nothing+  , enableMessageOrdering = False+  , expirationPolicy = Nothing+  , filter = Nothing+  , deadLetterPolicy = Nothing+  , retryPolicy = Nothing+  , enableExactlyOnceDelivery = Nothing+  }++createResult <- createSubscription "my-project" "my-subscription" config+case createResult of+  Left err -> putStrLn $ "Error: " ++ err+  Right subscription -> print subscription++-- Pull messages+pullResult <- pullMessages "my-project" "my-subscription" 10+case pullResult of+  Left err -> putStrLn $ "Error: " ++ err+  Right response -> do+    mapM_ print $ receivedMessages response+    -- Acknowledge messages+    let ackIds = map ackId $ receivedMessages response+    ackResult <- acknowledgeMessages "my-project" "my-subscription" ackIds+    case ackResult of+      Left err -> putStrLn $ "Ack error: " ++ err+      Right _ -> putStrLn "Messages acknowledged"++-- Delete a subscription+deleteResult <- deleteSubscription "my-project" "my-subscription"+case deleteResult of+  Left err -> putStrLn $ "Error: " ++ err+  Right _ -> putStrLn "Subscription deleted successfully"+```++## Authentication++Authentication is handled by `google-cloud-common` and follows this order:++1. If `GOOGLE_APPLICATION_CREDENTIALS` is set, a Service Account JSON is used to+   create a signed JWT which is exchanged for an access token.+2. Otherwise, the Compute metadata server is used (suitable for GCE/GKE).++## Dependencies++- `google-cloud-common` - Common functionality for Google Cloud libraries++## License++MIT © Contributors
+ Setup.hs view
@@ -0,0 +1,3 @@+import Distribution.Simple++main = defaultMain
+ google-cloud-pubsub.cabal view
@@ -0,0 +1,69 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.38.1.+--+-- see: https://github.com/sol/hpack++name:           google-cloud-pubsub+version:        1.1.0.0+synopsis:       GCP Pub/Sub Client for Haskell+description:    GCP Pub/Sub topic and subscription client for Haskell.+category:       Web+homepage:       https://github.com/tusharad/google-cloud-haskell#readme+bug-reports:    https://github.com/tusharad/google-cloud-haskell/issues+author:         tushar+maintainer:     tusharadhatrao@gmail.com+copyright:      2025 tushar+license:        MIT+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    CHANGELOG.md++source-repository head+  type: git+  location: https://github.com/tusharad/google-cloud-haskell++library+  exposed-modules:+      Google.Cloud.PubSub.Common+      Google.Cloud.PubSub.Subscription+      Google.Cloud.PubSub.Topic+  other-modules:+      Paths_google_cloud_pubsub+  hs-source-dirs:+      src+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints+  build-depends:+      aeson <3+    , base >=4.7 && <5+    , base64-bytestring+    , bytestring >=0.9.1.4+    , containers+    , google-cloud-common >=1.1.0.0 && <1.2.0.0+    , http-conduit+    , http-types+    , text >=1.2 && <3+  default-language: Haskell2010++test-suite google-cloud-pubsub-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Paths_google_cloud_pubsub+  hs-source-dirs:+      test+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      aeson <3+    , base >=4.7 && <5+    , base64-bytestring+    , bytestring >=0.9.1.4+    , containers+    , google-cloud-common >=1.1.0.0 && <1.2.0.0+    , google-cloud-pubsub+    , http-conduit+    , http-types+    , text >=1.2 && <3+  default-language: Haskell2010
+ src/Google/Cloud/PubSub/Common.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : Google.Cloud.PubSub.Common+License     : MIT+Maintainer  : maintainers@google-cloud-haskell+Stability   : experimental++Common helpers and constants for Google Cloud Pub/Sub client modules.+-}+module Google.Cloud.PubSub.Common (googlePubSubUrl, PubsubMessage (..)) where++import Data.Aeson+import Data.Map (Map)+import GHC.Generics (Generic)++googlePubSubUrl :: String+googlePubSubUrl = "https://pubsub.googleapis.com/v1"++-- | A message to be published.+data PubsubMessage = PubsubMessage+  { data_ :: Maybe String+  -- ^ The message data (base64 encoded).+  , attributes :: Maybe (Map String String)+  -- ^ Optional attributes for the message.+  , messageId :: Maybe String+  -- ^ The message ID (set by the server).+  , publishTime :: Maybe String+  -- ^ The publish time (set by the server).+  , orderingKey :: Maybe String+  -- ^ The ordering key for the message.+  }+  deriving (Show, Eq, Generic)++instance FromJSON PubsubMessage where+  parseJSON =+    genericParseJSON+      defaultOptions+        { fieldLabelModifier = \case+            "data_" -> "data"+            other -> other+        }++instance ToJSON PubsubMessage where+  toJSON =+    genericToJSON+      defaultOptions+        { fieldLabelModifier = \case+            "data_" -> "data"+            other -> other+        }
+ src/Google/Cloud/PubSub/Subscription.hs view
@@ -0,0 +1,299 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE OverloadedStrings #-}++{- | This module provides functions for interacting with Google Cloud Pub/Sub subscriptions.+It includes functionality for creating, deleting subscriptions, acknowledging messages,+modifying acknowledgment deadlines, and pulling messages.+-}+module Google.Cloud.PubSub.Subscription+  ( Subscription (..)+  , PullRequest (..)+  , PullResponse (..)+  , ReceivedMessage (..)+  , AcknowledgeRequest (..)+  , ModifyAckDeadlineRequest (..)+  , SubscriptionConfig (..)+  , createSubscription+  , deleteSubscription+  , acknowledgeMessages+  , modifyAckDeadline+  , pullMessages+  , googlePubSubUrl+  ) where++import Data.Aeson+import Data.Map (Map)+import GHC.Generics (Generic)+import Google.Cloud.Common.Core+import Google.Cloud.PubSub.Common++-- | Represents a Pub/Sub subscription.+data Subscription = Subscription+  { name :: String+  -- ^ The name of the subscription in the format projects/{project}/subscriptions/{subscription}.+  , topic :: String+  -- ^ The name of the topic this subscription belongs to.+  , pushConfig :: Maybe PushConfig+  -- ^ Push configuration for the subscription.+  , ackDeadlineSeconds :: Int+  -- ^ The acknowledgment deadline in seconds.+  , retainAckedMessages :: Maybe Bool+  -- ^ Whether to retain acknowledged messages.+  , messageRetentionDuration :: Maybe String+  -- ^ Message retention duration.+  , labels :: Maybe (Map String String)+  -- ^ Labels for the subscription.+  , enableMessageOrdering :: Maybe Bool+  -- ^ Whether message ordering is enabled.+  , expirationPolicy :: Maybe ExpirationPolicy+  -- ^ Expiration policy for the subscription.+  , filter :: Maybe String+  -- ^ Filter for the subscription.+  , deadLetterPolicy :: Maybe DeadLetterPolicy+  -- ^ Dead letter policy for the subscription.+  , retryPolicy :: Maybe RetryPolicy+  -- ^ Retry policy for the subscription.+  , detached :: Maybe Bool+  -- ^ Whether the subscription is detached.+  , enableExactlyOnceDelivery :: Maybe Bool+  -- ^ Whether exactly-once delivery is enabled.+  , state :: Maybe String+  -- ^ The state of the subscription.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Push configuration for a subscription.+data PushConfig = PushConfig+  { pushEndpoint :: Maybe String+  -- ^ The push endpoint URL.+  , attributes :: Maybe (Map String String)+  -- ^ Attributes for the push endpoint.+  , oidcToken :: Maybe OidcToken+  -- ^ OIDC token for authentication.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | OIDC token for push authentication.+data OidcToken = OidcToken+  { serviceAccountEmail :: String+  -- ^ Service account email for the token.+  , audience :: Maybe String+  -- ^ Audience for the token.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Expiration policy for a subscription.+newtype ExpirationPolicy = ExpirationPolicy+  { ttl :: String+  -- ^ Time-to-live for the subscription.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Dead letter policy for a subscription.+data DeadLetterPolicy = DeadLetterPolicy+  { deadLetterTopic :: String+  -- ^ The dead letter topic.+  , maxDeliveryAttempts :: Maybe Int+  -- ^ Maximum delivery attempts.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Retry policy for a subscription.+data RetryPolicy = RetryPolicy+  { minimumBackoff :: Maybe String+  -- ^ Minimum backoff duration.+  , maximumBackoff :: Maybe String+  -- ^ Maximum backoff duration.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Configuration for creating a subscription.+data SubscriptionConfig = SubscriptionConfig+  { topic :: String+  -- ^ The topic name for the subscription.+  , pushConfig :: Maybe PushConfig+  -- ^ Push configuration.+  , ackDeadlineSeconds :: Int+  -- ^ Acknowledgment deadline in seconds.+  , retainAckedMessages :: Maybe Bool+  -- ^ Whether to retain acknowledged messages.+  , messageRetentionDuration :: Maybe String+  -- ^ Message retention duration.+  , labels :: Maybe (Map String String)+  -- ^ Labels for the subscription.+  , enableMessageOrdering :: Maybe Bool+  -- ^ Whether message ordering is enabled.+  , expirationPolicy :: Maybe ExpirationPolicy+  -- ^ Expiration policy.+  , filter :: Maybe String+  -- ^ Filter for the subscription.+  , deadLetterPolicy :: Maybe DeadLetterPolicy+  -- ^ Dead letter policy.+  , retryPolicy :: Maybe RetryPolicy+  -- ^ Retry policy.+  , enableExactlyOnceDelivery :: Maybe Bool+  -- ^ Whether exactly-once delivery is enabled.+  }+  deriving (Show, Generic, ToJSON)++-- | Request to pull messages from a subscription.+data PullRequest = PullRequest+  { maxMessages :: Int+  -- ^ Maximum number of messages to pull.+  , returnImmediately :: Bool+  -- ^ Whether to return immediately if no messages are available.+  }+  deriving (Show, Generic, ToJSON)++-- | Response from pulling messages.+newtype PullResponse = PullResponse+  { receivedMessages :: [ReceivedMessage]+  -- ^ The received messages.+  }+  deriving (Show, Generic, FromJSON)++-- | A received message from a subscription.+data ReceivedMessage = ReceivedMessage+  { ackId :: String+  -- ^ The acknowledgment ID for the message.+  , message :: PubsubMessage+  -- ^ The actual message.+  , deliveryAttempt :: Maybe Int+  -- ^ The delivery attempt number.+  }+  deriving (Show, Generic, FromJSON)++-- | Request to acknowledge messages.+newtype AcknowledgeRequest = AcknowledgeRequest+  { ackIds :: [String]+  -- ^ The acknowledgment IDs to acknowledge.+  }+  deriving (Show, Generic, ToJSON)++-- | Request to modify acknowledgment deadline.+data ModifyAckDeadlineRequest = ModifyAckDeadlineRequest+  { ackIds :: [String]+  -- ^ The acknowledgment IDs to modify.+  , ackDeadlineSeconds :: Int+  -- ^ The new acknowledgment deadline in seconds.+  }+  deriving (Show, Generic, ToJSON)++{- | Creates a new subscription in Google Cloud Pub/Sub.++@createSubscription projectId subscriptionName config@ creates a new subscription+with the specified name and configuration. It returns either an error message+or the created 'Subscription'.++Example:++>>> let config = SubscriptionConfig "projects/my-project/topics/my-topic" Nothing 600 False Nothing Nothing False Nothing Nothing Nothing Nothing Nothing+>>> createSubscription "my-project" "my-subscription" config+Right (Subscription {name = "projects/my-project/subscriptions/my-subscription", ...})+-}+createSubscription :: String -> String -> SubscriptionConfig -> IO (Either String Subscription)+createSubscription pId subscriptionName config =+  doRequestJSON+    RequestOptions+      { reqMethod = PUT+      , reqUrl = googlePubSubUrl+      , mbReqPath = Just ("/projects/" <> pId <> "/subscriptions/" <> subscriptionName)+      , mbReqBody = Just (encode config)+      , mbReqHeaders = Just [("Content-Type", "application/json")]+      , mbQueryParams = Nothing+      }++{- | Deletes a subscription from Google Cloud Pub/Sub.++@deleteSubscription projectId subscriptionName@ deletes the specified subscription.+It returns either an error message or a unit value indicating success.++Example:++>>> deleteSubscription "my-project" "my-subscription"+Right ()+-}+deleteSubscription :: String -> String -> IO (Either String ())+deleteSubscription pId subscriptionName =+  doRequestJSON+    RequestOptions+      { reqMethod = DELETE+      , reqUrl = googlePubSubUrl+      , mbReqPath = Just ("/projects/" <> pId <> "/subscriptions/" <> subscriptionName)+      , mbReqBody = Nothing+      , mbReqHeaders = Nothing+      , mbQueryParams = Nothing+      }++{- | Acknowledges messages in a subscription.++@acknowledgeMessages projectId subscriptionName ackIds@ acknowledges the messages+with the given acknowledgment IDs. It returns either an error message or a unit+value indicating success.++Example:++>>> acknowledgeMessages "my-project" "my-subscription" ["ack-id-1", "ack-id-2"]+Right ()+-}+acknowledgeMessages :: String -> String -> [String] -> IO (Either String ())+acknowledgeMessages pId subscriptionName aIds =+  doRequestJSON+    RequestOptions+      { reqMethod = POST+      , reqUrl = googlePubSubUrl+      , mbReqPath =+          Just ("/projects/" <> pId <> "/subscriptions/" <> subscriptionName <> ":acknowledge")+      , mbReqBody = Just (encode $ AcknowledgeRequest aIds)+      , mbReqHeaders = Just [("Content-Type", "application/json")]+      , mbQueryParams = Nothing+      }++{- | Modifies the acknowledgment deadline for messages.++@modifyAckDeadline projectId subscriptionName ackIds newDeadline@ modifies+the acknowledgment deadline for the given messages. It returns either an error+message or a unit value indicating success.++Example:++>>> modifyAckDeadline "my-project" "my-subscription" ["ack-id-1"] 600+Right ()+-}+modifyAckDeadline :: String -> String -> [String] -> Int -> IO (Either String ())+modifyAckDeadline pId subscriptionName aIds newDeadline =+  doRequestJSON+    RequestOptions+      { reqMethod = POST+      , reqUrl = googlePubSubUrl+      , mbReqPath =+          Just ("/projects/" <> pId <> "/subscriptions/" <> subscriptionName <> ":modifyAckDeadline")+      , mbReqBody = Just (encode $ ModifyAckDeadlineRequest aIds newDeadline)+      , mbReqHeaders = Just [("Content-Type", "application/json")]+      , mbQueryParams = Nothing+      }++{- | Pulls messages from a subscription.++@pullMessages projectId subscriptionName maxMessages@ pulls up to maxMessages+from the specified subscription. It returns either an error message or a+'PullResponse' containing the received messages.++Example:++>>> pullMessages "my-project" "my-subscription" 10+Right (PullResponse {receivedMessages = [...]})+-}+pullMessages :: String -> String -> Int -> IO (Either String PullResponse)+pullMessages pId subscriptionName maxMsgs =+  doRequestJSON+    RequestOptions+      { reqMethod = POST+      , reqUrl = googlePubSubUrl+      , mbReqPath = Just ("/projects/" <> pId <> "/subscriptions/" <> subscriptionName <> ":pull")+      , mbReqBody = Just (encode $ PullRequest maxMsgs False)+      , mbReqHeaders = Just [("Content-Type", "application/json")]+      , mbQueryParams = Nothing+      }
+ src/Google/Cloud/PubSub/Topic.hs view
@@ -0,0 +1,175 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++{- | This module provides functions for interacting with Google Cloud Pub/Sub topics.+It includes functionality for creating, deleting topics, and publishing messages.+-}+module Google.Cloud.PubSub.Topic+  ( Topic (..)+  , PublishRequest (..)+  , PublishResponse (..)+  , Message (..)+  , PubsubMessage (..)+  , createTopic+  , deleteTopic+  , publishMessage+  , googlePubSubUrl+  ) where++import Data.Aeson+import qualified Data.ByteString.Base64 as Base64+import qualified Data.ByteString.Char8 as BS+import Data.Map (Map)+import GHC.Generics (Generic)+import Google.Cloud.Common.Core+import Google.Cloud.PubSub.Common++-- | Represents a Pub/Sub topic.+data Topic = Topic+  { name :: String+  -- ^ The name of the topic in the format projects/{project}/topics/{topic}.+  , kmsKeyName :: Maybe String+  -- ^ The KMS key name for encryption, if any.+  , messageStoragePolicy :: Maybe MessageStoragePolicy+  -- ^ The message storage policy.+  , labels :: Maybe (Map String String)+  -- ^ Labels for the topic.+  , messageRetentionDuration :: Maybe String+  -- ^ Message retention duration.+  , state :: Maybe String+  -- ^ The state of the topic.+  , satisfiesPzs :: Maybe Bool+  -- ^ Whether the topic satisfies PZS requirements.+  , schemaSettings :: Maybe SchemaSettings+  -- ^ Schema settings for the topic.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Message storage policy for a topic.+newtype MessageStoragePolicy = MessageStoragePolicy+  { allowedPersistenceRegions :: [String]+  -- ^ Allowed regions for message persistence.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Schema settings for a topic.+data SchemaSettings = SchemaSettings+  { schema :: String+  -- ^ The schema name.+  , encoding :: String+  -- ^ The encoding type (JSON or BINARY).+  , firstRevisionId :: Maybe String+  -- ^ First revision ID.+  , lastRevisionId :: Maybe String+  -- ^ Last revision ID.+  }+  deriving (Show, Generic, FromJSON, ToJSON)++-- | Request to publish messages to a topic.+newtype PublishRequest = PublishRequest+  { messages :: [PubsubMessage]+  -- ^ The messages to publish.+  }+  deriving (Show, Generic, ToJSON)++-- | Response from publishing messages.+newtype PublishResponse = PublishResponse+  { messageIds :: [String]+  -- ^ The message IDs of the published messages.+  }+  deriving (Show, Generic, FromJSON)++-- | Simplified message structure for easier usage.+data Message = Message+  { content :: String+  -- ^ The message content.+  , attributes :: Maybe (Map String String)+  -- ^ Optional attributes for the message.+  , orderingKey :: Maybe String+  -- ^ The ordering key for the message.+  }+  deriving (Show, Generic)++{- | Creates a new topic in Google Cloud Pub/Sub.++@createTopic projectId topicName@ creates a new topic with the specified name+in the given project. It returns either an error message or the created 'Topic'.++Example:++>>> createTopic "my-project" "my-topic"+Right (Topic {name = "projects/my-project/topics/my-topic", ...})+-}+createTopic :: String -> String -> IO (Either String Topic)+createTopic projectId topicName =+  doRequestJSON+    RequestOptions+      { reqMethod = PUT+      , reqUrl = googlePubSubUrl+      , mbReqPath = Just ("/projects/" <> projectId <> "/topics/" <> topicName)+      , mbReqBody = Just (encode $ object [])+      , mbReqHeaders = Just [("Content-Type", "application/json")]+      , mbQueryParams = Nothing+      }++{- | Deletes a topic from Google Cloud Pub/Sub.++@deleteTopic projectId topicName@ deletes the specified topic.+It returns either an error message or a unit value indicating success.++Example:++>>> deleteTopic "my-project" "my-topic"+Right ()+-}+deleteTopic :: String -> String -> IO (Either String ())+deleteTopic projectId topicName =+  doRequestJSON+    RequestOptions+      { reqMethod = DELETE+      , reqUrl = googlePubSubUrl+      , mbReqPath = Just ("/projects/" <> projectId <> "/topics/" <> topicName)+      , mbReqBody = Nothing+      , mbReqHeaders = Nothing+      , mbQueryParams = Nothing+      }++{- | Publishes messages to a topic.++@publishMessage projectId topicName messages@ publishes the given messages+to the specified topic. It returns either an error message or a 'PublishResponse'+containing the message IDs.++Example:++>>> let messages = [Message "Hello World" Nothing Nothing]+>>> publishMessage "my-project" "my-topic" messages+Right (PublishResponse {messageIds = ["1234567890"]})+-}+publishMessage :: String -> String -> [Message] -> IO (Either String PublishResponse)+publishMessage projectId topicName messages = do+  let pubsubMessages = map messageToPubsubMessage messages+      publishRequest = PublishRequest pubsubMessages+  doRequestJSON+    RequestOptions+      { reqMethod = POST+      , reqUrl = googlePubSubUrl+      , mbReqPath = Just ("/projects/" <> projectId <> "/topics/" <> topicName <> ":publish")+      , mbReqBody = Just (encode publishRequest)+      , mbReqHeaders = Just [("Content-Type", "application/json")]+      , mbQueryParams = Nothing+      }++-- | Converts a simple Message to a PubsubMessage for API calls.+messageToPubsubMessage :: Message -> PubsubMessage+messageToPubsubMessage Message {..} =+  PubsubMessage+    { data_ = Just (BS.unpack . Base64.encode $ BS.pack content)+    , attributes = attributes+    , messageId = Nothing+    , publishTime = Nothing+    , orderingKey = orderingKey+    }
+ test/Spec.hs view
@@ -0,0 +1,80 @@+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import Data.Aeson (eitherDecode, encode)+import qualified Data.ByteString.Base64 as Base64+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as BSL+import qualified Data.Map as M+import Test.Tasty (TestTree, defaultMain, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, testCase)++import Google.Cloud.PubSub.Common (PubsubMessage (..), googlePubSubUrl)+import Google.Cloud.PubSub.Subscription+  ( PullRequest (..)+  , PullResponse (..)+  , ReceivedMessage (..)+  )+import Google.Cloud.PubSub.Topic+  ( Message (..)+  , PublishRequest (..)+  , PublishResponse (..)+  )++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests =+  testGroup+    "google-cloud-pubsub"+    [ testCase "googlePubSubUrl is v1 endpoint" $ do+        assertEqual "Endpoint" "https://pubsub.googleapis.com/v1" googlePubSubUrl+    , testCase "PubsubMessage JSON field mapping of data_" $ do+        let msg =+              PubsubMessage+                { data_ = Just "Zm9v"+                , attributes = Nothing+                , messageId = Nothing+                , publishTime = Nothing+                , orderingKey = Nothing+                }+            json = encode msg+        assertBool "JSON must contain \"data\" key" ("\"data\"" `BS.isInfixOf` BSL.toStrict json)+        case eitherDecode json of+          Left err -> assertBool ("Failed to decode: " <> err) False+          Right (decoded :: PubsubMessage) -> assertEqual "Roundtrip" msg decoded+    , testCase "Message gets base64-encoded into PubsubMessage via PublishRequest" $ do+        let m =+              Message {content = "hello", attributes = Just (M.fromList [("k", "v")]), orderingKey = Just "ord"}+            pr =+              PublishRequest+                [ PubsubMessage+                    (Just (BS.unpack (Base64.encode (BS.pack "hello"))))+                    (Just (M.fromList [("k", "v")]))+                    Nothing+                    Nothing+                    (Just "ord")+                ]+            encoded = encode pr+        assertBool "Contains base64 of 'hello'" ("\"aGVsbG8=\"" `BS.isInfixOf` BSL.toStrict encoded)+    , testCase "Decode PublishResponse with messageIds" $ do+        let json = BSL.pack "{\n  \"messageIds\": [\"1\", \"2\"]\n}"+        case eitherDecode json of+          Left err -> assertBool ("Failed to decode: " <> err) False+          Right (PublishResponse mids) -> assertEqual "IDs" ["1", "2"] mids+    , testCase "Encode PullRequest and decode PullResponse with one message" $ do+        let req = PullRequest {maxMessages = 3, returnImmediately = False}+        assertBool "maxMessages present" ("\"maxMessages\"" `BS.isInfixOf` BSL.toStrict (encode req))+        let json =+              BSL.pack+                "{\n  \"receivedMessages\": [\n    {\n      \"ackId\": \"ack-1\",\n      \"message\": {\n        \"data\": \"aGVsbG8=\",\n        \"attributes\": {\n          \"k\": \"v\"\n        }\n      },\n      \"deliveryAttempt\": 1\n    }\n  ]\n}"+        case eitherDecode json of+          Left err -> assertBool ("Failed to decode: " <> err) False+          Right (PullResponse msgs) -> do+            assertEqual "One message" 1 (length msgs)+            let ReceivedMessage {ackId = ack, deliveryAttempt = att} = head msgs+            assertEqual "ackId" "ack-1" ack+            assertEqual "deliveryAttempt" (Just 1) att+    ]