Skip to content

feat!: rework commands to support all proper cqrs commands #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mhochsto merged 4 commits into from
Jun 25, 2025
Merged

feat!: rework commands to support all proper cqrs commands #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
617e739
fix: improve handling of commands in command-service
0x7061 Jun 25, 2025
41492cb
feat: add commandId promise resolve mapping
0x7061 Jun 25, 2025
912f72b
chore: rename
0x7061 Jun 25, 2025
a8caeb3
chore: update deps
0x7061 Jun 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 18 additions & 18 deletions package-lock.json
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion src/broker/broker.constants.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,6 @@ export const EVENT_QUERY_PAGED_REQUEST = 'event.query.paged.request';
export const EVENT_QUERY_PAGED_RESPONSE = 'event.query.paged.response';
export const EVENT_CQRS_REQUEST = 'event.cqrs.request';
export const EVENT_CQRS_RESPONSE = 'event.cqrs.response';
export const EVENT_ERROR_RESPONSE = 'event.error.response';
export const EVENT_RB_REQUEST = 'event.rb.request';
export const EVENT_RB_RESPONSE = 'event.rb.response';
export const EVENT_ERROR_RESPONSE = 'event.error.response';
24 changes: 15 additions & 9 deletions src/broker/mqtt/mqtt-broker.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,13 +225,18 @@ export class MqttBrokerService implements Broker {
/**
* Publishes a CQRS command to the broker.
*
* @param {CommandRequest} payload
* @param {CommandRequest} request
*/
@OnEvent(EVENT_CQRS_REQUEST, { async: true })
public async publishCQRSCommand(payload: CommandRequest) {
public async publishCQRSCommand(request: CommandRequest) {
try {
const topic = `${BROKER_TOPIC_PREFIXES.CMD}/${payload.type}`;
const dataStr = JSON.stringify(payload.data);
const topic = `${BROKER_TOPIC_PREFIXES.CMD}/${request.type}`;
const data = {
token: this.options.token,
commandId: request.commandId,
...request.data,
};
const dataStr = JSON.stringify(data);

await this.mqttService.publish(topic, dataStr);

Expand All @@ -244,19 +249,20 @@ export class MqttBrokerService implements Broker {
}

/**
* Publishes a RB command to the broker.
* Publishes an RB command to the broker.
*
* @param {CommandRequest} payload
* @param {CommandRequest} request
*/
@OnEvent(EVENT_RB_REQUEST, { async: true })
public async publishRelaisBoardCommand(payload: CommandRequest) {
public async publishRelaisBoardCommand(request: CommandRequest) {
try {
const topic = `${BROKER_TOPIC_PREFIXES.RB}/${(payload.data as CommandDataRelaisBoard).rb}/do`;
const topic = `${BROKER_TOPIC_PREFIXES.RB}/${(request.data as CommandDataRelaisBoard).rb}/do`;
const dataStr = JSON.stringify(
(payload.data as CommandDataRelaisBoard).config,
(request.data as CommandDataRelaisBoard).config,
);

await this.mqttService.publish(topic, dataStr);
this.eventEmitter.emit(EVENT_RB_RESPONSE, null);

this.logger.verbose(
`Message published\n\ttopic: ${topic}\n\tdata: ${dataStr}`,
Expand Down
14 changes: 8 additions & 6 deletions src/client/client.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { MqttBrokerService } from '../broker/mqtt/mqtt-broker.service';
import { CommandDataRelaisBoardConfig } from '../command/command';
import { CommandCQRS, CommandDataRelaisBoardConfig } from '../command/command';
import { CommandService } from '../command/command.service';
import { HashMap } from '../common/interface';
import { QueryService } from '../query/query.service';
import {
Query,
Expand Down Expand Up @@ -57,10 +58,11 @@ export class ClientService {
return this.queryService.queryPaged(q);
}

async commandRelaisBoard(rb: string, config: CommandDataRelaisBoardConfig) {
return this.commandService.commandRelaisBoard({
rb,
config,
});
async commandCQRS(type: CommandCQRS, data: HashMap<any>) {
return this.commandService.cqrs(type, data);
}

async commandRB(rb: string, config: CommandDataRelaisBoardConfig) {
return this.commandService.rb('ConfigureRelaisBoard', { rb, config });
}
}
180 changes: 10 additions & 170 deletions src/command/command.service.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,19 +35,21 @@ describe('CommandService', () => {
await moduleRef.close();
});

describe('assignAuthorizationProfileToMedium()', () => {
describe('cqrs()', () => {
// Arguments
const profileId = '7950ed5a-ddc1-4033-ac11-3487dac8cf3b';
const mediumId = '6087c4c8-1125-4af8-8d1d-d8d287406f98';
const commandId = '968b8708-9504-4552-8fff-71ceaabafdb9';
const type = 'RemoteDisengage';
const data = {
installationPointId: 'a7691e53-4c35-44f1-b3e1-cc77f48be4fc',
commandId,
};

it('should throw on no connection', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => false);

await expect(
commandService.assignAuthorizationProfileToMedium(profileId, mediumId),
).rejects.toThrow();
await expect(commandService.cqrs(type, data)).rejects.toThrow();
});

it('should emit EVENT_CQRS_REQUEST', async () => {
Expand All @@ -60,172 +62,13 @@ describe('CommandService', () => {
resolve(true);
});
});
void commandService.assignAuthorizationProfileToMedium(
profileId,
mediumId,
);
void commandService.cqrs(type, data);
const result = await asyncEvent;

expect(result).toBeTruthy();
});

it('should return command response', async () => {
const commandId = '968b8708-9504-4552-8fff-71ceaabafdb9';

jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);

eventEmitter.on(EVENT_CQRS_REQUEST, () => {
eventEmitter.emit(EVENT_CQRS_RESPONSE, {
commandId: commandId,
});
});
const result = await commandService.assignAuthorizationProfileToMedium(
profileId,
mediumId,
);

expect(result?.commandId).toBe(commandId);
});
});

describe('assignPersonToMedium()', () => {
// Arguments
const personId = '7950ed5a-ddc1-4033-ac11-3487dac8cf3b';
const mediumId = '6087c4c8-1125-4af8-8d1d-d8d287406f98';

it('should throw on no connection', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => false);

await expect(
commandService.assignPersonToMedium(personId, mediumId),
).rejects.toThrow();
});

it('should emit EVENT_CQRS_REQUEST', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);

const asyncEvent = new Promise<boolean>((resolve) => {
eventEmitter.on(EVENT_CQRS_REQUEST, () => {
resolve(true);
});
});
void commandService.assignPersonToMedium(personId, mediumId);
const result = await asyncEvent;

expect(result).toBeTruthy();
});

it('should return command response', async () => {
const commandId = '968b8708-9504-4552-8fff-71ceaabafdb9';

jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);

eventEmitter.on(EVENT_CQRS_REQUEST, () => {
eventEmitter.emit(EVENT_CQRS_RESPONSE, {
commandId: commandId,
});
});
const result = await commandService.assignPersonToMedium(
personId,
mediumId,
);

expect(result?.commandId).toBe(commandId);
});
});

describe('remoteDisengage()', () => {
// Arguments
const installationPointId = '7950ed5a-ddc1-4033-ac11-3487dac8cf3b';

it('should throw on no connection', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => false);

await expect(
commandService.remoteDisengage(installationPointId, true),
).rejects.toThrow();
});

it('should emit EVENT_CQRS_REQUEST', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);

const asyncEvent = new Promise<boolean>((resolve) => {
eventEmitter.on(EVENT_CQRS_REQUEST, () => {
resolve(true);
});
});
void commandService.remoteDisengage(installationPointId, true);
const result = await asyncEvent;

expect(result).toBeTruthy();
});

