feat: Add parser (#142)

Author: simenandre

README.md

@@ -60,6 +60,11 @@ documentation][docs]
 - `parseJson` (boolean, default is `true`). When set to true, uses `JSON.parse`
   to parse the data sent through PubSub.
 
+- `parser` (function, optional). This option can be used to parse data coming
+  from PubSub. The function must return data (this data is passed to the
+  handler, as `data`). You can optionally throw an exception if the data could
+  not be parsed, which in turn can be caught by `onError` if you'd like.
+
 ## Contributing
 
 We love contributions! 🙏 Bug reports and pull requests are welcome on [GitHub].

examples/fastify-plugin.ts

@@ -1,24 +1,26 @@
 import { TypeBoxTypeProvider } from '@fastify/type-provider-typebox';
 import Fastify from 'fastify';
-import { PubSubHandler, pubSubFastifyPlugin } from '../src';
+import { z } from 'zod';
+import { pubSubFastifyPlugin } from '../src';
 
-interface HandlerArguments {
-  name: string;
-  party: {
-    name: string;
-  };
-  bookingId: string;
-}
+const inputData = z.object({
+  name: z.string(),
+  party: z.object({
+    name: z.string(),
+  }),
+  bookingId: z.string(),
+});
 
 const server = () => {
   const fastify = Fastify().withTypeProvider<TypeBoxTypeProvider>();
 
-  const handler: PubSubHandler<HandlerArguments> = ({ data, log }) => {
-    const { name, party, bookingId } = data;
-    log.info(`${name} from ${party.name} had a booking with id ${bookingId}`);
-  };
-
-  fastify.register(pubSubFastifyPlugin, { handler });
+  fastify.register(pubSubFastifyPlugin, {
+    parser: d => inputData.parse(d),
+    handler: ({ data, log }) => {
+      const { name, party, bookingId } = data;
+      log.info(`${name} from ${party.name} had a booking with id ${bookingId}`);
+    },
+  });
 
   fastify.server.listen(8000);
 };

package.json

@@ -45,6 +45,7 @@
     "@sinclair/typebox": "^0.24.0",
     "@types/express": "^4.17.13",
     "fastify": "^4.0.1",
+    "fastify-plugin": "^4.2.0",
     "pino": "^8.0.0",
     "pino-cloud-logging": "^1.0.3"
   },
@@ -64,6 +65,7 @@
     "ts-jest": "^28.0.4",
     "typedoc": "^0.23.7",
     "typedoc-plugin-markdown": "^3.1.1",
-    "typescript": "^4.3"
+    "typescript": "^4.3",
+    "zod": "^3.18.0"
   }
 }

src/__tests__/common.test.ts

@@ -65,4 +65,27 @@ describe('common', () => {
       log: {} as pino.Logger,
     });
   });
+
+  it('should run parser', async () => {
+    const message = createPubSubdata({}, true);
+    const handler = jest.fn();
+    const parser = jest.fn(_ => '');
+
+    await handlePubSubMessage({
+      message,
+      handler,
+      parser: data => parser(data),
+      context: { hello: 'there' },
+      log: {} as pino.Logger,
+    });
+
+    expect(handler).toHaveBeenCalledWith({
+      message,
+      data: '',
+      context: { hello: 'there' },
+      log: {} as pino.Logger,
+    });
+
+    expect(parser).toHaveBeenCalledWith({});
+  });
 });

src/common.ts

@@ -1,33 +1,24 @@
 import pino from 'pino';
 import { gcpLogOptions } from 'pino-cloud-logging';
-import {
-  PubSubHandler,
-  PubSubHandlerResponse,
-  PubSubMessageType,
-} from './types';
+import { HandlePubSubMessageArgs, PubSubHandlerResponse } from './types';
 
