refactor(websockets): adjusted package to changed Effect interface
MODIFY
packages/@integration/src/websockets/http.server.ts
packages/@integration/src/websockets/http.server.ts
@@ -20,7 +20,7 @@ export const WsServerToken = createContextToken<MarbleWebSocketServer>('MarbleWe
const root$ = r.pipe(
r.matchPath('/'),
r.matchType('GET'),
- r.useEffect((req$, _, { ask }) => {
+ r.useEffect((req$, { ask }) => {
const wsServer = useContext(WsServerToken)(ask);
return req$.pipe(
@@ -29,7 +29,7 @@ const root$ = r.pipe(
);
}));
-const upgrade$: HttpServerEffect = (event$, _, { ask }) =>
+const upgrade$: HttpServerEffect = (event$, { ask }) =>
event$.pipe(
matchEvent(ServerEvent.upgrade),
mapToServer({

MODIFY
packages/@integration/src/websockets/websockets.server.ts
packages/@integration/src/websockets/websockets.server.ts
@@ -8,18 +8,18 @@ import {
import { iif, throwError, of } from 'rxjs';
import { tap, mergeMap, buffer, map } from 'rxjs/operators';
import { matchEvent, use } from '@marblejs/core';
-import { eventValidator$, t } from '../../../middleware-io/dist';
+import { eventValidator$, t } from '@marblejs/middleware-io';
const sum$: WsEffect = event$ =>
event$.pipe(
matchEvent('SUM')
);
-const add$: WsEffect = (event$, ...args) =>
+const add$: WsEffect = (event$, meta) =>
event$.pipe(
matchEvent('ADD'),
use(eventValidator$(t.number)),
- buffer(sum$(event$, ...args)),
+ buffer(sum$(event$, meta)),
map(addEvents => addEvents.reduce((a, e) => e.payload + a, 0)),
map(payload => ({ type: 'SUM_RESULT', payload })),
);

MODIFY
packages/websockets/src/error/specs/ws-error.handler.spec.ts
packages/websockets/src/error/specs/ws-error.handler.spec.ts
@@ -4,37 +4,43 @@ import { handleEffectsError } from '../ws-error.handler';
import { MarbleWebSocketClient } from '../../websocket.interface';
import { WsErrorEffect } from '../../effects/ws-effects.interface';
+const createMockClient = (): MarbleWebSocketClient =>
+ ({ sendResponse: jest.fn() }) as any;
+
describe('#handleEffectsError', () => {
- let defaultMetadata: EffectMetadata;
+ let metadata: EffectMetadata<MarbleWebSocketClient>;
beforeEach(() => {
- defaultMetadata = createEffectMetadata({ ask: lookup(createContext()) });
+ metadata = createEffectMetadata({
+ ask: lookup(createContext()),
+ client: createMockClient(),
+ });
});
test('handles error if error$ is defined', () => {
// given
- const client = { sendResponse: jest.fn() } as any as MarbleWebSocketClient;
const error = new EventError({ type: 'EVENT' }, '');
- const error$: WsErrorEffect = event$ => event$.pipe(
+ const error$: WsErrorEffect = event$ =>
+ event$.pipe(
mapTo({ type: error.event.type, error: {} }),
);
// when
- handleEffectsError(defaultMetadata, client, error$)(error);
+ handleEffectsError(metadata, error$)(error);
// then
- expect(client.sendResponse).toHaveBeenCalled();
+ expect(metadata.client.sendResponse).toHaveBeenCalled();
});
test('does nothing if error$ is undefined', () => {
// given
- const client = { sendResponse: jest.fn() } as any as MarbleWebSocketClient;
const error = new EventError({ type: 'EVENT' }, '');
+ const error$ = undefined;
// when
- handleEffectsError(defaultMetadata, client, undefined)(error);
+ handleEffectsError(metadata, error$)(error);
// then
- expect(client.sendResponse).not.toHaveBeenCalled();
+ expect(metadata.client.sendResponse).not.toHaveBeenCalled();
});
});

MODIFY
packages/websockets/src/error/ws-error.handler.ts
packages/websockets/src/error/ws-error.handler.ts
@@ -4,8 +4,7 @@ import { MarbleWebSocketClient } from '../websocket.interface';
import { WsErrorEffect } from '../effects/ws-effects.interface';
export const handleEffectsError = <IncomingError extends EventError>(
- metadata: EffectMetadata,
- client: MarbleWebSocketClient,
+ metadata: EffectMetadata<MarbleWebSocketClient>,
error$: WsErrorEffect<IncomingError, any, any> | undefined,
) => (error: IncomingError) => {
if (error$) {
@@ -14,6 +13,6 @@ export const handleEffectsError = <IncomingError extends EventError>(
error,
});
- error$(input$, client, metadata).subscribe(client.sendResponse);
+ error$(input$, metadata).subscribe(metadata.client.sendResponse);
}
};

MODIFY
packages/websockets/src/listener/specs/websocket.listener.spec.ts
packages/websockets/src/listener/specs/websocket.listener.spec.ts
@@ -36,7 +36,7 @@ describe('WebSocket listener', () => {
test('echoes back to all clients', done => {
// given
- const echo$: WsEffect = (event$, client) => event$.pipe(
+ const echo$: WsEffect = (event$, { client }) => event$.pipe(
mergeMap(client.sendBroadcastResponse),
);
const event = JSON.stringify({ type: 'EVENT', payload: 'test' });

MODIFY
packages/websockets/src/listener/websocket.listener.ts
packages/websockets/src/listener/websocket.listener.ts
@@ -57,22 +57,22 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
const handleMessage: HandleMessage = (client, ask) => {
const eventSubject$ = new Subject<any>();
const incomingEventSubject$ = new Subject<WS.WebSocketData>();
- const defaultMetadata = createEffectMetadata({ ask });
+ const metadata = createEffectMetadata({ ask, client });
const decodedEvent$ = incomingEventSubject$.pipe(map(providedTransformer.decode));
- const middlewares$ = combinedMiddlewares(decodedEvent$, client, defaultMetadata);
- const effects$ = combinedEffects(eventSubject$, client, defaultMetadata);
- const effectsOutput$ = output$(effects$, client, defaultMetadata);
+ const middlewares$ = combinedMiddlewares(decodedEvent$, metadata);
+ const effects$ = combinedEffects(eventSubject$, metadata);
+ const effectsOutput$ = output$(effects$, metadata);
const subscribeMiddlewares = (input$: Observable<any>) =>
input$.subscribe(
event => eventSubject$.next(event),
- error => handleEffectsError(defaultMetadata, client, providedError$)(error),
+ error => handleEffectsError(metadata, providedError$)(error),
);
const subscribeEffects = (input$: Observable<any>) =>
input$.subscribe(
event => client.sendResponse(event),
- error => handleEffectsError(defaultMetadata, client, providedError$)(error),
+ error => handleEffectsError(metadata, providedError$)(error),
);
let middlewaresSub = subscribeMiddlewares(middlewares$);
@@ -106,7 +106,7 @@ export const webSocketListener = (config: WebSocketListenerConfig = {}) => {
};
const verifyClient = (ask: ContextProvider): WebSocket.VerifyClientCallbackAsync => (info, callback) => {
- connection$(of(info.req), undefined, createEffectMetadata({ ask }))
+ connection$(of(info.req), createEffectMetadata({ ask, client: undefined }))
.pipe(map(Boolean))
.subscribe(
isVerified => callback(isVerified),

MODIFY
packages/websockets/src/operators/broadcast/broadcast.operator.spec.ts
packages/websockets/src/operators/broadcast/broadcast.operator.spec.ts
@@ -5,6 +5,7 @@ import { WsEffect } from '../../effects/ws-effects.interface';
describe('#broadcast operator', () => {
const client = { sendBroadcastResponse: jest.fn(() => EMPTY) };
+ const meta = { client };
test('sends broadcast response', () => {
// given
@@ -12,7 +13,7 @@ describe('#broadcast operator', () => {
const outgoingEvent = { type: 'TEST_EVENT_RESPONSE', payload: 'test_payload' };
// when
- const effect$: WsEffect = (event$, client) =>
+ const effect$: WsEffect = (event$, { client }) =>
event$.pipe(
broadcast(client, () => outgoingEvent),
);
@@ -21,7 +22,7 @@ describe('#broadcast operator', () => {
Marbles.assertEffect(effect$, [
['-a--', { a: incomingEvent }],
['----', {}],
- ], { client });
+ ], { meta });
expect(client.sendBroadcastResponse).toHaveBeenCalledWith(outgoingEvent);
});

refactor(websockets): adjusted package to changed Effect interface