refactor: firstValueFrom -> lastValueFrom
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, Observable } from 'rxjs';
+import { Subject, fromEvent, merge, from, lastValueFrom, 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';
@@ -139,7 +139,7 @@ class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP>
await this.channelWrapper.addSetup(modifyChannelSetup);
- return firstValueFrom(pipe(
+ return lastValueFrom(pipe(
resSubject$.asObservable(),
filter(raw => raw.msg.properties.correlationId === correlationId),
take(1),

MODIFY
packages/messaging/src/transport/strategies/local.strategy.ts
packages/messaging/src/transport/strategies/local.strategy.ts
-import { Subject, EMPTY, firstValueFrom } from 'rxjs';
+import { Subject, EMPTY, lastValueFrom } from 'rxjs';
import { share, take, filter } from 'rxjs/operators';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { TransportLayer, TransportLayerConnection, TransportMessage, Transport, DEFAULT_TIMEOUT, TransportLayerConfig } from '../transport.interface';
@@ -66,7 +66,7 @@ class LocalStrategyConnection implements TransportLayerConnection<Transport.LOCA
setImmediate(() => this.msgSubject$.next(message));
- return firstValueFrom(this.rpcSubject$
+ return lastValueFrom(this.rpcSubject$
.asObservable()
.pipe(filter(msg => msg.replyTo === replyChannel && msg.correlationId === correlationId))
.pipe(take(1)));

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, Observable } from 'rxjs';
+import { Subject, fromEvent, merge, lastValueFrom, 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';
@@ -107,7 +107,7 @@ class RedisStrategyConnection implements TransportLayerConnection<Transport.REDI
await RedisHelper.subscribeChannel(this.rpcSubscriber)(replyChannel);
await this.emitMessage(channel, message);
- return firstValueFrom(pipe(
+ return lastValueFrom(pipe(
this.producerSubject.asObservable(),
filter(msg => msg.channel === replyChannel),
take(1),

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
-import { Observable, fromEvent, Subject, defer, firstValueFrom } from 'rxjs';
+import { Observable, fromEvent, Subject, defer, lastValueFrom } from 'rxjs';
import { map, takeUntil, catchError } from 'rxjs/operators';
import {
combineEffects,
@@ -115,7 +115,7 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
input$
.pipe(takeUntil(close$))
.subscribe({
- next: (event: Event) => firstValueFrom(client.sendResponse(event)),
+ next: (event: Event) => lastValueFrom(client.sendResponse(event)),
error: (err: EventError) => {
const type = 'ServerListener';
const message = `Unexpected error for OutgoingEvent stream: "${err.name}", "${err.message}"`;

MODIFY
packages/websockets/src/server/websocket.server.response.ts
packages/websockets/src/server/websocket.server.response.ts
@@ -2,7 +2,7 @@ import * as WebSocket from 'ws';
import * as O from 'fp-ts/lib/Option';
import * as T from 'fp-ts/lib/Task';
import * as TE from 'fp-ts/lib/TaskEither';
-import { firstValueFrom, forkJoin, Observable, of } from 'rxjs';
+import { lastValueFrom, forkJoin, Observable, of } from 'rxjs';
import { catchError, mapTo } from 'rxjs/operators';
import { constant, constVoid, flow, pipe } from 'fp-ts/lib/function';
import { ContextProvider, Event, LoggerToken, LoggerTag, LoggerLevel } from '@marblejs/core';
@@ -76,7 +76,7 @@ export const broadcastEvent: ServerResponseHandler = ({ server, eventTransformer
// mapTo(true),
// );
- firstValueFrom(pipe(
+ lastValueFrom(pipe(
forkJoin([...server.clients].map(client => emitEvent({ client, eventTransformer, ask })(event))),
catchError(constant(of(false))),
mapTo(true),

refactor: firstValueFrom -> lastValueFrom