style(messaging): subscribeEffects subscribe methods
MODIFY
.eslintrc.js
.eslintrc.js
@@ -17,7 +17,7 @@ module.exports = {
'@typescript-eslint/no-parameter-properties': 0,
'@typescript-eslint/no-empty-interface': 0,
'@typescript-eslint/no-unused-vars': ['error', { 'argsIgnorePattern': '^_' }],
- '@typescript-eslint/no-use-before-define': ['error', { 'typedefs': false }],
+ '@typescript-eslint/no-use-before-define': 0,
'@typescript-eslint/prefer-interface': 0,
'@typescript-eslint/explicit-function-return-type': 0,
'@typescript-eslint/explicit-member-accessibility': 0,

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
@@ -59,10 +59,8 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
.pipe(takeUntil(conn.close$))
.subscribe();
- const subscribeEffects = (input$: Observable<TransportMessage<any>>) => input$
- .pipe(takeUntil(conn.close$))
- .subscribe(
- msg => {
+
+ const onSubscribeEffectsOutput = (conn: TransportLayerConnection) => (msg: TransportMessage<any>) => {
if (msg.replyTo) {
conn.sendMessage(msg.replyTo, {
data: msgTransformer.encode(msg.data),
@@ -71,11 +69,18 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
});
}
conn.ack(msg.raw);
- },
- error => {
+ }
+
+ const onSubscribeEffectsError = (errorSubject: Subject<Error>) => (error: Error) => {
errorSubject.next(error);
if (effectsSub.closed) { effectsSub = subscribeEffects(message$); }
- },
+ }
+
+ const subscribeEffects = (input$: Observable<TransportMessage<any>>) => input$
+ .pipe(takeUntil(conn.close$))
+ .subscribe(
+ onSubscribeEffectsOutput(conn),
+ onSubscribeEffectsError(errorSubject),
);
effectsSub = subscribeEffects(message$);

style(messaging): subscribeEffects subscribe methods