Problem using EventEmitters [Weird behavior]

I'm trying to implement a simple notification in a webapp, I can't make it work with EventEmmiters

streamNotifications: publicProcedure.subscription(async function* (opts) {
    try {
      console.log(
        "Listeners before setup:",
        notificationEventEmitter.listenerCount("add"),
      );
      const iterable = on(notificationEventEmitter, "add");
      console.log(
        "Listeners after setup:",
        notificationEventEmitter.listenerCount("add"),
      );

      opts.signal?.addEventListener("abort", () => {
        console.log("Subscription aborted");
      });
      for await (const [notification] of iterable) {
        yield notification as TNotification;
      }
    } catch (error) {
      console.error("Error in notification stream:", error);
      throw error; // Rethrow or handle as appropriate
    } finally {
      console.log("Stream closed");
    }
  }),

streamNotifications: publicProcedure.subscription(async function* (opts) {
    try {
      console.log(
        "Listeners before setup:",
        notificationEventEmitter.listenerCount("add"),
      );
      const iterable = on(notificationEventEmitter, "add");
      console.log(
        "Listeners after setup:",
        notificationEventEmitter.listenerCount("add"),
      );

      opts.signal?.addEventListener("abort", () => {
        console.log("Subscription aborted");
      });
      for await (const [notification] of iterable) {
        yield notification as TNotification;
      }
    } catch (error) {
      console.error("Error in notification stream:", error);
      throw error; // Rethrow or handle as appropriate
    } finally {
      console.log("Stream closed");
    }
  }),

streamNotifications: publicProcedure.subscription(async function* (opts) {
    try {
      console.log(
        "Listeners before setup:",
        notificationEventEmitter.listenerCount("add"),
      );
      const iterable = on(notificationEventEmitter, "add");
      console.log(
        "Listeners after setup:",
        notificationEventEmitter.listenerCount("add"),
      );

      opts.signal?.addEventListener("abort", () => {
        console.log("Subscription aborted");
      });
      for await (const [notification] of iterable) {
        yield notification as TNotification;
      }
    } catch (error) {
      console.error("Error in notification stream:", error);
      throw error; // Rethrow or handle as appropriate
    } finally {
      console.log("Stream closed");
    }
  }),

streamNotifications: publicProcedure.subscription(async function* (opts) {
    try {
      console.log(
        "Listeners before setup:",
        notificationEventEmitter.listenerCount("add"),
      );
      const iterable = on(notificationEventEmitter, "add");
      console.log(
        "Listeners after setup:",
        notificationEventEmitter.listenerCount("add"),
      );

      opts.signal?.addEventListener("abort", () => {
        console.log("Subscription aborted");
      });
      for await (const [notification] of iterable) {
        yield notification as TNotification;
      }
    } catch (error) {
      console.error("Error in notification stream:", error);
      throw error; // Rethrow or handle as appropriate
    } finally {
      console.log("Stream closed");
    }
  }),

Looks like its connected but when I do something like emit an event

emitFirstNotification: publicProcedure.mutation(async ({ ctx }) => {
    const notifications = await ctx.db.notification.findFirstOrThrow({
      select: selectnotificacion.select,
      orderBy: { createdAt: "desc" },
    });
    console.log("Emmiting");
    const emmiting = notificationEventEmitter.emit("add", notifications);
    console.log(emmiting, "listeners?");
  }),

emitFirstNotification: publicProcedure.mutation(async ({ ctx }) => {
    const notifications = await ctx.db.notification.findFirstOrThrow({
      select: selectnotificacion.select,
      orderBy: { createdAt: "desc" },
    });
    console.log("Emmiting");
    const emmiting = notificationEventEmitter.emit("add", notifications);
    console.log(emmiting, "listeners?");
  }),

emitFirstNotification: publicProcedure.mutation(async ({ ctx }) => {
    const notifications = await ctx.db.notification.findFirstOrThrow({
      select: selectnotificacion.select,
      orderBy: { createdAt: "desc" },
    });
    console.log("Emmiting");
    const emmiting = notificationEventEmitter.emit("add", notifications);
    console.log(emmiting, "listeners?");
  }),

emitFirstNotification: publicProcedure.mutation(async ({ ctx }) => {
    const notifications = await ctx.db.notification.findFirstOrThrow({
      select: selectnotificacion.select,
      orderBy: { createdAt: "desc" },
    });
    console.log("Emmiting");
    const emmiting = notificationEventEmitter.emit("add", notifications);
    console.log(emmiting, "listeners?");
  }),

It simply doesnt work.
It does nothing, and it's like my emmiter

import type { Prisma } from "@prisma/client";
import type { selectnotificacion } from "../api/routers/webcheckinRouter";
import { EventEmitter } from "stream";

export type TNotification = Prisma.NotificationGetPayload<
  typeof selectnotificacion
>;
type NotificationEvents = {
  add: TNotification[];
};
const notificationEventEmitter = new EventEmitter<NotificationEvents>();
export default notificationEventEmitter;

import type { Prisma } from "@prisma/client";
import type { selectnotificacion } from "../api/routers/webcheckinRouter";
import { EventEmitter } from "stream";

export type TNotification = Prisma.NotificationGetPayload<
  typeof selectnotificacion
>;
type NotificationEvents = {
  add: TNotification[];
};
const notificationEventEmitter = new EventEmitter<NotificationEvents>();
export default notificationEventEmitter;

import type { Prisma } from "@prisma/client";
import type { selectnotificacion } from "../api/routers/webcheckinRouter";
import { EventEmitter } from "stream";

export type TNotification = Prisma.NotificationGetPayload<
  typeof selectnotificacion
>;
type NotificationEvents = {
  add: TNotification[];
};
const notificationEventEmitter = new EventEmitter<NotificationEvents>();
export default notificationEventEmitter;

import type { Prisma } from "@prisma/client";
import type { selectnotificacion } from "../api/routers/webcheckinRouter";
import { EventEmitter } from "stream";

export type TNotification = Prisma.NotificationGetPayload<
  typeof selectnotificacion
>;
type NotificationEvents = {
  add: TNotification[];
};
const notificationEventEmitter = new EventEmitter<NotificationEvents>();
export default notificationEventEmitter;

has 0 clients. (But the subscription is still connected according to the logger)

Solution

I use redis for my project https://github.com/bebore/ei-noah-bot but that's quite complex, I recommend just following the redis package docs. It's not much different from the event emitter other than that you need to (de)serialize the input and output.

GitHub

GitHub - BeBoRE/ei-noah-bot: De officiële Discord Bot voor de Sweat...

De officiële Discord Bot voor de Sweaty GG Chat. Contribute to BeBoRE/ei-noah-bot development by creating an account on GitHub.

Jump to solution

Problem using EventEmitters [Weird behavior]

Similar Threads

Similar Threads

Similar Threads