diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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 diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,3 @@ +import Distribution.Simple + +main = defaultMain diff --git a/google-cloud-pubsub.cabal b/google-cloud-pubsub.cabal new file mode 100644 --- /dev/null +++ b/google-cloud-pubsub.cabal @@ -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 diff --git a/src/Google/Cloud/PubSub/Common.hs b/src/Google/Cloud/PubSub/Common.hs new file mode 100644 --- /dev/null +++ b/src/Google/Cloud/PubSub/Common.hs @@ -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 + } diff --git a/src/Google/Cloud/PubSub/Subscription.hs b/src/Google/Cloud/PubSub/Subscription.hs new file mode 100644 --- /dev/null +++ b/src/Google/Cloud/PubSub/Subscription.hs @@ -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 + } diff --git a/src/Google/Cloud/PubSub/Topic.hs b/src/Google/Cloud/PubSub/Topic.hs new file mode 100644 --- /dev/null +++ b/src/Google/Cloud/PubSub/Topic.hs @@ -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 + } diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -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 + ]