Balancer core

Usage sample

``typescript
import {
MessageBalancer,
MessageCache,
MessageStorage,
Db,
migrateDb,
createMethodPerLevelLoggerAdapter,
DefaultMessageBalancerDiagnosticsAdapter,
DefaultMessageCacheDiagnosticsAdapter,
DefaultMessageStorageDiagnosticsAdapter,
ProcessMessageBatchResultItem
} from '@targetprocess/balancer-core'
import {makeLogger} from 'loggerism'
import {Pool} from 'pg'
import * as promClient from 'prom-client'

type MessageBalancerA = MessageBalancer<{data?: string; retry?: number}>

type MessageBalancerB = MessageBalancer<{data?: string}>

async function main() {
const [balancerA, balancerB] = await createAndInitBalancers()

await balancerA.storeMessage({
partitionKey: 'partition#1',
content: Buffer.alloc(128),
properties: {data: 'some arbitrary data'}
})

await balancerB.storeMessage({
partitionKey: 'partition#2',
content: Buffer.alloc(128),
properties: {data: 'some arbitrary data'}
})

await balancerA.processNextMessage(async message => {
try {
const {partitionGroup, partitionKey} = message
console.log(Processed message from partition "${partitionGroup}/${partitionKey}")
return {type: 'Ok'}
} catch {
const properties = {
...message.properties,
retry: (message.properties?.retry || 0) + 1
}
// Push message back with updated properties
return {type: 'Requeue', update: {properties}}
}
})

// Using batch API
balancerB.processNextMessageBatch(
async messages => {
const results = [] as ProcessMessageBatchResultItem<{data?: string}>[]

for (const message of messages) {
try {
const {partitionGroup, partitionKey} = message
console.log(Processed message from partition "${partitionGroup}/${partitionKey}")
results.push({messageId: message.messageId, type: 'Ok'})
} catch {
// Push message back with no properties update
results.push({messageId: message.messageId, type: 'Requeue'})
}
}

return results
},
{maxBatchSize: 10}
)
}

async function createAndInitBalancers(): Promise<[MessageBalancerA, MessageBalancerB]> {
const pool = new Pool({
connectionString: process.env.POSTGRES_CONNECTION_STRING,
max: process.env.POSTGRES_POOL_MAX
})

pool.on('error', error => {
// Handle error here
console.error(error)
})

await migrateDb({pool})

const logger = createMethodPerLevelLoggerAdapter(
makeLogger({
logLevel: process.env.LOG_LEVEL,
handleExceptions: false
})
)

const db = new Db({pool})
const storage = new MessageStorage({
db,
diagnostics: new DefaultMessageStorageDiagnosticsAdapter({
logger,
createMessagesDurationMetric: summaryMetric('create_messages_duration_in_ms'),
updateMessagesDurationMetric: summaryMetric('update_messages_duration_in_ms'),
removeMessagesDurationMetric: summaryMetric('remove_messages_duration_in_ms'),
readMessagesDurationMetric: summaryMetric('read_messages_duration_in_ms')
})
})
const cache = new MessageCache({
maxMessageSize: Number(process.env.MESSAGE_CACHE_MAX_MESSAGE_SIZE),
maxSize: Number(process.env.MESSAGE_CACHE_MAX_SIZE),
diagnostics: new DefaultMessageCacheDiagnosticsAdapter({
logger,
messageCountMetric: gaugeMetric('cache_message_count'),
messageSizeMetric: gaugeMetric('cache_message_size')
})
})
const balancerA = new MessageBalancer<{data?: string; retry?: number}>({
partitionGroup: 'A',
lockPartition: true,
storage,
cache,
diagnostics: new DefaultMessageBalancerDiagnosticsAdapter({
logger,
endToEndMessageProcessingDurationMetric: summaryMetric('balancer_a_end_to_end_processing_duration_in_ms'),
centrifugePartitionCountMetric: gaugeMetric('balancer_a_centrifuge_partition_count'),
centrifugeMessageCountMetric: gaugeMetric('balancer_a_centrifuge_message_count')
})
})
const balancerB = new MessageBalancer<{data?: string}>({
partitionGroup: 'B',
partitionSizeLimit: Number(process.env.PARTITION_SIZE_LIMIT),
lockPartition: true,
storage,
cache,
diagnostics: new DefaultMessageBalancerDiagnosticsAdapter({
logger,
endToEndMessageProcessingDurationMetric: summaryMetric('balancer_b_end_to_end_processing_duration_in_ms'),
centrifugePartitionCountMetric: gaugeMetric('balancer_b_centrifuge_partition_count'),
centrifugeMessageCountMetric: gaugeMetric('balancer_b_centrifuge_message_count')
})
})

await balancerA.init()
await balancerB.init()

return [balancerA, balancerA]
}

function summaryMetric(name: string) {
return new promClient.Summary({
name,
help: 'Write it yourself',
percentiles: [0.1, 0.5, 0.9, 0.99],
maxAgeSeconds: 10 * 60,
ageBuckets: 10
})
}

function gaugeMetric(name: string) {
return new promClient.Gauge({
name,
help: 'Write it yourself'
})
}

main()
``