10909848788

Committed 17 Sep 2024 07:35PM UTC coverage: 9.367% (-4.5%) from 13.825%

Build # 10909848788

Build Type

Pull #9226

github

Committed by

web-flow

Commit Message

Merge f3ffda16d into e48016d6c

Pull Request Pull Request #9226: Multi-Adapter Ports

Run Details

2138 of 20652 branches covered (10.35%)

Branch coverage included in aggregate %.

5 of 73 new or added lines in 12 files covered. (6.85%)

1516 existing lines in 179 files now uncovered.

2658 of 30549 relevant lines covered (8.7%)

2.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/packages/commonwealth/server/workers/messageRelayer/messageRelayer.ts

import {
  getRabbitMQConfig,
  RabbitMQAdapter,
  RascalConfigServices,
  ServiceKey,
  startHealthCheckLoop,
} from '@hicommonwealth/adapters';
import { broker, logger } from '@hicommonwealth/core';
import { config } from '../../config';
import { setupListener } from './pgListener';
import { incrementNumUnrelayedEvents, relayForever } from './relayForever';

const log = logger(import.meta);

let isServiceHealthy = false;

startHealthCheckLoop({
  enabled: import.meta.url.endsWith(process.argv[1]),
  service: ServiceKey.MessageRelayer,
  // eslint-disable-next-line @typescript-eslint/require-await
  checkFn: async () => {
    if (!isServiceHealthy) {
      throw new Error('service not healthy');
    }
  },
});

export async function startMessageRelayer(maxRelayIterations?: number) {
  const { models } = await import('@hicommonwealth/model');

  try {
    const rmqAdapter = new RabbitMQAdapter(
      getRabbitMQConfig(
        config.BROKER.RABBITMQ_URI,
        RascalConfigServices.CommonwealthService,
      ),
    );
    await rmqAdapter.init();
    broker({
      adapter: rmqAdapter,
    });
  } catch (e) {
    log.error(
      'Rascal consumer setup failed. Please check the Rascal configuration',
    );
    throw e;
  }

  const count = await models.Outbox.count({
    where: {
      relayed: false,
    },
  });
  incrementNumUnrelayedEvents(count);

  const pgClient = await setupListener();

  isServiceHealthy = true;
  relayForever(maxRelayIterations).catch((err) => {
    log.fatal(
      'Unknown error fatal requires immediate attention. Restart REQUIRED!',
      err,
    );
  });

  return pgClient;
}

if (import.meta.url.endsWith(process.argv[1])) {
  startMessageRelayer().catch((err) => {
    log.fatal(
      'Unknown error fatal requires immediate attention. Restart REQUIRED!',
      err,
    );
    // if we process.exit(1) here we start an infinite loop of retrying to publish events
  });
}

1	import {
2	getRabbitMQConfig,
3	RabbitMQAdapter,
4	RascalConfigServices,
5	ServiceKey,
6	startHealthCheckLoop,
7	} from '@hicommonwealth/adapters';
8	import { broker, logger } from '@hicommonwealth/core';
9	import { config } from '../../config';
10	import { setupListener } from './pgListener';
11	import { incrementNumUnrelayedEvents, relayForever } from './relayForever';
12
UNCOV 13	const log = logger(import.meta);	×
14
UNCOV 15	let isServiceHealthy = false;	×
16
UNCOV 17	startHealthCheckLoop({	×
18	enabled: import.meta.url.endsWith(process.argv[1]),
19	service: ServiceKey.MessageRelayer,
20	// eslint-disable-next-line @typescript-eslint/require-await
21	checkFn: async () => {
22	if (!isServiceHealthy) {	×
23	throw new Error('service not healthy');	×
24	}
25	},
26	});
27
28	export async function startMessageRelayer(maxRelayIterations?: number) {
UNCOV 29	const { models } = await import('@hicommonwealth/model');	×
30
UNCOV 31	try {	×
UNCOV 32	const rmqAdapter = new RabbitMQAdapter(	×
33	getRabbitMQConfig(
34	config.BROKER.RABBITMQ_URI,
35	RascalConfigServices.CommonwealthService,
36	),
37	);
UNCOV 38	await rmqAdapter.init();	×
NEW 39	broker({	×
40	adapter: rmqAdapter,
41	});
42	} catch (e) {
43	log.error(	×
44	'Rascal consumer setup failed. Please check the Rascal configuration',
45	);
46	throw e;	×
47	}
48
UNCOV 49	const count = await models.Outbox.count({	×
50	where: {
51	relayed: false,
52	},
53	});
UNCOV 54	incrementNumUnrelayedEvents(count);	×
55
UNCOV 56	const pgClient = await setupListener();	×
57
UNCOV 58	isServiceHealthy = true;	×
UNCOV 59	relayForever(maxRelayIterations).catch((err) => {	×
60	log.fatal(	×
61	'Unknown error fatal requires immediate attention. Restart REQUIRED!',
62	err,
63	);
64	});
65
UNCOV 66	return pgClient;	×
67	}
68
UNCOV 69	if (import.meta.url.endsWith(process.argv[1])) {	×
70	startMessageRelayer().catch((err) => {	×
71	log.fatal(	×
72	'Unknown error fatal requires immediate attention. Restart REQUIRED!',
73	err,
74	);
75	// if we process.exit(1) here we start an infinite loop of retrying to publish events
76	});
77	}

hicommonwealth / commonwealth / 10909848788

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous