refactor(messaging): usage of #createListener from
MODIFY
packages/core/src/http/server/http.server.listener.ts
packages/core/src/http/server/http.server.listener.ts
@@ -29,7 +29,7 @@ export const httpListener = createListener<HttpListenerConfig, HttpListener>(con
middlewares = [],
output$,
error$,
- } = config;
+ } = config ?? {};
const client = useContext(ServerClientToken)(ask);
const effectContext = createEffectContext({ ask, client });

MODIFY
packages/core/src/listener/listener.factory.ts
packages/core/src/listener/listener.factory.ts
@@ -3,7 +3,7 @@ import { createReader, ReaderHandler } from '../context/context.reader.factory';
import { ListenerConfig, Listener } from './listener.interface';
export type ListenerHandler<T extends ListenerConfig, U> =
- (config: T) => ReaderHandler<U>;
+ (config?: T) => ReaderHandler<U>;
export const createListener = <T extends ListenerConfig, U>(
handler: ListenerHandler<T, U>

MODIFY
packages/core/src/listener/listener.interface.ts
packages/core/src/listener/listener.interface.ts
@@ -10,5 +10,5 @@ export interface ListenerConfig<T = any> {
}
export interface Listener<T extends ListenerConfig, U> {
- (config: T): Reader<Context, U>;
+ (config?: T): Reader<Context, U>;
}

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
import {
- reader,
Event,
- ContextProvider,
combineMiddlewares,
combineEffects,
createEffectContext,
useContext,
EventError,
+ createListener,
} from '@marblejs/core';
import * as O from 'fp-ts/lib/Option';
-import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
import { flow } from 'fp-ts/lib/function';
import { Observable, Subscription, Subject, of, zip, OperatorFunction} from 'rxjs';
@@ -18,7 +16,6 @@ import {
TransportMessage,
TransportMessageTransformer,
TransportLayerConnection,
- TransportLayer,
} from '../transport/transport.interface';
import { jsonTransformer } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
@@ -36,24 +33,26 @@ export interface MessagingListenerConfig {
msgTransformer?: TransportMessageTransformer<any>;
}
+export interface MessagingListener {
+ listen: () => Promise<TransportLayerConnection>;
+}
+
const defaultOutput$: MsgOutputEffect = msg$ => msg$.pipe(map(m => m.event));
const defaultError$: MsgErrorEffect = msg$ => msg$;
const defaultEffect$: MsgEffect = msg$ => msg$;
-export const messagingListener = (config: MessagingListenerConfig = {}) => {
+export const messagingListener = createListener<MessagingListenerConfig, MessagingListener>(config => ask => {
+ const transportLayer = useContext(TransportLayerToken)(ask);
+
const {
effects = [defaultEffect$],
middlewares = [],
output$ = defaultOutput$,
error$ = defaultError$,
msgTransformer = jsonTransformer,
- } = config;
+ } = config ?? {};
- const handleConnection = (
- conn: TransportLayerConnection,
- serverEventsSubject: Subject<AllServerEvents>,
- ask: ContextProvider,
- ) => {
+ const handleConnection = (conn: TransportLayerConnection, serverEventsSubject: Subject<AllServerEvents>) => {
let effectsSub: Subscription;
const errorSubject = new Subject<Error>();
@@ -158,12 +157,12 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
effectsSub = subscribeEffects(message$);
};
- const listen = (transportLayer: TransportLayer, ask: ContextProvider) => async () => {
+ const listen = async () => {
const { host, channel } = transportLayer.config;
const serverEventsSubject = pipe(ask(ServerEventsToken), O.getOrElse(() => undefined as unknown as Subject<AllServerEvents>));
const connection = await transportLayer.connect({ isConsumer: true });
- handleConnection(connection, serverEventsSubject, ask);
+ handleConnection(connection, serverEventsSubject);
connection.status$
.pipe(takeUntil(connection.close$))
@@ -172,11 +171,5 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
return connection;
};
- return pipe(reader, R.map(ask => {
- const transportLayer = useContext(TransportLayerToken)(ask);
-
- return {
- listen: listen(transportLayer, ask),
- }
- }));
-};
+ return { listen };
+});

refactor(messaging): usage of #createListener from @marblejs/core