<commit_before>import { delay, promisifyAll } from 'bluebird';
import { events } from '../src/system/common/events';
import { RedisClient } from '../src/system/redis-client/redis-client';
import { Producer, Message, MonitorServer, Consumer } from '../index';
import { config } from './config';
import { ICallback, IConfig, TTimeSeriesRange } from '../types';
import { WebsocketMainStreamWorker } from '../src/monitor-server/workers/websocket-main-stream.worker';
import { QueueManagerFrontend } from '../src/system/queue-manager/queue-manager-frontend';
import { MessageManagerFrontend } from '../src/system/message-manager/message-manager-frontend';
import { MessageManager } from '../src/system/message-manager/message-manager';
import * as supertest from 'supertest';
import { Logger } from '../src/system/common/logger';
import { QueueManager } from '../src/system/queue-manager/queue-manager';
import { WebsocketRateStreamWorker } from '../src/monitor-server/workers/websocket-rate-stream.worker';
import { WebsocketHeartbeatStreamWorker } from '../src/monitor-server/workers/websocket-heartbeat-stream.worker';
import { WebsocketOnlineStreamWorker } from '../src/monitor-server/workers/websocket-online-stream.worker';

type TMonitorServer = ReturnType<typeof MonitorServer>;

type TGetConsumerArgs = {
  queueName?: string;
  cfg?: IConfig;
  consumeMock?: ((msg: Message, cb: ICallback<void>) => void) | null;
};

export interface ISuperTestResponse<TData> extends supertest.Response {
  body: {
    data?: TData;
    error?: {
      code: string;
      message: string;
      details: Record<string, any>;
    };
  };
}

Message.setDefaultOptions(config.message);
const MessageManagerFrontendAsync = promisifyAll(MessageManagerFrontend);
const QueueManagerFrontendAsync = promisifyAll(QueueManagerFrontend);

class TestConsumer extends Consumer {
  // eslint-disable-next-line class-methods-use-this
  consume(message: Message, cb: ICallback<void>) {
    cb(null);
  }
}

const redisClients: RedisClient[] = [];
const consumersList: Consumer[] = [];
const producersList: Producer[] = [];
let monitorServer: TMonitorServer | null = null;
let websocketMainStreamWorker: WebsocketMainStreamWorker | null = null;
let websocketRateStreamWorker: WebsocketRateStreamWorker | null = null;
let websocketHeartbeatStreamWorker: WebsocketHeartbeatStreamWorker | null =
  null;
let websocketOnlineStreamWorker: WebsocketOnlineStreamWorker | null = null;
let messageManager: MessageManager | null = null;
let messageManagerFrontend: MessageManagerFrontend | null = null;
let queueManager: QueueManager | null = null;
let queueManagerFrontend: QueueManagerFrontend | null = null;

export async function startUp(): Promise<void> {
  const redisClient = await getRedisInstance();
  await redisClient.flushallAsync();
}

export async function shutdown(): Promise<void> {
  const p = async (list: (Consumer | Producer)[]) => {
    for (const i of list) {
      if (i.isRunning()) {
        // eslint-disable-next-line no-await-in-loop
        await new Promise((resolve) => {
          i.shutdown(resolve);
        });
      }
    }
  };

  await p(consumersList);
  await p(producersList);
  while (redisClients.length) {
    const redisClient = redisClients.pop();
    if (redisClient) {
      redisClient.end(true);
    }
  }
  if (messageManager) {
    const m = promisifyAll(messageManager);
    await m.quitAsync();
    messageManager = null;
  }
  if (messageManagerFrontend) {
    const m = promisifyAll(messageManagerFrontend);
    await m.quitAsync();<commit_msg>Fix random "Connection is closed" error while testing<commit_after>import { delay, promisifyAll } from 'bluebird';
import { events } from '../src/system/common/events';
import { RedisClient } from '../src/system/redis-client/redis-client';
import { Producer, Message, MonitorServer, Consumer } from '../index';
import { config } from './config';
import { ICallback, IConfig, TTimeSeriesRange } from '../types';
import { WebsocketMainStreamWorker } from '../src/monitor-server/workers/websocket-main-stream.worker';
import { QueueManagerFrontend } from '../src/system/queue-manager/queue-manager-frontend';
import { MessageManagerFrontend } from '../src/system/message-manager/message-manager-frontend';
import { MessageManager } from '../src/system/message-manager/message-manager';
import * as supertest from 'supertest';
import { Logger } from '../src/system/common/logger';
import { QueueManager } from '../src/system/queue-manager/queue-manager';
import { WebsocketRateStreamWorker } from '../src/monitor-server/workers/websocket-rate-stream.worker';
import { WebsocketHeartbeatStreamWorker } from '../src/monitor-server/workers/websocket-heartbeat-stream.worker';
import { WebsocketOnlineStreamWorker } from '../src/monitor-server/workers/websocket-online-stream.worker';

type TMonitorServer = ReturnType<typeof MonitorServer>;

type TGetConsumerArgs = {
  queueName?: string;
  cfg?: IConfig;
  consumeMock?: ((msg: Message, cb: ICallback<void>) => void) | null;
};

export interface ISuperTestResponse<TData> extends supertest.Response {
  body: {
    data?: TData;
    error?: {
      code: string;
      message: string;
      details: Record<string, any>;
    };
  };
}

Message.setDefaultOptions(config.message);
const MessageManagerFrontendAsync = promisifyAll(MessageManagerFrontend);
const QueueManagerFrontendAsync = promisifyAll(QueueManagerFrontend);

class TestConsumer extends Consumer {
  // eslint-disable-next-line class-methods-use-this
  consume(message: Message, cb: ICallback<void>) {
    cb(null);
  }
}

const redisClients: RedisClient[] = [];
const consumersList: Consumer[] = [];
const producersList: Producer[] = [];
let monitorServer: TMonitorServer | null = null;
let websocketMainStreamWorker: WebsocketMainStreamWorker | null = null;
let websocketRateStreamWorker: WebsocketRateStreamWorker | null = null;
let websocketHeartbeatStreamWorker: WebsocketHeartbeatStreamWorker | null =
  null;
let websocketOnlineStreamWorker: WebsocketOnlineStreamWorker | null = null;
let messageManager: MessageManager | null = null;
let messageManagerFrontend: MessageManagerFrontend | null = null;
let queueManager: QueueManager | null = null;
let queueManagerFrontend: QueueManagerFrontend | null = null;

export async function startUp(): Promise<void> {
  const redisClient = await getRedisInstance();
  await redisClient.flushallAsync();
}

export async function shutdown(): Promise<void> {
  const p = async (list: (Consumer | Producer)[]) => {
    for (const i of list) {
      if (i.isRunning()) {
        // eslint-disable-next-line no-await-in-loop
        await new Promise((resolve) => {
          i.shutdown(resolve);
        });
      }
    }
  };

  await p(consumersList);
  await p(producersList);
  while (redisClients.length) {
    const redisClient = redisClients.pop();
    if (redisClient) {
      await promisifyAll(redisClient).haltAsync();
    }
  }
  if (messageManager) {
    const m = promisifyAll(messageManager);
    await m.quitAsync();
    messageManager = null;
  }
  if (messageManagerFrontend) {
    const m = promisifyAll(messageManagerFrontend);
    await m.quitAsync();
{'fraction_non_alphanumeric': 0.12659988870339456, 'fraction_numerical': 0.0, 'mean_word_length': 4.542791056283732, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1949395', 'n_tokens_mistral': 2115, 'n_tokens_neox': 2097, 'n_words': 485}