-
-
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 (#143)
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
17 changed files
with
960 additions
and
16 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
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
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
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,49 @@ | ||
| #!/bin/bash | ||
|
|
||
| # This script is used to start the Pub/Sub emulator and create the required | ||
| # topic and subscription upfront (defined in the environment variables) | ||
| # | ||
| # 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 (push) 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}'", | ||
| "pushConfig": { | ||
| "pushEndpoint": "'${PUSH_ENDPOINT}'" | ||
| } | ||
| }' | 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| import type { Context } from 'hono'; | ||
|
|
||
| import type { HonoContextVariables } from '../../../app'; | ||
| import { EVENT_MQ_MESSAGE_RECEIVED } from '../../../constants'; | ||
| import { MqMessageReceivedEvent } from '../../../events/mq-message-received-event'; | ||
|
|
||
| export async function handleMessageAction( | ||
| ctx: Context<{ Variables: HonoContextVariables }>, | ||
| ): Promise<Response> { | ||
| const logger = ctx.get('logger'); | ||
| const eventBus = ctx.get('eventBus'); | ||
|
|
||
| const json = await ctx.req.json(); | ||
| const pubSubId = json?.message?.message_id ?? 'unknown'; | ||
|
|
||
| // If no listeners are attached, we should not process the message | ||
| if (eventBus.listenerCount(EVENT_MQ_MESSAGE_RECEIVED) === 0) { | ||
| logger.info( | ||
| `No event listeners attached to [${EVENT_MQ_MESSAGE_RECEIVED}], nacking incoming message [PubSub ID: ${pubSubId}]`, | ||
| ); | ||
|
|
||
| return new Response(null, { status: 429 }); | ||
| } | ||
|
|
||
| // Return a promise that will eventually resolve when a message is ack'd or nack'd | ||
| return await new Promise<Response>((resolve) => { | ||
| let data = {}; | ||
|
|
||
| try { | ||
| data = JSON.parse( | ||
| Buffer.from(json.message.data, 'base64').toString(), | ||
| ); | ||
| } catch (error) { | ||
| logger.error( | ||
| `Failed to parse message data [PubSub ID: ${pubSubId}]: ${error}`, | ||
| ); | ||
|
|
||
| return resolve(new Response(null, { status: 500 })); | ||
| } | ||
|
|
||
| const event = new MqMessageReceivedEvent({ | ||
| id: json.message.message_id, | ||
| subscriptionIdentifier: json.subscription, | ||
| data, | ||
| attributes: json.message.attributes, | ||
| onAck: () => resolve(new Response(null, { status: 200 })), | ||
| onNack: () => resolve(new Response(null, { status: 500 })), | ||
| }); | ||
|
|
||
| eventBus.emit(EVENT_MQ_MESSAGE_RECEIVED, event); | ||
|
|
||
| // TODO: Handle timeout | ||
| }); | ||
| } |
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
Oops, something went wrong.