fix(messaging): message acknowledgement
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
@@ -11,7 +11,7 @@ import {
HttpServerEffect,
use,
} from '@marblejs/core';
-import { merge, of, forkJoin } from 'rxjs';
+import { merge, of, forkJoin, EMPTY } from 'rxjs';
import { tap, map, mergeMap } from 'rxjs/operators';
import { requestValidator$, t } from '@marblejs/middleware-io';
@@ -32,6 +32,17 @@ const rootValiadtor$ = requestValidator$({
}),
});
+const test$ = r.pipe(
+ r.matchPath('/test'),
+ r.matchType('GET'),
+ r.useEffect((req$, _, { ask }) => req$.pipe(
+ mergeMap(() => ask(ClientToken)
+ .map(client => client.emit({ type: 'TEST' }))
+ .getOrElse(EMPTY)),
+ map(() => ({ body: 'OK' })),
+ )),
+);
+
const fib$ = r.pipe(
r.matchPath('/fib/:number'),
r.matchType('GET'),
@@ -62,7 +73,7 @@ export const server = createServer({
port: Number(process.env.PORT) || 1337,
httpListener: httpListener({
middlewares: [logger$()],
- effects: [fib$],
+ effects: [test$, fib$],
}),
dependencies: [
bindTo(ClientToken)(client.run),

MODIFY
packages/messaging/src/client/messaging.client.spec.ts
packages/messaging/src/client/messaging.client.spec.ts
@@ -29,7 +29,6 @@ describe('#messagingClient', () => {
server.consumeMessage().subscribe(async msg => {
expect(msg.data.toString()).toEqual(JSON.stringify({ test: 'test' }));
- server.ack(msg);
await server.close();
done();
});
@@ -48,7 +47,6 @@ describe('#messagingClient', () => {
const response = createMessage(count, msg.correlationId);
await server.emitMessage(msg.replyTo || '', response);
- server.ack(msg);
await server.close();
});

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -65,7 +65,6 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
raw: msg.raw,
});
}
- conn.ack(msg);
}
const onSubscribeEffectsError = (errorSubject: Subject<Error>) => (error: Error) => {

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -25,6 +25,7 @@ class AmqpStrategyConnection implements TransportLayerConnection {
this.channel.consume(
this.options.queue,
msg => msg && msgSubject$.next(msg),
+ { noAck: true },
);
return msgSubject$.asObservable().pipe(
@@ -78,8 +79,6 @@ class AmqpStrategyConnection implements TransportLayerConnection {
}));
};
- ack = (msg: TransportMessage) => this.channel.ack(msg.raw);
-
getChannel = () => this.options.queue;
close = async () => {

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -18,7 +18,6 @@ export interface TransportLayerConnection {
sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
emitMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
consumeMessage: () => Observable<TransportMessage<Buffer>>;
- ack: (msg: TransportMessage<any>) => void;
close: () => Promise<any>;
getChannel(): string;
error$: Observable<Error>;

fix(messaging): message acknowledgement