it('should return command response', async () => {
const commandId = '968b8708-9504-4552-8fff-71ceaabafdb9';

jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);

eventEmitter.on(EVENT_CQRS_REQUEST, () => {
eventEmitter.emit(EVENT_CQRS_RESPONSE, {
commandId: commandId,
});
});
const result = await commandService.remoteDisengage(
installationPointId,
true,
);

expect(result?.commandId).toBe(commandId);
});
});

describe('remoteDisengagePermanent()', () => {
// Arguments
const installationPointId = '7950ed5a-ddc1-4033-ac11-3487dac8cf3b';

it('should throw on no connection', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => false);

await expect(
commandService.remoteDisengagePermanent(installationPointId, true),
).rejects.toThrow();
});

it('should emit EVENT_CQRS_REQUEST', async () => {
jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);

const asyncEvent = new Promise<boolean>((resolve) => {
eventEmitter.on(EVENT_CQRS_REQUEST, () => {
resolve(true);
});
});
void commandService.remoteDisengagePermanent(installationPointId, true);
const result = await asyncEvent;

expect(result).toBeTruthy();
});

it('should return command response', async () => {
const commandId = '968b8708-9504-4552-8fff-71ceaabafdb9';

jest
.spyOn(mqttBrokerService, 'isConnected')
.mockImplementation(() => true);
Expand All @@ -235,10 +78,7 @@ describe('CommandService', () => {
commandId: commandId,
});
});
const result = await commandService.remoteDisengagePermanent(
installationPointId,
true,
);
const result = await commandService.cqrs(type, data);

expect(result?.commandId).toBe(commandId);
});
Expand Down
Loading
Loading