feat(messaging): always emit outgoing events
MODIFY
packages/@integration/src/cqrs/domain/Offer.event.ts
packages/@integration/src/cqrs/domain/Offer.event.ts
@@ -3,8 +3,8 @@ import { createEvent, EventsUnion } from '@marblejs/core';
export enum OfferEventType {
OFFER_CREATED = 'OFFER_CREATED',
OFFER_UPDATED = 'OFFER_UPDATED',
- OFFER_DOCUMENT_SAVED = 'OFFER_DOCUMENT_CREATED',
- OFFER_DOCUMENT_CREATED = 'OFFER_DOCUMENT_SAVED',
+ OFFER_DOCUMENT_SAVED = 'OFFER_DOCUMENT_SAVED',
+ OFFER_DOCUMENT_CREATED = 'OFFER_DOCUMENT_CREATED',
};
export const OfferEvent = {

MODIFY
packages/@integration/src/cqrs/effects/eventbus.effects.ts
packages/@integration/src/cqrs/effects/eventbus.effects.ts
import { of } from 'rxjs';
import { map, tap, delay } from 'rxjs/operators';
import { act, matchEvent, useContext, LoggerToken, LoggerTag } from '@marblejs/core';
-import { MsgEffect, EVENT_BUS_CHANNEL, reply } from '@marblejs/messaging';
+import { MsgEffect } from '@marblejs/messaging';
import { OfferCommand } from '../domain/Offer.command';
import { OfferEvent } from '../domain/Offer.event';
@@ -16,9 +16,7 @@ export const generateOfferDocument$: MsgEffect = (event$, ctx) => {
map(event => event.payload.offerId),
tap(logger({ tag, type: 'generateOfferDocument$', message: 'Generating offer document...'})),
delay(5 * 1000),
- map(offerId => reply(EVENT_BUS_CHANNEL)(
- OfferEvent.offerDocumentCreated(offerId),
- )),
+ map(offerId => OfferEvent.offerDocumentCreated(offerId)),
)),
);
}

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
@@ -2,7 +2,7 @@ import { Event, matchEvent, use, combineEffects, act, EventError } from '@marble
import { wait, NamedError } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { throwError, of } from 'rxjs';
-import { delay, map, tap } from 'rxjs/operators';
+import { delay, map, tap, ignoreElements } from 'rxjs/operators';
import { MsgEffect } from '../effects/messaging.effects.interface';
import { runEventBus, runEventBusClient } from '../util/messaging.test.util';
import { reply } from '../effects/messaging.effects.helper';
@@ -119,6 +119,7 @@ describe('#eventBus', () => {
await eventBusClient.close();
done();
}),
+ ignoreElements(),
);
const eventBus = await runEventBus(foo$);

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -63,17 +63,14 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const send = (connection: TransportLayerConnection) => (event: Event): void => {
const { metadata, type, payload, error } = event;
+ const { replyTo, correlationId, raw } = metadata ?? {};
- if (metadata && metadata.replyTo) {
- const { replyTo, correlationId, raw } = metadata;
-
- connection.emitMessage(replyTo, {
+ connection.emitMessage(replyTo ?? '', {
data: msgTransformer.encode({ type, payload, error }),
correlationId,
replyTo,
raw,
});
- }
};
return connection => {

MODIFY
packages/messaging/src/transport/strategies/local.strategy.ts
packages/messaging/src/transport/strategies/local.strategy.ts
@@ -36,7 +36,7 @@ class LocalStrategyConnection implements TransportLayerConnection {
}
emitMessage = async (channel: string, message: TransportMessage<Buffer>) => {
- if (channel !== this.opts.channel) {
+ if (channel && channel !== this.opts.channel) {
this.rpcSubject$.next(message);
} else {
this.msgSubject$.next(message);

feat(messaging): always emit outgoing events