fix(websockets): broadcast operator type inference
MODIFY
packages/websockets/src/operators/broadcast/websocket.broadcast.operator.spec.ts
packages/websockets/src/operators/broadcast/websocket.broadcast.operator.spec.ts
import { Marbles } from '@marblejs/core/dist/+internal';
-import { EMPTY } from 'rxjs';
+import { of } from 'rxjs';
import { WsEffect } from '../../effects/websocket.effects.interface';
import { broadcast } from './websocket.broadcast.operator';
describe('#broadcast operator', () => {
- const client = { sendBroadcastResponse: jest.fn(() => EMPTY) };
+ const client = { sendBroadcastResponse: jest.fn(() => of(true)) };
const ctx = { client };
test('sends broadcast response', () => {
@@ -21,7 +21,7 @@ describe('#broadcast operator', () => {
// then
Marbles.assertEffect(effect$, [
['-a--', { a: incomingEvent }],
- ['----', {}],
+ ['-b--', { b: outgoingEvent }],
], { ctx });
expect(client.sendBroadcastResponse).toHaveBeenCalledWith(outgoingEvent);

MODIFY
packages/websockets/src/operators/broadcast/websocket.broadcast.operator.ts
packages/websockets/src/operators/broadcast/websocket.broadcast.operator.ts
import { Observable } from 'rxjs';
-import { mergeMap, map } from 'rxjs/operators';
+import { mergeMap, map, mapTo } from 'rxjs/operators';
import { WebSocketClientConnection } from '../../server/websocket.server.interface';
export const broadcast =
- <Input>(client: WebSocketClientConnection, fn: (input: Input) => any) =>
- (input$: Observable<Input>): Observable<boolean> =>
+ <Input, T>(client: WebSocketClientConnection, fn: (input: Input) => T) =>
+ (input$: Observable<Input>): Observable<T> =>
input$.pipe(
map(fn),
- mergeMap(data => client.sendBroadcastResponse(data)),
+ mergeMap(data => client.sendBroadcastResponse(data).pipe(
+ mapTo(data),
+ )),
);

fix(websockets): broadcast operator type inference