Emitter #366
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,6 @@ | ||
| import { CloudEvent } from "../event/cloudevent"; | ||
| import { HTTP, Message, Mode } from "../message"; | ||
| import { EventEmitter } from "events"; | ||
|
|
||
| /** | ||
| * Options is an additional, optional dictionary of options that may | ||
|
|
@@ -58,3 +59,66 @@ export function emitterFor(fn: TransportFunction, options = { binding: HTTP, mod | |
| } | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * A static class to emit CloudEvents within an application | ||
| */ | ||
| export class Emitter extends EventEmitter { | ||
| /** | ||
| * Singleton store | ||
| */ | ||
| static instance: Emitter | undefined = undefined; | ||
|
|
||
| /** | ||
| * Create an Emitter | ||
| * On v4.0.0 this class will only remains as Singleton to allow using the | ||
| * EventEmitter of NodeJS | ||
| */ | ||
| private constructor() { | ||
| super(); | ||
| } | ||
|
|
||
| /** | ||
| * Return or create the Emitter singleton | ||
| * | ||
| * @return {Emitter} return Emitter singleton | ||
| */ | ||
| static getInstance(): Emitter { | ||
| if (!Emitter.instance) { | ||
| Emitter.instance = new Emitter(); | ||
| } | ||
| return Emitter.instance; | ||
| } | ||
|
|
||
| /** | ||
| * Add a listener for eventing | ||
| * | ||
| * @param {string} event type to listen to | ||
| * @param {Function} listener to call on event | ||
| * @return {void} | ||
| */ | ||
| static on(event: "cloudevent" | "newListener" | "removeListener", listener: (...args: any[]) => void): void { | ||
lance marked this conversation as resolved.
Show resolved
Hide resolved
There was a problem hiding this comment. Why are we creating an event listening system for an event emitter? I do not expect an eventing system like this. There was a problem hiding this comment. This is a shortcut to avoid the There was a problem hiding this comment. In my opinion, the example added to the README is very illustrative. This PR makes it so that you can set up your emitter function using There was a problem hiding this comment. Yes this was the exact purpose, thanks for summarizing it so well There was a problem hiding this comment. Yup, the README example is great. const axios = require("axios").default;
const { emitterFor, Mode, CloudEvent, Emitter } = require("cloudevents");
function sendWithAxios(message) {
// Do what you need with the message headers
// and body in this function, then send the
// event
axios({
method: "post",
url: "...",
data: message.body,
headers: message.headers,
});
}
const emit = emitterFor(sendWithAxios, { mode: Mode.BINARY });
// Set the emit
Emitter.on("cloudevent", emit);
...
// In any part of the code will send the event
new CloudEvent({ type, source, data }).emit();I think the fundamental question is why are we adding back the A CloudEvent object and a CloudEvent emitter/receiver seem like separate concepts. So I wouldn't imagine a CloudEvent has a send/receive function. const axios = require("axios").default;
const { emitterFor, Mode, CloudEvent, Emitter } = require("cloudevents");
function sendWithAxios(message) {
// Do what you need with the message headers
// and body in this function, then send the
// event
axios({
method: "post",
url: "...",
data: message.body,
headers: message.headers,
});
}
const emit = emitterFor(sendWithAxios, { mode: Mode.BINARY });
// In any part of the code will send the event
const ce = new CloudEvent({ type, source, data });
emit(ce);Folks can add an eventing layer on-top of the primitives here. Maybe point out an error in the above alternative code? There was a problem hiding this comment. The emitEvent function was adding because EventEmitter is not able to "wait" for all listeners to finish async jobs. There was a problem hiding this comment.
Just to be clear, we are talking about two separate concepts. It's easy to get things mixed up when all of the terms are so similar. The two separate things as I see it are:
This pull request is about using what is described in 2) to facilitate and orchestrate 1) in an efficient way across a large system that may need to send
This pull request is not an alternative to that code. It builds upon that code to add a very useful feature, in my opinion, with very low overhead using Node.js built in, idiomatic capabilities. |
||
| this.getInstance().on(event, listener); | ||
| } | ||
|
|
||
| /** | ||
| * Emit an event inside this application | ||
| * | ||
| * @param {CloudEvent} event to emit | ||
| * @param {boolean} ensureDelivery fail the promise if one listener fail | ||
| * @return {void} | ||
| */ | ||
| static async emitEvent(event: CloudEvent, ensureDelivery = true): Promise<void> { | ||
| if (!ensureDelivery) { | ||
| // Ensure delivery is disabled so we don't wait for Promise | ||
| Emitter.getInstance().emit("cloudevent", event); | ||
| } else { | ||
| // Execute all listeners and wrap them in a Promise | ||
| await Promise.all( | ||
| Emitter.getInstance() | ||
| .listeners("cloudevent") | ||
| .map(async (l) => l(event)), | ||
| ); | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| import "mocha"; | ||
|
|
||
| import { emitterFor, HTTP, Mode, Message, Emitter } from "../../src"; | ||
|
|
||
| import { fixture, assertStructured } from "./emitter_factory_test"; | ||
|
|
||
| import { rejects, doesNotReject } from "assert"; | ||
lance marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| describe("Emitter Singleton", () => { | ||
|
There was a problem hiding this comment. The casts used in this test seem very fishy (casting to unknown, then an interface). Additionally, there are no comments, making the tests hard to understand. There was a problem hiding this comment. That is why we removed the eslint rule according with @lance |
||
| it("emit a Node.js 'cloudevent' event as an EventEmitter", async () => { | ||
| const msg: Message | unknown = await new Promise((resolve) => { | ||
| const fn = async (message: Message) => { | ||
| resolve(message); | ||
| }; | ||
| const emitter = emitterFor(fn, { binding: HTTP, mode: Mode.STRUCTURED }); | ||
| Emitter.on("cloudevent", emitter); | ||
|
|
||
| fixture.emit(false); | ||
| }); | ||
| let body: unknown = (<Message>(<unknown>msg)).body; | ||
| if (typeof body === "string") { | ||
| body = JSON.parse(body); | ||
| } | ||
| assertStructured({ ...(<any>body), ...(<Message>(<unknown>msg)).headers }); | ||
| }); | ||
|
|
||
| it("emit a Node.js 'cloudevent' event as an EventEmitter with ensureDelivery", async () => { | ||
| let msg: Message | unknown = undefined; | ||
| const fn = async (message: Message) => { | ||
| msg = message; | ||
| }; | ||
| const emitter = emitterFor(fn, { binding: HTTP, mode: Mode.STRUCTURED }); | ||
| Emitter.on("cloudevent", emitter); | ||
| await fixture.emit(true); | ||
| let body: any = (<Message>msg).body; | ||
| if (typeof body === "string") { | ||
| body = JSON.parse(body); | ||
| } | ||
| assertStructured({ ...(<any>body), ...(<Message>(<unknown>msg)).headers }); | ||
| }); | ||
|
|
||
| it("emit a Node.js 'cloudevent' event as an EventEmitter with ensureDelivery Error", async () => { | ||
| const emitter = async () => { | ||
| throw new Error("Not sent"); | ||
| }; | ||
| Emitter.on("cloudevent", emitter); | ||
| // Should fail with emitWithEnsureDelivery | ||
| await rejects(() => fixture.emit(true)); | ||
| // Should not fail with emitWithEnsureDelivery | ||
| // Work locally but not on Github Actions | ||
| if (!process.env.GITHUB_WORKFLOW) { | ||
| await doesNotReject(() => fixture.emit(false)); | ||
| } | ||
| }); | ||
| }); | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Modifying ESlint rules is great, but really out of scope for a PR about emitters.
I would imagine 2 PRs that are separate:
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think the point of this change is that the existing linter rules was causing this PR to generate warnings during the build. See #366 (comment). I'm not sure of the benefit returned by extracting this small change into a separate PR, landing that first, then landing this. In fact, I recall recently that you were not happy about the number of commits in this repo - this seems counter to that complaint! 😉
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hey, I'd quickly approve a change/fix to the eslint rules, but I don't want to approve a change that includes
while (msg === undefined) {. The eslint change is a different decision than emitter functionality and would like to be merged even if any emitter PRs were not merge. Does that make sense?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sure. It just seems pedantic to demand this.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Lance, I'm not demanding. This is just a review suggestion.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The goal was to merge the PR in a timely manner so at that time we thought it would make sense to have only one PR
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Having a PR reviewed in a timely manner is fair. In general, it would be easier to not go back and forth and review faster with a GitHub issue and small PR. There have been discussions about removing this before, and we just talked about splitting this repo into different modules, so it's difficult to approve this PR right away.
That said, this repo is likely going to change a bit with the split, and I don't want to block you if Lance approves.