import logger from '@mds-core/mds-logger' ; import redis from 'redis' ; import bluebird from 'bluebird' ; import { Device , VehicleEvent , Telemetry } from '@mds-core/mds-types' ; import { Stream , StreamItem , ReadStreamResult , DEVICE_INDEX_STREAM , DEVICE_RAW_STREAM , ReadStreamOptions , StreamItemID } from './types' ; import { AgencyStreamKafka } from './kafka/agency-stream-kafka' ; import { KafkaStreamConsumer , KafkaStreamProducer } from './kafka' ; import { NatsStreamConsumer } from './nats/stream-consumer' ; import { NatsStreamProducer } from './nats/stream-producer' ; import { AgencyStreamNats } from './nats/agency-stream-nats' ; export { KafkaStreamConsumerOptions , KafkaStreamProducerOptions } from './kafka' ; export { StreamConsumer , StreamProducer } from './stream-interface' ; const { env } = process ; declare module 'redis' { interface RedisClient { dbsizeAsync : ( ) => Promise < number > ; flushdbAsync : ( ) => Promise < 'OK' > ; pingAsync : < TPong extends string = 'PONG' > ( response ? : TPong ) => Promise < TPong > ; xaddAsync : ( ... args : unknown [ ] ) => Promise < string > ; xinfoAsync : < T extends 'STREAM' | 'GROUPS' > ( arg : T , stream : Stream ) => Promise < T extends 'STREAM' ? [ 'length' , number , 'radix-tree-keys' , number , 'radix-tree-nodes' , number , 'groups' , number , 'last-generated-id' , string , 'first-entry' , StreamItem | null , 'last-entry' , StreamItem | null ] : [ 'name' , string , 'consumers' , number , 'pending' , number , 'last-delivered-id' , string ] [ ] > ; xreadAsync : ( ... args : unknown [ ] ) => Promise < ReadStreamResult [ ] > ; xgroupAsync : ( ... args : unknown [ ] ) => Promise < 'OK' > ; xreadgroupAsync : ( ... args : unknown [ ] ) => Promise < ReadStreamResult [ ] > ; } interface Multi { xadd : ( ... args : unknown [ ] ) => string ; execAsync : ( ) => Promise < object [ ] > ; } } bluebird . promisifyAll ( redis . RedisClient . prototype ) ; bluebird . promisifyAll ( redis . Multi . prototype ) ; const { now } = Date ; let cachedClient = null ; const STREAM_MAXLEN = { 'device:index' : 10000 , 'device:raw' : 1000000 } ; async function getClient ( ) { if ( ! cachedClient ) { const { REDIS_HOST , REDIS_PORT } = process . env ; const { host = 'localhost' , port = 6379 } = { host : REDIS_HOST , port : REDIS_PORT } ; if ( ! REDIS_PORT ) { logger . info ( `no redis port found, falling back to ${ port }` ) ; } if ( ! REDIS_HOST ) { logger . info ( `no redis host found, falling back to ${ host }` ) ; } logger . info ( `connecting to redis on ${ host }:${ port }` ) ; cachedClient = redis . createClient ( Number ( port ) , host ) ; cachedClient . on ( 'error' , async ( err ) => { logger . error ( `redis error ${ err }` ) ; } ) ; await cachedClient . dbsizeAsync ( ) . then ( size => logger . info ( `redis has ${ size } keys` ) ) ; } return cachedClient ; } async function initialize ( ) { await AgencyStreamKafka . initialize ( ) ; await AgencyStreamNats . initialize ( ) ; await getClient ( ) ; } async function reset ( ) { const client = await getClient ( ) ; await client . flushdbAsync ( ) . then ( ( ) => logger . info ( 'redis flushed' ) ) ; } async function startup ( ) { await getClient ( ) ; } async function shutdown ( ) { if ( cachedClient ) { await cachedClient . quit ( ) ; cachedClient = null ; } await AgencyStreamKafka . shutdown ( ) ; await AgencyStreamNats . shutdown ( ) ; } async function writeStream ( stream , field , value ) { const client = await getClient ( ) ; return client . xaddAsync ( stream , 'MAXLEN' , '~' , STREAM_MAXLEN [ stream ] , '*' , field , JSON . stringify ( value ) ) ; } async function writeStreamBatch ( stream , field , values ) { const client = await getClient ( ) ; const batch = client . batch ( ) ; values . forEach ( value => batch . xadd ( stream , 'MAXLEN' , '~' , STREAM_MAXLEN [ stream ] , '*' , field , JSON . stringify ( value ) ) ) ; await batch . execAsync ( ) ; } async function writeDevice ( device ) { if ( env . NATS ) { await AgencyStreamNats . writeDevice ( device ) ; } if ( env . KAFKA_HOST ) { await AgencyStreamKafka . writeDevice ( device ) ; } return writeStream ( DEVICE_INDEX_STREAM , 'data' , device ) ; } async function writeEvent ( event ) { if ( env . NATS ) { await AgencyStreamNats . writeEvent ( event ) ; } if ( env . KAFKA_HOST ) { await AgencyStreamKafka . writeEvent ( event ) ; } return writeStream ( DEVICE_RAW_STREAM , 'event' , event ) ; } async function writeTelemetry ( telemetry ) { if ( env . NATS ) { await AgencyStreamNats . writeTelemetry ( telemetry ) ; } if ( env . KAFKA_HOST ) { await AgencyStreamKafka . writeTelemetry ( telemetry ) ; } const start = now ( ) ; await writeStreamBatch ( DEVICE_RAW_STREAM , 'telemetry' , telemetry ) ; const delta = now ( ) - start ; if ( delta > 200 ) { logger . info ( 'stream writeTelemetry' , telemetry . length , 'points in' , delta , 'ms' ) ; } } async function readStream ( stream , id , { count , block } ) { const client = await getClient ( ) ; const results = await client . xreadAsync ( [ ... ( typeof block === 'number' ? [ 'BLOCK' , block ] : [ ] ) , ... ( typeof count === 'number' ? [ 'COUNT' , count ] : [ ] ) , 'STREAMS' , stream , id || '$' ] ) ; if ( results ) { const [ result ] = results ; return result ; } return [ stream , [ ] ] ; } async function createStreamGroup ( stream , group ) { const client = await getClient ( ) ; return client . xgroupAsync ( 'CREATE' , stream , `${ stream }::${ group }` , 0 , 'MKSTREAM' ) ; } async function readStreamGroup ( stream , group , consumer , id , { count , block , noack } ) { const client = await getClient ( ) ; const results = await client . xreadgroupAsync ( 'GROUP' , `${ stream }::${ group }` , consumer , ... [ ... ( typeof block === 'number' ? [ 'BLOCK' , block ] : [ ] ) , ... ( typeof count === 'number' ? [ 'COUNT' , count ] : [ ] ) , ... ( noack ? [ 'NOACK' ] : [ ] ) , 'STREAMS' , stream , id || '>' ] ) ; if ( results ) { const [ result ] = results ; return result ; } return [ stream , [ ] ] ; } async function getStreamInfo ( stream ) { const client = await getClient ( ) ; try { const [ , length , , radixTreeKeys , , radixTreeNodes , , groups , , lastGeneratedId , , firstEntry , , lastEntry ] = await client . xinfoAsync ( 'STREAM' , stream ) ; return { length , radixTreeKeys , radixTreeNodes , groups , lastGeneratedId , firstEntry , lastEntry } ; } catch ( err ) { return null ; } } async function health ( ) { if ( env . SINK ) { return Promise . resolve ( { using : 'cloudevents-emitter' , status : true , stats : { } } ) ; } const client = await getClient ( ) ; const status = await client . pingAsync ( 'connected' ) ; return { using : 'redis' , status } ; } export default { createStreamGroup , getStreamInfo , health , initialize , readStream , readStreamGroup , reset , shutdown , startup , writeDevice , writeEvent , writeStream , writeStreamBatch , writeTelemetry , KafkaStreamConsumer , KafkaStreamProducer , NatsStreamConsumer , NatsStreamProducer } ;