Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | import { Kafka, SASLOptions } from 'kafkajs'
import Adapter from '../../lib/adapter.js'
import GleeQuoreMessage from '../../lib/message.js'
import { KafkaAdapterConfig, KafkaAuthConfig } from '../../index.d.js'
class KafkaAdapter extends Adapter {
private kafka: Kafka
private firstConnect = true
name(): string {
return 'Kafka adapter'
}
async connect(): Promise<void> {
await this._connect()
}
async _connect() {
const kafkaOptions: KafkaAdapterConfig = await this.resolveProtocolConfig(
'kafka'
)
const auth: KafkaAuthConfig = await this.getAuthConfig(kafkaOptions?.auth)
const securityRequirements = this.AsyncAPIServer.security().map(
(sec) => {
const secName = Object.keys(sec.values())[0]
return this.parsedAsyncAPI.components().securitySchemes().get(secName)
}
)
const userAndPasswordSecurityReq = securityRequirements.find(
(sec) => sec.type() === 'userPassword'
)
const scramSha256SecurityReq = securityRequirements.find(
(sec) => sec.type() === 'scramSha256'
)
const scramSha512SecurityReq = securityRequirements.find(
(sec) => sec.type() === 'scramSha512'
)
const brokerUrl = new URL(this.AsyncAPIServer.url())
this.kafka = new Kafka({
clientId: 'glee-app',
brokers: [brokerUrl.host],
ssl: {
rejectUnauthorized: auth?.rejectUnauthorized,
key: auth?.key,
cert: auth?.cert,
},
sasl: {
mechanism:
(scramSha256SecurityReq ? 'scram-sha-256' : undefined) ||
(scramSha512SecurityReq ? 'scram-sha-512' : undefined) ||
'plain',
username: userAndPasswordSecurityReq ? auth?.username : undefined,
password: userAndPasswordSecurityReq ? auth?.password : undefined,
} as SASLOptions,
})
const consumer = this.kafka.consumer({ groupId: 'glee-group' })
consumer.on('consumer.connect', () => {
Iif (this.firstConnect) {
this.firstConnect = false
this.emit('connect', {
name: this.name(),
adapter: this,
connection: consumer,
channels: this.getSubscribedChannels(),
})
}
})
await consumer.connect()
const subscribedChannels = this.getSubscribedChannels()
await consumer.subscribe({
topics: subscribedChannels,
fromBeginning: true,
})
await consumer.run({
eachMessage: async ({ topic, partition, message }) => {
const msg = this._createMessage(topic, partition, message)
this.emit('message', msg, consumer)
},
})
}
async send(message: GleeQuoreMessage) {
const producer = this.kafka.producer()
await producer.connect()
await producer.send({
topic: message.channel,
messages: [
{
key: message.headers.key,
value: message.payload,
timestamp: message.headers.timestamp,
},
],
})
await producer.disconnect()
}
_createMessage(topic, partition, message) {
return new GleeQuoreMessage({
channel: topic,
payload: message.value,
headers: {
partition,
key: message.key,
offset: message.offset,
timestamp: message.timestamp,
...message.headers,
},
})
}
}
export default KafkaAdapter
|