Method run

packages/server/src/commands/worker.ts:22–66

()

    scheduleWorkerId: string

    async run(): Promise<void> {
        logger.info('Starting Flowise Worker...')

        const { appDataSource, telemetry, componentNodes, cachePool, abortControllerPool, usageCacheManager, identityManager } =
            await this.prepareData()

        const queueManager = QueueManager.getInstance()
        queueManager.setupAllQueues({
            componentNodes,
            telemetry,
            cachePool,
            appDataSource,
            abortControllerPool,
            usageCacheManager,
            identityManager
        })

        /** Prediction */
        const predictionQueue = queueManager.getQueue('prediction')
        const predictionWorker = predictionQueue.createWorker()
        this.predictionWorkerId = predictionWorker.id
        logger.info(`Prediction Worker ${this.predictionWorkerId} created`)

        const predictionQueueName = predictionQueue.getQueueName()
        const queueEvents = new QueueEvents(predictionQueueName, { connection: queueManager.getConnection() })

        queueEvents.on<CustomListener>('abort', async ({ id }: { id: string }) => {
            abortControllerPool.abort(id)
        })

        /** Upsertion */
        const upsertionQueue = queueManager.getQueue('upsert')
        const upsertionWorker = upsertionQueue.createWorker()
        this.upsertionWorkerId = upsertionWorker.id
        logger.info(`Upsertion Worker ${this.upsertionWorkerId} created`)

        /** Schedule */
        const scheduleQueue = queueManager.getQueue('schedule')
        const scheduleWorker = scheduleQueue.createWorker()
        this.scheduleWorkerId = scheduleWorker.id
        logger.info(`Schedule Worker ${this.scheduleWorkerId} created`)

        // Keep the process running
        process.stdin.resume()
    }

    async prepareData() {
        // Init database