-
-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added GCloud Pub/Sub message queue implementation
refs [AP-566](https://linear.app/ghost/issue/AP-566/implement-a-pubsub-backed-queue-for-fedify) Added and configured a GCloud Pub/Sub message queue implementation for Fedify
- Loading branch information
Showing
7 changed files
with
572 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| #!/bin/bash | ||
|
|
||
| # This script is used to start the Pub/Sub emulator and create the required | ||
| # topic and subscription (defined in the environment variables) upfront | ||
| # | ||
| # See: | ||
| # https://cloud.google.com/pubsub/docs/emulator | ||
| # https://cloud.google.com/pubsub/docs/create-topic#pubsub_create_topic-rest | ||
| # https://cloud.google.com/pubsub/docs/create-push-subscription#pubsub_create_push_subscription-rest | ||
|
|
||
| # Ensure we explicitly set the host to 0.0.0.0:8085 so that the emulator will | ||
| # listen on all ip addresses and not just IPv6 which is the default | ||
| HOST=0.0.0.0:8085 | ||
|
|
||
| # Start the emulator | ||
| gcloud beta emulators pubsub start --host-port=${HOST} --project=${PROJECT_ID} & | ||
|
|
||
| # Wait for the emulator to be ready | ||
| until curl -f http://${HOST}; do | ||
| echo "Waiting for Pub/Sub emulator to start..." | ||
|
|
||
| sleep 1 | ||
| done | ||
|
|
||
| # Create the topic via REST API | ||
| if curl -s -o /dev/null -w "%{http_code}" -X PUT http://${HOST}/v1/projects/${PROJECT_ID}/topics/${TOPIC_NAME} | grep -q "200"; then | ||
| echo "Topic created: ${TOPIC_NAME}" | ||
| else | ||
| echo "Failed to create topic: ${TOPIC_NAME}" | ||
| exit 1 | ||
| fi | ||
|
|
||
| # Create the subscription via REST API | ||
| if curl -s -o /dev/null -w "%{http_code}" -X PUT http://${HOST}/v1/projects/${PROJECT_ID}/subscriptions/${SUBSCRIPTION_NAME} \ | ||
| -H "Content-Type: application/json" \ | ||
| -d '{ | ||
| "topic": "projects/'${PROJECT_ID}'/topics/'${TOPIC_NAME}'" | ||
| }' | grep -q "200"; then | ||
| echo "Subscription created: ${SUBSCRIPTION_NAME}" | ||
| else | ||
| echo "Failed to create subscription: ${SUBSCRIPTION_NAME}" | ||
| exit 1 | ||
| fi | ||
|
|
||
| # Keep the container running | ||
| tail -f /dev/null |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| import type { Message, PubSub } from '@google-cloud/pubsub'; | ||
|
|
||
| import type { | ||
| MessageQueue, | ||
| MessageQueueEnqueueOptions, | ||
| MessageQueueListenOptions, | ||
| } from '@fedify/fedify'; | ||
| import type { Logger } from '@logtape/logtape'; | ||
|
|
||
| export class GCloudPubSubMessageQueue implements MessageQueue { | ||
| private pubSubClient: PubSub; | ||
| private topicIdentifier: string; | ||
| private subscriptionIdentifier: string; | ||
| private logger: Logger; | ||
|
|
||
| constructor( | ||
| pubSubClient: PubSub, | ||
| topicIdentifier: string, | ||
| subscriptionIdentifier: string, | ||
| logger: Logger, | ||
| ) { | ||
| this.pubSubClient = pubSubClient; | ||
| this.topicIdentifier = topicIdentifier; | ||
| this.subscriptionIdentifier = subscriptionIdentifier; | ||
| this.logger = logger; | ||
| } | ||
|
|
||
| async enqueue( | ||
| message: any, | ||
| options?: MessageQueueEnqueueOptions, | ||
| ): Promise<void> { | ||
| const delay = options?.delay?.total('millisecond') ?? 0; | ||
|
|
||
| this.logger.info( | ||
| `Enqueuing message [FedifyID: ${message.id}] with delay: ${delay}ms`, | ||
| ); | ||
|
|
||
| if (delay > 0) { | ||
| await new Promise((resolve) => setTimeout(resolve, delay)); | ||
| } | ||
|
|
||
| try { | ||
| const messageId = await this.pubSubClient | ||
| .topic(this.topicIdentifier) | ||
| .publishMessage({ | ||
| json: message, | ||
| attributes: { | ||
| fedifyId: message.id, | ||
| }, | ||
| }); | ||
|
|
||
| this.logger.info( | ||
| `Message [FedifyID: ${message.id}] was enqueued [PubSubID: ${messageId}]`, | ||
| ); | ||
| } catch (error) { | ||
| this.logger.error( | ||
| `Failed to enqueue message [FedifyID: ${message.id}]: ${error}`, | ||
| ); | ||
| } | ||
| } | ||
|
|
||
| async listen( | ||
| handler: (message: any) => Promise<void> | void, | ||
| options: MessageQueueListenOptions = {}, | ||
| ): Promise<void> { | ||
| const subscription = this.pubSubClient.subscription( | ||
| this.subscriptionIdentifier, | ||
| ); | ||
|
|
||
| subscription.on('message', async (message: Message) => { | ||
| const fedifyId = message.attributes.fedifyId ?? 'unknown'; | ||
|
|
||
| this.logger.info( | ||
| `Handling message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]`, | ||
| ); | ||
|
|
||
| try { | ||
| const json = JSON.parse(message.data.toString()); | ||
|
|
||
| await handler(json); | ||
|
|
||
| message.ack(); | ||
|
|
||
| this.logger.info( | ||
| `Acknowledged message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]`, | ||
| ); | ||
| } catch (error) { | ||
| message.nack(); | ||
|
|
||
| this.logger.error( | ||
| `Failed to handle message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]: ${error}`, | ||
| ); | ||
| } | ||
| }); | ||
|
|
||
| return await new Promise((resolve) => { | ||
| options.signal?.addEventListener('abort', () => { | ||
| subscription | ||
| .removeAllListeners() | ||
| .close() | ||
| .then(() => resolve()); | ||
| }); | ||
| }); | ||
| } | ||
| } |
Oops, something went wrong.