import { DEFAULT_REDIS_OPTIONS , RedisOptions } from "./options" ; import { MessageBroker } from "../message_broker" ; import { TaskMessage } from "../messages" ; import { isNullOrUndefined } from "../utility" ; import * as IoRedis from "ioredis" ; /**
 * RedisBroker implements MessageBroker using the Redis in-memory database.
 * Messages are not durable and will not survive a broker restart.
 */ export class RedisBroker implements MessageBroker { private readonly options : RedisOptions ; private readonly connection : IoRedis . Redis ; param options returns public constructor ( options ? ) { this . options = ( ( ) => { if ( isNullOrUndefined ( options ) ) { return DEFAULT_REDIS_OPTIONS ; } return options ; } ) ( ) ; this . connection = this . options . createClient ( { keyPrefix : "" } ) ; } returns public async disconnect ( ) { this . connection . disconnect ( ) ; } returns public async end ( ) { await this . connection . quit ( ) ; this . connection . disconnect ( ) ; } param message returns public async publish ( message ) { const toPublish = JSON . stringify ( message ) ; const queue = "celery" ; return this . connection . lpush ( queue , toPublish ) ; } }