Live Event App

* Added .gitignore
* Added a new directory for @pubnub/tomato

Author: jguz-pubnub

live-event-app/.gitignore

@@ -0,0 +1,2 @@
+tomato/node_modules
+tomato/package-lock.json

live-event-app/tomato/contracts/load-test-1.ts

@@ -0,0 +1,85 @@
+import { rand, randUuid, randDatabaseType, randFirstName } from '@ngneat/falso'
+import { envelope, successfulResponse, timetokenData } from '../utils/subscribe'
+
+export const name = 'loadTest'
+
+const SUB_KEY = 'demo'
+
+const MSGS_PER_SECOND = 100000
+const CHANNEL = 'demo'
+const CONCURRENT_USERS = 4
+const CHUNKS_PER_SECOND = 100
+const MSGS_PER_CHUNK = MSGS_PER_SECOND / CHUNKS_PER_SECOND
+const CHUNK_DELAY_MS = (1 / CHUNKS_PER_SECOND) * 1000
+
+const users = Array.from(Array(CONCURRENT_USERS), () => ({
+  uuid: randUuid(),
+}))
+
+function generatePayload(currentItemIndex: number) {
+  return {
+    id: randUuid(),
+    text: "Lorem ipsum dolor sit amet",
+    createdAt: (new Date().toISOString())
+  }
+}
+
+function generateEnvelopes(startTimetoken: string, amount: number) {
+  const result = []
+  let start = BigInt(startTimetoken)
+
+  for (let i = 0; i < amount; i++) {
+    const messageTimetoken = (start++).toString()
+    const user = rand(users)
+
+    result.push(
+      envelope({
+        channel: CHANNEL,
+        sender: "user_0",
+        subKey: SUB_KEY,
+        publishingTimetoken: {
+          t: messageTimetoken,
+          r: 0,
+        },
+        payload: generatePayload(i),
+      })
+    )
+  }
+
+  return result
+}
+
+export default async function () {
+  let currentTimetoken = timetoken.now()
+
+  const request = await expect({
+    description: 'subscribe with timetoken zero',
+    validations: [],
+  })
+
+  await request.respond({
+    status: 200,
+    body: successfulResponse(timetokenData(currentTimetoken)),
+  })
+
+  while (true) {
+    await Promise.race([
+      sleep(CHUNK_DELAY_MS),
+      expect({
+        description: 'subscribe next',
+        validations: [],
+      }).then(async (request) => {
+        const envelopes = generateEnvelopes(currentTimetoken, MSGS_PER_CHUNK)
+
+        const nextTimetoken = timetoken.now()
+
+        await request.respond({
+          status: 200,
+          body: successfulResponse(timetokenData(nextTimetoken), envelopes),
+        })
+
+        currentTimetoken = nextTimetoken
+      }),
+    ])
+  }
+}

live-event-app/tomato/contracts/load-test-2.ts

@@ -0,0 +1,16 @@
+import { router } from '../utils/router'
+
+export const name = 'loadTestRouter'
+
+export default router(expect, {
+  'GET /time/:param': (req, params) => {
+    console.log(req.url.path)
+
+    return {
+      status: 200,
+      body: {
+        result: params.param,
+      },
+    }
+  },
+})

live-event-app/tomato/package.json

@@ -0,0 +1,12 @@
+{
+  "name": "tomato-chat",
+  "version": "1.0.0",
+  "description": "Description goes here",
+  "dependencies": {
+    "@ngneat/falso": "^6.2.0",
+    "@pubnub/tomato": "^1.6.0",
+    "find-my-way": "^7.3.1",
+    "pubnub": "^7.2.1",
+    "typescript": "^4.9.3"
+  }
+}

live-event-app/tomato/utils/router.ts

@@ -0,0 +1,64 @@
+import * as Router from 'find-my-way'
+
+type RouteHandler = (
+  req: {
+    method: string
+    body?: any
+    headers: Record<string, any>
+    url: { path: string; query: Record<string, string> }
+  },
+  params?: Record<string, any>
+) =>
+  | Promise<{
+      status: number
+      headers?: Record<string, any>
+      body?: any
+    }>
+  | {
+      status: number
+      headers?: Record<string, any>
+      body?: any
+    }
+
+type Routes = {
+  [k: string]: RouteHandler
+}
+
+export function router(expect: any, routes: Routes) {
+  const router = Router()
+
+  for (const [key, handler] of Object.entries(routes)) {
+    const [method, path] = key.split(' ')
+
+    router.on(method.toUpperCase() as Router.HTTPMethod, path, handler as any)
+  }
+
+  return async function () {
+    while (true) {
+      const request = await expect({
+        description: 'any request',
+        validations: [],
+      })
+
+      const route = router.find(
+        request.method.toUpperCase() as Router.HTTPMethod,
+        request.url.path
+      )
+
+      if (!route) {
+        await request.respond({ status: 404 })
+      } else {
+        const response = await (route.handler as RouteHandler)(
+          request,
+          route.params
+        )
+
+        await request.respond({
+          status: response?.status,
+          headers: response?.headers,
+          body: response?.body,
+        })
+      }
+    }
+  }
+}

live-event-app/tomato/utils/subscribe.ts

@@ -0,0 +1,78 @@
+export type TimetokenData = {
+  t: string
+  r: number
+}
+
+export type Envelope = {
+  a: string
+  b?: string
+  c: string
+  d: any
+  e?: number
+  f: number
+  i: string
+  k: string
+  o?: TimetokenData
+  p: TimetokenData
+  u?: any
+}
+
+export type SubscribeResponse = {
+  t: TimetokenData
+  m: Envelope[]
+}
+
+export type SubscribeErrorResponse = {
+  error: true
+  status: number
+  message: string
+  service: 'Access Manager'
+  payload?: {
+    channels?: string[]
+    'channel-groups'?: string[]
+  }
+}
+
+export type PublishResponse = [number, string, string]
+
+export const timetokenData = (
+  timetoken: string,
+  region?: number
+): TimetokenData => ({
+  t: timetoken,
+  r: region ?? 1,
+})
+
+export const envelope = (input: {
+  shard?: string
+  subscriptionMatch?: string
+  channel: string
+  payload: any
+  messageType?: number
+  flags?: number
+  sender: string
+  subKey: string
+  metadata?: any
+  originatingTimetoken?: TimetokenData
+  publishingTimetoken: TimetokenData
+}): Envelope => ({
+  a: input.shard ?? '1',
+  b: input.subscriptionMatch ?? input.channel,
+  c: input.channel,
+  d: input.payload,
+  e: input.messageType ?? 0,
+  f: input.flags ?? 0,
+  i: input.sender,
+  k: input.subKey,
+  o: input.originatingTimetoken,
+  p: input.publishingTimetoken,
+  u: input.metadata,
+})
+
+export const successfulResponse = (
+  timetoken: TimetokenData,
+  envelopes: Envelope[] = []
+): SubscribeResponse => ({
+  t: timetoken,
+  m: envelopes,
+})