-
Notifications
You must be signed in to change notification settings - Fork 140
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
24 changed files
with
459 additions
and
384 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| --- | ||
| '@segment/analytics-signals': minor | ||
| --- | ||
|
|
||
| Allow registration of middleware to allow for dropping and modification of signals | ||
|
|
||
| ```ts | ||
| class MyMiddleware implements SignalsMiddleware { | ||
| process(signal: Signal) { | ||
| if ( | ||
| signal.type === 'network' && | ||
| signal.data.action === 'request' && | ||
| ... | ||
| ) { | ||
| // drop or modify signal | ||
| return null | ||
| } else { | ||
| return signal | ||
| } | ||
| } | ||
| } | ||
| const signalsPlugin = new SignalsPlugin({ | ||
| middleware: [new MyMiddleware()] | ||
| }) | ||
| ``` |
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
61 changes: 61 additions & 0 deletions
61
packages/signals/signals-integration-tests/src/tests/signals-vanilla/middleware.test.ts
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,61 @@ | ||
| import { test, expect } from '@playwright/test' | ||
| import { IndexPage } from './index-page' | ||
|
|
||
| const basicEdgeFn = `const processSignal = (signal) => {}` | ||
|
|
||
| let indexPage: IndexPage | ||
|
|
||
| test('middleware', async ({ page }) => { | ||
| indexPage = await new IndexPage().load( | ||
| page, | ||
| basicEdgeFn, | ||
| {}, | ||
| { | ||
| skipSignalsPluginInit: true, | ||
| } | ||
| ) | ||
|
|
||
| await page.evaluate( | ||
| ({ settings }) => { | ||
| window.signalsPlugin = new window.SignalsPlugin({ | ||
| middleware: [ | ||
| { | ||
| load() { | ||
| return undefined | ||
| }, | ||
| process: function (signal) { | ||
| // @ts-ignore | ||
| signal.data['middleware'] = 'test' | ||
| return signal | ||
| }, | ||
| }, | ||
| ], | ||
| ...settings, | ||
| }) | ||
| window.analytics.load({ | ||
| writeKey: '<SOME_WRITE_KEY>', | ||
| plugins: [window.signalsPlugin], | ||
| }) | ||
| }, | ||
| { | ||
| settings: { | ||
| ...indexPage.defaultSignalsPluginTestSettings, | ||
| flushAt: 1, | ||
| }, | ||
| } | ||
| ) | ||
|
|
||
| /** | ||
| * Make an analytics.page() call, see that the middleware can modify the event | ||
| */ | ||
| await Promise.all([ | ||
| indexPage.makeAnalyticsPageCall(), | ||
| indexPage.waitForSignalsApiFlush(), | ||
| ]) | ||
|
|
||
| const instrumentationEvents = | ||
| indexPage.signalsAPI.getEvents('instrumentation') | ||
| expect(instrumentationEvents).toHaveLength(1) | ||
| const ev = instrumentationEvents[0] | ||
| expect(ev.properties!.data['middleware']).toEqual('test') | ||
| }) |
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 |
|---|---|---|
| @@ -1,47 +1,109 @@ | ||
| import { Emitter } from '@segment/analytics-generic-utils' | ||
| import { logger } from '../../lib/logger' | ||
| import { Signal } from '@segment/analytics-signals-runtime' | ||
| import { SignalGlobalSettings } from '../signals' | ||
|
|
||
| export interface EmitSignal { | ||
| emit: (signal: Signal) => void | ||
| } | ||
|
|
||
| const logSignal = (signal: Signal) => { | ||
| logger.info( | ||
| 'New signal:', | ||
| signal.type, | ||
| signal.data, | ||
| ...(signal.type === 'interaction' && 'change' in signal.data | ||
| ? ['change:', JSON.stringify(signal.data.change, null, 2)] | ||
| : []) | ||
| ) | ||
| export interface SignalsMiddlewareContext { | ||
| /** | ||
| * These are global application settings. They are considered unstable, and should only be used internally. | ||
| * @interal | ||
| */ | ||
| unstableGlobalSettings: SignalGlobalSettings | ||
| writeKey: string | ||
| } | ||
|
|
||
| export interface PluginSettings { | ||
| writeKey: string | ||
| } | ||
|
|
||
| export interface SignalsMiddleware { | ||
| /** | ||
| * Wait for .load to complete before emitting signals | ||
| * This blocks the signal emitter until all plugins are loaded. | ||
| */ | ||
| load(ctx: SignalsMiddlewareContext): Promise<void> | void | ||
| process(signal: Signal): Signal | null | ||
| } | ||
|
|
||
| export interface SignalEmitterSettings { | ||
| middleware?: SignalsMiddleware[] | ||
| } | ||
| export class SignalEmitter implements EmitSignal { | ||
| private emitter = new Emitter<{ add: [Signal] }>() | ||
| private listeners = new Set<(signal: Signal) => void>() | ||
| private middlewares: SignalsMiddleware[] = [] | ||
| private initialized = false // Controls buffering vs eager signal processing | ||
| private signalQueue: Signal[] = [] // Buffer for signals emitted before initialization | ||
|
|
||
| constructor(settings?: SignalEmitterSettings) { | ||
| if (settings?.middleware) this.middlewares.push(...settings.middleware) | ||
| } | ||
|
|
||
| // Emit a signal | ||
| emit(signal: Signal): void { | ||
| if (!this.initialized) { | ||
| // Buffer the signal if not initialized | ||
| this.signalQueue.push(signal) | ||
| return | ||
| } | ||
|
|
||
| // Process and notify listeners | ||
| this.processAndEmit(signal) | ||
| } | ||
|
|
||
| emit(signal: Signal) { | ||
| logSignal(signal) | ||
| this.emitter.emit('add', signal) | ||
| // Process and emit a signal | ||
| private processAndEmit(signal: Signal): void { | ||
| // Apply plugin; drop the signal if any plugin returns null | ||
| for (const plugin of this.middlewares) { | ||
| const processed = plugin.process(signal) | ||
| if (processed === null) return // Drop the signal | ||
| } | ||
|
|
||
| // Notify listeners | ||
| for (const listener of this.listeners) { | ||
| listener(signal) | ||
| } | ||
| } | ||
|
|
||
| subscribe(listener: (signal: Signal) => void) { | ||
| // Prevent duplicate subscriptions | ||
| // Initialize the emitter, load plugin, flush the buffer, and enable eager processing | ||
| async initialize({ | ||
| globalSettings, | ||
| writeKey, | ||
| }: { | ||
| globalSettings: SignalGlobalSettings | ||
| writeKey: string | ||
| }): Promise<void> { | ||
| if (this.initialized) return | ||
|
|
||
| // Wait for all plugin to complete their load method | ||
| await Promise.all( | ||
| this.middlewares.map((mw) => | ||
| mw.load({ unstableGlobalSettings: globalSettings, writeKey }) | ||
| ) | ||
| ) | ||
|
|
||
| this.initialized = true | ||
|
|
||
| // Process and emit all buffered signals | ||
| while (this.signalQueue.length > 0) { | ||
| const signal = this.signalQueue.shift() as Signal | ||
| this.processAndEmit(signal) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Listen to signals emitted, once they have travelled through the plugin pipeline. | ||
| * This is equivalent to a destination plugin. | ||
| */ | ||
| subscribe(listener: (signal: Signal) => void): void { | ||
| if (!this.listeners.has(listener)) { | ||
| logger.debug('subscribed') | ||
| this.listeners.add(listener) | ||
| } | ||
| this.emitter.on('add', listener) | ||
| } | ||
|
|
||
| unsubscribe(listener: (signal: Signal) => void) { | ||
| // Unsubscribe a listener | ||
| unsubscribe(listener: (signal: Signal) => void): void { | ||
| this.listeners.delete(listener) | ||
| logger.debug('unsubscribed') | ||
| this.emitter.off('add', listener) | ||
| } | ||
|
|
||
| once(listener: (signal: Signal) => void) { | ||
| this.emitter.once('add', listener) | ||
| } | ||
| } |
Oops, something went wrong.