• @PgNotifyEventPattern()
• @PgNotifyMessagePattern()

These are an alternative to standard decorators:

• @EventPattern()
• @MessagePattern()

Message handler's binding can be used only within controller classes.

import { PgNotifyContext, PgNotifyEventPattern, PgNotifyMessagePattern } from 'nestjs-pg-notify';

@Controller()
export class AppController {

  @PgNotifyEventPattern({event: 'greeting'})
  @UsePipes(new ValidationPipe())
  onGreetingEvent(@Payload() payload: any, @Ctx() context: PgNotifyContext): void {
    Logger.log(payload.message);
  }

  @PgNotifyMessagePattern('greeting')
  @UsePipes(new ValidationPipe())
  onGreetingRequest(@Payload() payload: any, @Ctx() context: PgNotifyContext): string {
    Logger.log(payload.message);
    return 'Hello!';
  }

}

The standard decorator @Ctx() allows access to the context of the incoming request. In our case, the context object is an instance of PgNotifyContext.

Setup PgNotifyClient as client proxy

The client proxy can be registered as a custom provider. The configuration is the same as the configuration of the PgNotifyServer.

import { PgNotifyClient } from 'nestjs-pg-notify';

@Module({
  providers: [
    {
      provide: 'PG_NOTIFY_CLIENT',
      useFactory: (): ClientProxy => new PgNotifyClient({
        connection: {
          host: 'localhost',
          port: 5432,
          database: 'pgnotify',
          user: 'pgnotify',
          password: 'pgnotify',
        },
        strategy: {
          retryInterval: 1_000,
          retryTimeout: Infinity,
        }, 
      })
    },
  ],
  exports: [
    'PG_NOTIFY_CLIENT',
  ]
})
export class AppModule {}

Then we can inject the client proxy.

import { PgNotifyResponse } from 'nestjs-pg-notify';

export class AppService {
  constructor(
    @Inject('PG_NOTIFY_CLIENT')
    private readonly client: ClientProxy,
  ) {}
   
  sendRequest(): Observable<PgNotifyResponse> {
    // Send request and expect response
    return this.client.send('greeting', {message: 'Hello!'}).pipe(
      timeout(2_000),
      tap(response => Logger.debug(response)),
    );
  }
  
  emitEvent(): Observable<void> {
    // Emit event
    return this.client.emit({event: 'greeting'}, {message: 'Hello!'});
  }
}

Exception filters

The client proxy generates request identifier when we send requests using client.send(). The request identifier in the context of the incoming request means that we need to prepare an error response for the client.

We can use the PgNotifyResponse.error() factory in order to unify the structure of the response.

import { PgNotifyContext, PgNotifyResponse } from 'nestjs-pg-notify';

@Catch()
export class ExceptionFilter implements ExceptionFilter {
  catch(error: Error, host: ArgumentsHost): Observable<PgNotifyResponse|void> {
    const {status, message} = parseError(error);
    const context = host.switchToRpc().getContext<PgNotifyContext>();
    const requestId = context.getRequestId();

    Logger.error(message, error.stack, 'PgNotifyExceptionFilter');

    if (requestId) {
      return of(PgNotifyResponse.error(message, status));
    }

    return of(undefined);
  }
}

Then we can register the filter using the standard @UseFilters() decorator. It supports method-scope and controller-scope modes.

@Controller()
@UseFilters(ExceptionFilter)
export class AppController {
  // ...
}

Interceptors

import { PgNotifyContext } from 'nestjs-pg-notify';

@Injectable()
export class LoggingInterceptor implements NestInterceptor {
  public intercept(context: ExecutionContext, next: CallHandler): Observable<void> {
    const pgNotifyContext = context
      .switchToRpc()
      .getContext<PgNotifyContext>();

    return next.handle().pipe(
      tap(() => Logger.log(JSON.stringify(pgNotifyContext), LoggingInterceptor.name)),
    );
  }
}

To register interceptor we can use @UseInterceptors() decorator. It also supports method-scope and controller-scope modes.

@Controller()
@UseInterceptors(LoggingInterceptor)
export class AppController {
  // ...
}

API

API documentation is available here.

References

1. PostgreSQL Documentation:
   • Asynchronous Notification
   • NOTIFY
   • LISTEN
2. PgBouncer Documentation:
   • Transaction pool mode does not support NOTIFY/LISTEN features
3. NestJS Documentation:
   • Microservices
   • Hybrid applications
   • Custom transporters
4. Dependencies:
   • pg-listen

License

This project is licensed under the MIT License.