-export interface HandlePubSubMessageArgs<Context = unknown> {
-  message: PubSubMessageType;
-  handler: PubSubHandler;
-  parseJson?: boolean;
-  context?: Context;
-  log?: pino.Logger;
-}
-
-export async function handlePubSubMessage<Context = unknown>(
-  args: HandlePubSubMessageArgs<Context>,
+export async function handlePubSubMessage<Data, Context>(
+  args: HandlePubSubMessageArgs<Data, Context>,
 ): Promise<PubSubHandlerResponse | void> {
   const {
     message,
     parseJson = true,
+    parser,
     handler,
     context,
     log = pino(gcpLogOptions()),
   } = args;
-  let data = Buffer.from(message.data, 'base64').toString().trim();
+  const bufferString = Buffer.from(message.data, 'base64').toString().trim();
+
+  let data = parseJson ? JSON.parse(bufferString) : bufferString;
 
-  if (parseJson) {
-    data = JSON.parse(data);
+  if (parser) {
+    data = await parser(data);
   }
 
   return handler({

src/methods/cloud-functions.ts

@@ -4,8 +4,8 @@ import { gcpLogOptions } from 'pino-cloud-logging';
 import { handlePubSubMessage } from '../common';
 import { PubSubConfig, PubSubHandler } from '../types';
 
-export interface PubSubCloudFunctionsConfig
-  extends Omit<PubSubConfig, 'handler' | 'path'> {
+export interface PubSubCloudFunctionsConfig<Data, Context>
+  extends Omit<PubSubConfig<Data, Context>, 'handler' | 'path'> {
   logger?: pino.LoggerOptions;
 }
 
@@ -14,9 +14,9 @@ export type CloudFunctionFun = (
   res: express.Response,
 ) => Promise<void>;
 
-export function createPubSubCloudFunctions<T = unknown>(
-  handler: PubSubHandler<T>,
-  options: PubSubCloudFunctionsConfig = {},
+export function createPubSubCloudFunctions<Data = unknown, Context = unknown>(
+  handler: PubSubHandler<Data, Context>,
+  options: PubSubCloudFunctionsConfig<Data, Context> = {},
 ): CloudFunctionFun {
   const { parseJson, onError, logger } = options;
   return async (req, res): Promise<void> => {

src/methods/fastify-plugin.ts

@@ -1,11 +1,12 @@
-import { FastifyPluginAsync, FastifyRequest } from 'fastify';
+import { FastifyInstance } from 'fastify';
+import fp from 'fastify-plugin';
 import { handlePubSubMessage } from '../common';
 import { PubSubConfig, PubSubRequest, PubSubRequestType } from '../types';
 
-export const pubSubFastifyPlugin: FastifyPluginAsync<PubSubConfig> = async (
-  fastify,
-  options,
-) => {
+const pubSubFastifyPluginFn = async <Data, Context>(
+  fastify: FastifyInstance,
+  options: PubSubConfig<Data, Context>,
+): Promise<void> => {
   const { path = '/', handler, parseJson, onError } = options;
   fastify.post<{
     Body: PubSubRequestType;
@@ -19,11 +20,11 @@ export const pubSubFastifyPlugin: FastifyPluginAsync<PubSubConfig> = async (
     },
     async (req, reply) => {
       try {
-        const res = await handlePubSubMessage<FastifyRequest>({
+        const res = await handlePubSubMessage({
           message: req.body.message,
           handler,
           parseJson,
-          context: req,
+          // context: req,
           log: req.log,
         });
 
@@ -38,3 +39,8 @@ export const pubSubFastifyPlugin: FastifyPluginAsync<PubSubConfig> = async (
     },
   );
 };
+
+export const pubSubFastifyPlugin = fp(pubSubFastifyPluginFn, {
+  name: 'pubsub-http-handler',
+  fastify: '4.x',
+});

src/methods/server.ts

@@ -2,7 +2,8 @@ import Fastify, { FastifyInstance, FastifyServerOptions } from 'fastify';
 import { PubSubConfig, PubSubHandler } from '../types';
 import { pubSubFastifyPlugin } from './fastify-plugin';
 
-export interface PubSubServerConfig extends Omit<PubSubConfig, 'handler'> {
+export interface PubSubServerConfig<Data, Context>
+  extends Omit<PubSubConfig<Data, Context>, 'handler'> {
   /**
    * Will automatically pick up PORT environment variable.
    * @default 8000
@@ -31,9 +32,9 @@ export interface CreatePubSubHandlerResponse {
   fastify: FastifyInstance;
 }
 
-export function createPubSubServer<T = unknown>(
-  handler: PubSubHandler<T>,
-  config: PubSubServerConfig = {},
+export function createPubSubServer<Data, Context>(
+  handler: PubSubHandler<Data, Context>,
+  config: PubSubServerConfig<Data, Context> = {},
 ): CreatePubSubHandlerResponse {
   let { host = '0.0.0.0' } = config;
   const {

src/types.ts

@@ -1,25 +1,29 @@
 import { Static, Type } from '@sinclair/typebox';
 import { FastifyLoggerInstance } from 'fastify';
+import pino from 'pino';
 
-export interface PubSubConfig {
+export interface PubSubConfig<Data, Context> {
   /**
    * Handler
    */
-  handler: PubSubHandler;
+  handler: PubSubHandler<Data, Context>;
   /**
    * OnError Handler
    *
    * When this is set, errors will not be
    * thrown.
    */
   onError?: OnErrorHandler;
+
+  parser?: (data: unknown) => Data | Promise<Data>;
   /**
    * This will run JSON.parse on request data
    *
    * **Tip**: `false` when sending strings
    * @default true
    */
   parseJson?: boolean; // Defaults to true
+
   /**
    * Use this to set a different path
    * @default /
@@ -46,11 +50,20 @@ export class PubSubHandlerResponse {
   statusCode?: number;
 }
 
-export type PubSubHandler<T = any> = (args: {
+export type PubSubHandler<Data, Context> = (args: {
   message: PubSubMessageType;
-  data?: T;
-  context?: unknown;
+  data: Data;
+  context?: Context;
   log: FastifyLoggerInstance;
 }) => Promise<PubSubHandlerResponse | void> | PubSubHandlerResponse | void;
 
 export type OnErrorHandler = (error: unknown) => void | Promise<void>;
+
+export interface HandlePubSubMessageArgs<Data, Context> {
+  message: PubSubMessageType;
+  handler: PubSubHandler<Data, Context>;
+  parseJson?: boolean;
+  parser?: (data: unknown) => Data;
+  context?: Context;
+  log?: pino.Logger;
+}

yarn.lock

@@ -2022,6 +2022,11 @@ fast-uri@^2.1.0:
   resolved "https://registry.yarnpkg.com/fast-uri/-/fast-uri-2.1.0.tgz#9279432d6b53675c90116b947ed2bbba582d6fb5"
   integrity sha512-qKRta6N7BWEFVlyonVY/V+BMLgFqktCUV0QjT259ekAIlbVrMaFnFLxJ4s/JPl4tou56S1BzPufI60bLe29fHA==
 
+fastify-plugin@^4.2.0:
+  version "4.2.0"
+  resolved "https://registry.yarnpkg.com/fastify-plugin/-/fastify-plugin-4.2.0.tgz#fb28ec08312777ad343059177f13277ca3e7c3b1"
+  integrity sha512-hovKzEXZc2YgeuXn41/2EA/IaIOdRu1pB9WKgnzDBj3lhKSdDCEsckHa7I6LiT/LhflvAQX7ZY8IQ6eBX0htTg==
+
 fastify@^4.0.1:
   version "4.2.0"
   resolved "https://registry.yarnpkg.com/fastify/-/fastify-4.2.0.tgz#e8a56875083cab0fb0f1dd844b31f0fd145a6900"
@@ -4492,3 +4497,8 @@ yargs@^17.3.1:
     string-width "^4.2.3"
     y18n "^5.0.5"
     yargs-parser "^21.0.0"
+
+zod@^3.18.0:
+  version "3.18.0"
+  resolved "https://registry.yarnpkg.com/zod/-/zod-3.18.0.tgz#2eed58b3cafb8d9a67aa2fee69279702f584f3bc"
+  integrity sha512-gwTm8RfUCe8l9rDwN5r2A17DkAa8Ez4Yl4yXqc5VqeGaXaJahzYYXbTwvhroZi0SNBqTwh/bKm2N0mpCzuw4bA==