mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-03-05 20:50:17 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge 84f96096ce into d2dd1796a8

Browse source
This commit is contained in:
JupiterWalker 2025-03-06 05:40:43 +01:00 committed by GitHub
parent d2dd1796a8 84f96096ce
commit 530d6666ec
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 24 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
packages/cli/src/eventbus/message-event-bus/message-event-bus.ts
View file

@ -99,7 +99,7 @@ export class MessageEventBus extends EventEmitter {
for (const destinationData of savedEventDestinations) { for (const destinationData of savedEventDestinations) {
try { try {
const destination = messageEventBusDestinationFromDb(this, destinationData); const destination = messageEventBusDestinationFromDb(this, destinationData);
if (destination) { if (destination && destination.enabled) {
await this.addDestination(destination, false); await this.addDestination(destination, false);
} }
} catch (error) { } catch (error) {
@ -206,21 +206,36 @@ export class MessageEventBus extends EventEmitter {
} }
async addDestination(destination: MessageEventBusDestination, notifyWorkers: boolean = true) { async addDestination(destination: MessageEventBusDestination, notifyWorkers: boolean = true) {
await this.removeDestination(destination.getId(), false);
this.destinations[destination.getId()] = destination;
this.destinations[destination.getId()].startListening();
if (notifyWorkers) { if (notifyWorkers) {
await this.removeDestination(destination.getId(), false);
await destination.saveToDb();
void this.publisher.publishCommand({ command: 'restart-event-bus' }); void this.publisher.publishCommand({ command: 'restart-event-bus' });
} }
this.destinations[destination.getId()] = destination;
this.destinations[destination.getId()].startListening();
return destination; return destination;
} }
async findDestination(id?: string): Promise<MessageEventBusDestinationOptions[]> { async findDestination(id?: string): Promise<MessageEventBusDestinationOptions[]> {
let result: MessageEventBusDestinationOptions[]; let result: MessageEventBusDestinationOptions[] = [];
if (id && Object.keys(this.destinations).includes(id)) { if (id) {
result = [this.destinations[id].serialize()]; const savedEventDestination = await this.eventDestinationsRepository.findOne({ where: { id } });
} else { if (savedEventDestination) {
result = Object.keys(this.destinations).map((e) => this.destinations[e].serialize()); const destination = messageEventBusDestinationFromDb(this, savedEventDestination);
if (destination) {
result = [destination.serialize()];
}
}
} else {
const savedEventDestinations = await this.eventDestinationsRepository.find();
if (savedEventDestinations.length > 0) {
for (const destinationData of savedEventDestinations) {
const destination = messageEventBusDestinationFromDb(this, destinationData)
if (destination) {
result.push(destination.serialize());
}
}
}
} }
return result.sort((a, b) => (a.__type ?? '').localeCompare(b.__type ?? '')); return result.sort((a, b) => (a.__type ?? '').localeCompare(b.__type ?? ''));
} }