feat(pkg): refactor the implementation of package and add event bridge startegy

refactor the implementation of package and add event bridge startegy

gh-3

Author: Sunny Tyagi

README.md

@@ -1,140 +1,220 @@
 # Message bus queue connectors
+
 This is the package for the message bus queue connectors component for LoopBack 4 applications.
-It provides components to work with queues such as SQS, BullMQ
+It provides components to work with queues such as SQS, BullMQ and EventBridge
 
-[![LoopBack](https://github.com/loopbackio/loopback-next/raw/master/docs/site/imgs/branding/Powered-by-LoopBack-Badge-(blue)-@2x.png)](http://loopback.io/)
+[![LoopBack](<https://github.com/loopbackio/loopback-next/raw/master/docs/site/imgs/branding/Powered-by-LoopBack-Badge-(blue)-@2x.png>)](http://loopback.io/)
 
 ## Installation
 
-Install MessageBusQueueConnectorsComponent using `npm`;
+Install EventStreamConnectorComponent using `npm`;
 
 ```sh
-$ [npm install | yarn add] message-bus-queue-connectors
+$ [npm install | yarn add] @sourceloop/message-bus-queue-connectors
 ```
+## Flow Diagram
+
+<img width="659" alt="Screenshot 2025-06-06 at 10 53 06 AM" src="https://github.com/user-attachments/assets/baf1bcaa-5f67-44bb-a01a-b8d1c41644bc" />
 
 ## Basic Use
 
-Configure and load MessageBusQueueConnectorsComponent in the application constructor
+Configure and load EventStreamConnectorComponent in the application constructor
 as shown below.
 
-### SQS
 ```ts
-import {SqsProducerProvider, SQSBindings, SQSConsumerObserver, SQSConsumerProvider} from 'message-bus-queue-connectors/sqs';
+import {
+  EventStreamConnectorComponent
+} from '@sourceloop/message-bus-queue-connectors';
 
 // ...
-export class MyApplication extends BootMixin(ServiceMixin(RepositoryMixin(RestApplication))) {
+export class MyApplication extends BootMixin(
+  ServiceMixin(RepositoryMixin(RestApplication)),
+) {
   constructor(options: ApplicationConfig = {}) {
     super();
-    this.bind(SqsClientBindings.Config).to(
-      options.sqsConfig
-    );
-    
-    this.bind(SQSBindings.SQSProducerProvider).toProvider(SqsProducerProvider);
+    this.component(EventStreamConnectorComponent);
     // ...
+  }
+  // ...
+}
+```
 
-    // Add lifecycle observer
-    this.lifeCycleObserver(SQSConsumerObserver);
+### SQS
+
+To use SQS as their message queue, bind its required config and connector component in your application.
+
+```ts
+import {
+  SQSConnector,
+  SQSBindings,
+  EventStreamConnectorComponent
+} from '@sourceloop/message-bus-queue-connectors';
+
+// ...
+export class MyApplication extends BootMixin(
+  ServiceMixin(RepositoryMixin(RestApplication)),
+) {
+  constructor(options: ApplicationConfig = {}) {
+    super();
+
+    this.component(EventStreamConnectorComponent);
+    // SQS Config and its connector
+    this.bind(SQSBindings.Config).to({
+      queueConfig: {
+        QueueUrl: 'http://127.0.0.1:4566/000000000000/my-test-queue',
+        MessageRetentionPeriod: 60, // at least 60 seconds
+        MaximumMessageSize: 262144,
+        ReceiveMessageWaitTimeSeconds: 20, // typical polling time
+        VisibilityTimeout: 30, // 30 seconds
+      },
+      Credentials: {
+        region: 'us-east-1',
+        accessKeyId: 'test',
+        secretAccessKey: 'test',
+      },
+      ConsumerConfig: {
+        MaxNumberOfMessages: 10,
+        WaitTimeSeconds: 20,
+        maxConsumers: 2,
+      },
+    });
+
+    this.component(SQSConnector);
 
     // ...
   }
   // ...
 }
 ```
 
-#### SQS Config
+to make the application as consumer, pass 'isConsumer' flag to be true in SQS config. like
+
 ```ts
 const config = {
-  queueConfig: {
-    QueueUrl: "sqs-queue-url",,
-    MessageRetentionPeriod: 1,
-    MaximumMessageSize: 262144 ,
-    ReceiveMessageWaitTimeSeconds:  60,
-    VisibilityTimeout: 300,
-    },
-    Credentials: {
-      region: "aws-region",
-      accessKeyId: "aws-access-key-id",
-      secretAccessKey: "aws-secret-access-key",
-    },
-    ConsumerConfig: {
-      MaxNumberOfMessages: 10,
-      WaitTimeSeconds: 20,
-      maxConsumers: 2,
-    },
-}
-
+  // rest of ur config
+  isConsumer: true,
+};
 ```
-#### Consumer setup
-Below is consumer handler example.
-```ts
-this.bind(SQSBindings.SQSConsumerProvider).toProvider(SQSConsumerProvider);
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    this.bind(SQSBindings.SQSConsumerHandler).to(async (message: string) => {
-      console.log('Processing message SQS---------:', message);
-      
-    });
 
-```
 Please follow the [AWS SDK for JavaScript](https://docs.aws.amazon.com/sdk-for-javascript/v2/developer-guide/sqs-examples-send-receive-messages.html) for more information on the configuration.
 
-
 ### BullMQ
 
+To use BullMq as their message queue, bind its required config and connector component in your application.
+
 ```ts
-import {BullMQProducerProvider, BullMQBindings,  BullMQConsumerObserver} from 'message-bus-queue-connectors/bullmq';
+import {
+  BullMQConnector,
+  BullMQBindings,
+  EventStreamConnectorComponent,
+} from '@sourceloop/message-bus-queue-connectors';
 
 // ...
-export class MyApplication extends BootMixin(ServiceMixin(RepositoryMixin(RestApplication))) {
+export class MyApplication extends BootMixin(
+  ServiceMixin(RepositoryMixin(RestApplication)),
+) {
   constructor(options: ApplicationConfig = {}) {
     super();
-    this.bind(BullMQBindings.Config).to(
-      options.config
-    );
-    
-    this.bind(BullMQBindings.BullMQProducerProvider).toProvider(BullMQProducerProvider);
-    // ...
-    // Add lifecycle observer
-    this.lifeCycleObserver(BullMQConsumerObserver);
+
+    this.component(EventStreamConnectorComponent);
+
+    // Bull Mq config and connector
+    this.bind(BullMQBindings.Config).to({
+      QueueName: process.env.QUEUE_NAME ?? 'default-queue',
+      redisConfig: {
+        host: process.env.REDIS_HOST ?? 'localhost',
+        port: parseInt(process.env.REDIS_PORT ?? '6379'),
+        password: process.env.REDIS_PASSWORD ?? undefined,
+      },
+      producerConfig: {
+        defaultJobOptions: {
+          attempts: 3,
+          backoff: 5000,
+        },
+      },
+      consumerConfig: {
+        MinConsumers: 1,
+        MaxConsumers: 5,
+        QueuePollInterval: 2000,
+      },
+    });
+    this.component(BullMQConnector);
     // ...
   }
   // ...
 }
 ```
 
-#### BullMQ config
+to make the application as consumer, pass 'isConsumer' flag to be true in Bull config. like
+
 ```ts
 const config = {
-  queueConfig:{
-    QueueName: 'BullMQ1',
-  },
-  QueueName: 'BullMQ1',
-  producerConfig: {
-  defaultJobOptions: {attempts: 3,
-      backoff: {
-        type: 'exponential',
-        delay: 5000,
-      },
-    } 
-  },
-  consumerConfig: {
-    MaxConsumers: 1,
-    MinConsumers: 1,
-  },
-  redisConfig: {
-    host: process.env.REDIS_HOST ?? 'localhost',
-    port: +(process.env.REDIS_PORT ?? 6379),
-  }
- 
+  // rest of ur config
+  isConsumer: true,
+};
+```
+
+## Integration
+
+ @sourceloop/message-bus-queue-connectors provides a decorator '@producer()' that can be used to access the producer of each msg queue. It expects one arguement defining the type of queue, of which producer u want to use. like
+
+ ```ts 
+ @injectable({scope: BindingScope.TRANSIENT})
+export class EventConnector implements IEventConnector<PublishedEvents> {
+  constructor(
+    @producer(QueueType.EventBridge)
+    private producer: Producer,
+    @producer(QueueType.SQS)
+    private sqsProducer: Producer,
+    @producer(QueueType.BullMQ)
+    private bullMqProducer: Producer,
+  ) {}
+
+  // rest of implementation
+
 }
+ ```
+
+ Producer provider two ways of sending events - single event at a time and multiple event at a time.
+
+ ```ts
+ export type Producer<Stream extends AnyObject = AnyObject> = {
+    send: <Event extends keyof Stream>(data: Stream[Event], topic?: Event) => Promise<void>;
+    sendMultiple: <Event extends keyof Stream>(data: Stream[Event][], topic?: Event) => Promise<void>;
+};
+ ```
 
+It provides '@consumer' decorator to make a service as consumer. consumer needs to follow an interface.
+
+```ts
+export interface IConsumer<Stream extends AnyObject, Event extends keyof Stream> {
+    event: Event;
+    queue: QueueType;
+    handle(data: Stream[Event]): Promise<void>;
+}
 ```
 
-#### Consumer setup
-Below is consumer handler example.
+and can be used as 
+
 ```ts
-this.bind(BullMQBindings.BullMQConsumerProvider).toProvider(BullMQConsumerProvider);
-    
-    this.bind(BullMQBindings.BullMQConsumerHandler).to(async (message: string) => {
-      console.log('Processing message ---------:', message);      
-    });   
+import {
+  IConsumer,
+  QueueType,
+  consumer,
+} from '@sourceloop/message-bus-queue-connectors';
+import { OrchestratorStream, EventTypes, ProvisioningInputs } from '../../types';
+
+@consumer
+export class TenantProvisioningConsumerForEventSQS
+  implements IConsumer<OrchestratorStream, EventTypes.TENANT_PROVISIONING>
+{
+  constructor(
+  ) {}
+  event: EventTypes.TENANT_PROVISIONING = EventTypes.TENANT_PROVISIONING;
+  queue: QueueType = QueueType.SQS;
+  async handle(data: ProvisioningInputs): Promise<void> {    
+    console.log(`SQS: ${this.event} Event Recieved ` + JSON.stringify(data));
+    return;
+  }
+}
 ```

local-message-bus-queue-connectors-0.0.1.tgz

@@ -43,7 +43,7 @@
     "eslint": "eslint --report-unused-disable-directives .",
     "eslint:fix": "npm run eslint -- --fix",
     "pretest": "npm run rebuild",
-    "test": "lb-mocha --exit --allow-console-logs \"dist/__tests__\"",
+    "test": "lb-mocha --allow-console-logs \"dist/__tests__\"",
     "test:dev": "lb-mocha --allow-console-logs dist/__tests__/**/*.js",
     "clean": "lb-clean dist *.tsbuildinfo .eslintcache",
     "rebuild": "npm run clean && npm run build",

package.json

@@ -0,0 +1,64 @@
+import {expect, sinon} from '@loopback/testlab';
+import {ProducerApp} from './fixtures/producers-app';
+import {TestProducerService} from './fixtures/services/test-producer.service';
+import {setupProducerApplication} from './helpers/app-builder';
+import {QueueStub} from './stubs/bullmq-queue.stub';
+import {Events} from '../test-stream';
+import {DEFAULT_SOURCE} from '../../../constants';
+
+[
+  {
+    type: 'Stubbed queue',
+    queue: new QueueStub(),
+    condition: () => true,
+  },
+].forEach(({type, queue, condition}) => {
+  describe(`BullMq Connector: With ${type}`, () => {
+    let producerApp: ProducerApp | undefined;
+    let producerService: TestProducerService;
+    let listenerStub: sinon.SinonStub;
+    before(async function () {
+      if (!condition()) {
+        // eslint-disable-next-line  @typescript-eslint/no-invalid-this
+        this.skip();
+      }
+      producerApp = await setupProducerApplication(queue);
+      producerService = producerApp.getSync<TestProducerService>(
+        `services.TestProducerService`,
+      );
+      listenerStub = sinon.stub().resolves();
+      if (queue) {
+        queue.register(listenerStub);
+      }
+    });
+    beforeEach(() => {
+      listenerStub.reset();
+    });
+    after(async () => {
+      await producerApp?.stop();
+    });
+    describe('Producer', () => {
+      it('should produce an event for a particular topic', async () => {
+        await producerService.produceEventA('test string');
+        sinon.assert.calledWithExactly(
+          listenerStub,
+          'Event A',
+          'test string',
+          {},
+        );
+      });
+      it('should produce multiple events for a particular topic', async () => {
+        const data = ['test string 1', 'test string 2'];
+        await producerService.produceMultipleA(data);
+        sinon.assert.calledWithExactly(
+          listenerStub,
+          JSON.stringify(
+            data.map(d => ({name: 'Event A', data: d, type: Events.A})),
+          ),
+          '',
+          {},
+        );
+      });
+    });
+  });
+});