# shared-utils-kafka-common-lib

This library was generated with [Nx](https://nx.dev).

## Usage

Setup the required objects in your constructor.
You can specify topics to subscribe to in the config itself

```typescript
eventBus: EventBusService;

constructor(
    private readonly busLogger: MyLogger,
    private readonly kafkaLogger: MyLogger,
) {
    const options: EventBusOptions = {
        brokers: ['localhost:9092'],
        clientId: 'testClientId',
        groupId: 'testGroupId',
        //You can ask the bus to subscribe to these as it starts
        topics: ['topic-1', 'topic-2'],
    };

    this.eventBus = new KafkaEventBusService(options, new KafkaTransport(options, busLogger), kafkaLogger);
}
```
### Subscribing to topics
Subscribe to topics you want. In nest, you could do this in `onModuleInit()`

```typescript
async onModuleInit() {
    await this.eventBus.initialize();

    // Subscribing to topics given in the configuration will add a listener to that topic
    await this.eventBus.subscribe('topic-1', (event: Event) => {
      const obj = JSON.parse(event.value);
      console.log(obj);
    });

    // You can subscribe multiple times to the same topic with diferent listeners
    // You can even do this after calling .listen() on the bus
    await this.eventBus.subscribe('topic-1', (event: Event) => {
      const obj = JSON.parse(event.value);
      this.anotherListener(obj);
    });

    // You can subscribe to brand new topics
    // but you CANT DO THIS AFTER calling .listen()
    await this.eventBus.subscribe('a-whole-new-world', (event: Event) => {
      const obj = JSON.parse(event.value);
      this.anotherListener(obj);
    });
}
```

### Start Listening
When you're ready to rock, call `listen()`. **NOTE: You can't subscribe to new topics after this**, however you can call subscribe to topics you have already subscribed to before calling `listen()`. In Nest you could do this in the `onApplicationBootstrap` lifecycle event to be safe

```ts
async onApplicationBootstrap() {
    await this.eventBus.listen();
}
```

### Publishing Messages
To Publish Messages, Serialize your objects into a string and pass it to the event bus 

```ts
const obj = {
    foo: 'bar',
};

// Client code is responsible for serilalizing messages to be sent across the bus
await this.eventBus.publish('topic-1', {
      type: 'add',
      correlationId: '1234',
      value: JSON.stringify(obj),
});
```

### Logging
Implement EventBusLogger in your logger class and pass it into the event bus constructor. The bus uses `logCreator()` which is a special method that returns a function to be used by the transport to create logs. 

Note: In this example we are extending Nest's `Logger`, which is very similar to `EventBusLogger` which is why you don't see any other methods being implemented here

Note: Here we have set the context (the second parameter we have passed into the log methods) to "KafkaTransport"

```ts
import { Logger } from '@nestjs/common';
import { EventBusLogger, LogEntry, logLevel } from '@brandix/common';

export class MyLogger extends Logger implements EventBusLogger {

    logCreator(): (level: string) => (entry: LogEntry) => void {
        return (level: string) => {
            return (event: LogEntry) => {
                switch (Number.parseInt(level, 10)) {
                    case logLevel.ERROR:
                    case logLevel.NOTHING:
                        this.error(event.log.message, 'KafkaTransport');
                        break;
                    case logLevel.WARN:
                        this.warn(event.log.message, 'KafkaTransport');
                        break;
                    case logLevel.INFO:
                        this.log(event.log.message, 'KafkaTransport');
                        break;
                    case logLevel.DEBUG:
                        this.debug(event.log.message, 'KafkaTransport');
                        break;
                }
            };
        };
    }
}

```

## Configurations

```typescript
const options: EventBusOptions = {
        brokers: ['localhost:9092'],
        clientId: 'testClientId',
        groupId: 'testGroupId',
        topics: ['topic-1', 'topic-2'],
};
```

Here's a quick rundown of what this all means

| Property | Description |
|----------| ------------|
| brokers | a list of addresses of where the broker services are located
| clientId | the clientId for this client **Must be unique per client**
| groupId | the group this client belongs to there are two different behaviors based on if you use a unique groupid per client, or the same see the groupId section below
| topics | the client will subscribe to these topics as it starts, you dont have to specify them here, but can subscribe in the code before you start listening

### Choosing Group Ids
When choosing your groupId, you should be mindful that choosing a unique groupId for each client, or using the same one has different effects on how messages are processed. See [the official documentation on clients](https://kafka.apache.org/intro#intro_consumers)

In Essence:
* Using Unique groupId: All clients will receive messages sent to a subscribed topic (Pub/Sub style)
* Using Same groupId: Clients in the group will dequeue messages, meaning only one client will get the message from the subscribed topic (Message Queue Style)