build: rxjs upgrade - got rid of `fromEvent` deprecation warning
MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
import { pipe } from 'fp-ts/lib/function';
-import { Subject, fromEvent, merge, from, firstValueFrom } from 'rxjs';
+import { Subject, fromEvent, merge, from, firstValueFrom, Observable } from 'rxjs';
import { map, filter, take, mapTo, first, mergeMap, share, tap } from 'rxjs/operators';
import { Channel, ConsumeMessage, Replies } from 'amqplib';
import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
@@ -43,26 +43,28 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
get error$() {
return merge(
- fromEvent<Error>(this.channelWrapper, 'error'),
+ fromEvent(this.channelWrapper, 'error') as Observable<Error>,
this.errorSubject$.asObservable(),
);
}
get status$() {
- const connect$ = fromEvent(this.connectionManager, 'connect')
- .pipe(mapTo(AmqpConnectionStatus.CONNECTED));
+ const connect$ = pipe(
+ fromEvent(this.connectionManager, 'connect'),
+ mapTo(AmqpConnectionStatus.CONNECTED));
- const connectChannel$ = fromEvent(this.channelWrapper, 'connect')
- .pipe(mapTo(AmqpConnectionStatus.CHANNEL_CONNECTED));
+ const connectChannel$ = pipe(
+ fromEvent(this.channelWrapper, 'connect'),
+ mapTo(AmqpConnectionStatus.CHANNEL_CONNECTED));
- const diconnect$ = fromEvent<{ err?: Error }>(this.connectionManager, 'disconnect')
- .pipe(
+ const diconnect$ = pipe(
+ fromEvent(this.connectionManager, 'disconnect') as Observable<{ err?: Error }>,
tap(({ err }) => err && this.errorSubject$.next(err)),
- mapTo(AmqpConnectionStatus.CONNECTION_LOST),
- );
+ mapTo(AmqpConnectionStatus.CONNECTION_LOST));
- const diconnectChannel$ = fromEvent(this.channelWrapper, 'close')
- .pipe(mapTo(AmqpConnectionStatus.CHANNEL_CONNECTION_LOST));
+ const diconnectChannel$ = pipe(
+ fromEvent(this.channelWrapper, 'close'),
+ mapTo(AmqpConnectionStatus.CHANNEL_CONNECTION_LOST));
return merge(
connect$,
@@ -74,7 +76,8 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
}
get message$() {
- return this.msgSubject$.asObservable().pipe(
+ return pipe(
+ this.msgSubject$.asObservable(),
share(),
map(raw => ({
data: raw.content,
@@ -101,9 +104,10 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
const replyToSubject = new Subject<string>();
const resSubject$ = new Subject<{ msg: ConsumeMessage; tag: string }>();
- replyToSubject
- .pipe(first())
- .subscribe(async replyTo => {
+ pipe(
+ replyToSubject,
+ first()
+ ).subscribe(async replyTo => {
await this.channelWrapper.sendToQueue(queue, data, {
correlationId,
replyTo,
@@ -135,7 +139,8 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
await this.channelWrapper.addSetup(modifyChannelSetup);
- return firstValueFrom(resSubject$.asObservable().pipe(
+ return firstValueFrom(pipe(
+ resSubject$.asObservable(),
filter(raw => raw.msg.properties.correlationId === correlationId),
take(1),
mergeMap(raw =>

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
import { createUuid } from '@marblejs/core/dist/+internal/utils';
-import { Subject, fromEvent, merge, firstValueFrom } from 'rxjs';
+import { Subject, fromEvent, merge, firstValueFrom, Observable } from 'rxjs';
import { map, mapTo, take, tap, share, filter } from 'rxjs/operators';
+import { pipe } from 'fp-ts/lib/function';
import { RedisClient, ClientOpts } from 'redis';
import { TransportLayer, TransportLayerConnection, TransportMessage, Transport, DEFAULT_TIMEOUT } from '../transport.interface';
import { throwUnsupportedError } from '../transport.error';
@@ -47,19 +48,24 @@ class RedisStrategyConnection implements TransportLayerConnection<Transport.REDI
}
get status$() {
- const ready$ = merge(fromEvent(this.subscriber, 'ready'))
- .pipe(mapTo(RedisConnectionStatus.READY));
+ const ready$ = pipe(
+ merge(fromEvent(this.subscriber, 'ready')),
+ mapTo(RedisConnectionStatus.READY));
- const connect$ = merge(fromEvent(this.subscriber, 'connect'))
- .pipe(mapTo(RedisConnectionStatus.CONNECT));
+ const connect$ = pipe(
+ merge(fromEvent(this.subscriber, 'connect')),
+ mapTo(RedisConnectionStatus.CONNECT));
- const reconnecting$ = merge(fromEvent(this.subscriber, 'reconnecting'))
- .pipe(mapTo(RedisConnectionStatus.RECONNECTING));
+ const reconnecting$ = pipe(
+ merge(fromEvent(this.subscriber, 'reconnecting')),
+ mapTo(RedisConnectionStatus.RECONNECTING));
- const end$ = merge(fromEvent(this.subscriber, 'end'))
- .pipe(mapTo(RedisConnectionStatus.END));
+ const end$ = pipe(
+ merge(fromEvent(this.subscriber, 'end')),
+ mapTo(RedisConnectionStatus.END));
- return merge(ready$, connect$, reconnecting$, end$, this.statusSubject$.asObservable()).pipe(
+ return pipe(
+ merge(ready$, connect$, reconnecting$, end$, this.statusSubject$.asObservable()),
share(),
);
}
@@ -69,17 +75,19 @@ class RedisStrategyConnection implements TransportLayerConnection<Transport.REDI
}
get error$() {
- return merge(
- fromEvent<Error>(this.publisher, 'error'),
- fromEvent<Error>(this.subscriber, 'error'),
- fromEvent<Error>(this.rpcSubscriber, 'error'),
- ).pipe(
+ return pipe(
+ merge(
+ fromEvent(this.publisher, 'error') as Observable<Error>,
+ fromEvent(this.subscriber, 'error') as Observable<Error>,
+ fromEvent(this.rpcSubscriber, 'error') as Observable<Error>,
+ ),
share(),
);
}
get message$() {
- return this.consumerSubject.asObservable().pipe(
+ return pipe(
+ this.consumerSubject.asObservable(),
map(msg => RedisHelper.decodeMessage(msg.content)),
);
}
@@ -99,7 +107,8 @@ class RedisStrategyConnection implements TransportLayerConnection<Transport.REDI
await RedisHelper.subscribeChannel(this.rpcSubscriber)(replyChannel);
await this.emitMessage(channel, message);
- return firstValueFrom(this.producerSubject.asObservable().pipe(
+ return firstValueFrom(pipe(
+ this.producerSubject.asObservable(),
filter(msg => msg.channel === replyChannel),
take(1),
tap(() => RedisHelper.unsubscribeChannel(this.rpcSubscriber)(replyChannel)),

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
@@ -52,8 +52,8 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
const errorSubject = new Subject<Error>();
const eventSubject = new Subject<Event>();
const ctx = createEffectContext({ ask, client });
- const close$ = fromEvent(client, 'close');
- const message$ = fromEvent<{ data: any }>(client, 'message');
+ const close$ = fromEvent(client, 'close') as Observable<unknown>;
+ const message$ = fromEvent(client, 'message') as Observable<{ data: any }>;
const applyMetadata = (event: Event): Event => ({
...event,

build: rxjs upgrade - got rid of `fromEvent` deprecation warning