-
-
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.
- Loading branch information
Showing
7 changed files
with
568 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,42 @@ | ||
| #!/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 | ||
|
|
||
| HOST=0.0.0.0:8085 | ||
|
|
||
| # Start the Pub/Sub emulator | ||
| gcloud beta emulators pubsub start --host-port=${HOST} --project=${PROJECT_ID} & | ||
|
|
||
| # Wait for the Pub/Sub emulator to be ready | ||
| until curl -f http://${HOST}; do | ||
| echo "Waiting for Pub/Sub emulator to start..." | ||
|
|
||
| sleep 1 | ||
| done | ||
|
|
||
| # Create the topic | ||
| 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 | ||
| 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.