Repository files navigation

Pub/Sub is an asynchronous messaging service that decouples services that produce events from services that process events.

You can use Pub/Sub as messaging-oriented middleware or event ingestion and delivery for streaming analytics pipelines.

Pub/Sub offers durable message storage and real-time message delivery with high availability and consistent performance at scale

To start building Pub/Sub-based microservices, first install the required packages:

$ npm i --save @google-cloud/pubsub nestjs-google-pubsub-microservice

To use the Pub/Sub transporter, pass the following options object to thecreateMicroservice() method:

constapp=awaitNestFactory.createMicroservice<MicroserviceOptions>(ApplicationModule,{strategy:newGCPubSubServer({topic:'cats_topic',subscription:'cats_subscription',client:{projectId:'microservice',},}),},);

Theoptions property is specific to the chosen transporter. TheGCloud Pub/Sub transporter exposes the properties described below.

constclient=newGCPubSubClient({client:{apiEndpoint:'localhost:8681',projectId:'microservice',},});client.send('pattern','Hello world!').subscribe((response)=>console.log(response));

In more sophisticated scenarios, you may want to access more information about the incoming request. When using the Pub/Sub transporter, you can access theGCPubSubContext object.

@MessagePattern('notifications')getNotifications(@Payload()data:number[], @Ctx()context:GCPubSubContext){console.log(`Pattern:${context.getPattern()}`);}

To access the original Pub/Sub message (with theattributes,data,ack andnack), use thegetMessage() method of theGCPubSubContext object, as follows:

@MessagePattern('notifications')getNotifications(@Payload()data:number[], @Ctx()context:GCPubSubContext){console.log(context.getMessage());}

To make sure a message is never lost, Pub/Sub supportsmessage acknowledgements. An acknowledgement is sent back by the consumer to tell Pub/Sub that a particular message has been received, processed and that Pub/Sub is free to delete it. If a consumer dies (its subscription is closed, connection is closed, or TCP connection is lost) without sending an ack, Pub/Sub will understand that a message wasn't processed fully and will re-deliver it.

To enable manual acknowledgment mode, set thenoAck property tofalse:

{replyTopic:'cats_topic_reply',replySubscription:'cats_subscription_reply',noAck:false,client:{projectId:'microservice',},},

When manual consumer acknowledgements are turned on, we must send a proper acknowledgement from the worker to signal that we are done with a task.

@MessagePattern('notifications')getNotifications(@Payload()data:number[], @Ctx()context:GCPubSubContext){constoriginalMsg=context.getMessage();originalMsg.ack();}

Pub/Sub requires a graceful shutdown properly configured in order to work correctly, otherwise some messages acknowledges can be lost. Therefore, don't forget to call client close:

exportclassGCPubSubControllerimplementsOnApplicationShutdown{client:ClientProxy;constructor(){this.client=newGCPubSubClient({});}onApplicationShutdown(){returnthis.client.close();}}