Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | 1x 1x 1x 1x 1x 1x 1x 1x 24x 47x 24x 24x 24x 24x 25x 25x 25x 50x 25x 25x 25x 25x 25x 25x 25x 25x 1x 32x 15x 16x 14x 14x 15x 15x 7x 7x 22x 22x 1x 7x 4x 4x 4x 4x 4x 4x 13x 2x 2x 4x 4x 1x 25x 1x 25x 1x 56x 56x 56x 100x 100x 71x 71x 71x 25x 25x 17x 17x 16x 16x 4x 34x 25x 1x | import uuid from "uuid/v4";
import http, { Server } from "http";
import StompServer from "stomp-broker-js";
import waitUntil from "../util/waitUntil";
type CallNextMiddleWare = () => boolean;
type MiddlewareStrategy = [
string,
(args: { sessionId: string; frame: Frame }) => void
];
interface Socket {
sessionId: string;
}
interface MiddlewareArgs {
frame: Frame;
}
interface Frame {
headers: {
mockMessageId: string;
};
}
interface Session {
sessionId: string;
hasConnected: boolean;
hasReceivedSubscription: boolean;
hasSentMessage: boolean;
hasDisconnected: boolean;
}
interface Sessions {
[sessionId: string]: Session;
}
interface Config {
port?: number;
portRange?: [number, number];
endpoint?: string;
}
export class MockStompBroker {
private static PORTS_IN_USE: number[] = [];
private static BASE_SESSION = {
hasConnected: false,
hasReceivedSubscription: false,
hasSentMessage: false,
hasDisconnected: false
};
private static getRandomInt(min: number, max: number): number {
return Math.floor(Math.random() * (max - min)) + min;
}
private static getPort(portRange: [number, number] = [8000, 9001]): number {
const minInclusive = portRange[0];
const maxExclusive = portRange[1];
const port = this.getRandomInt(minInclusive, maxExclusive);
return this.PORTS_IN_USE.includes(port) ? this.getPort() : port;
}
private readonly port: number;
private readonly httpServer: Server;
private readonly stompServer: any;
private readonly sentMessageIds: string[] = [];
private queriedSessionIds: string[] = [];
private sessions: Sessions = {};
constructor({ port, portRange, endpoint = "/websocket" }: Config = {}) {
this.thereAreNewSessions = this.thereAreNewSessions.bind(this);
this.registerMiddlewares = this.registerMiddlewares.bind(this);
this.setMiddleware = this.setMiddleware.bind(this);
this.port = port || MockStompBroker.getPort(portRange);
this.httpServer = http.createServer();
this.stompServer = new StompServer({
server: this.httpServer,
path: endpoint
});
this.registerMiddlewares();
this.httpServer.listen(this.port);
}
public async newSessionsConnected(): Promise<string[]> {
await waitUntil(this.thereAreNewSessions, "No new sessions established");
const newSessionsIds = Object.values(this.sessions)
.filter(({ sessionId }) => !this.queriedSessionIds.includes(sessionId))
.filter(({ hasConnected }) => hasConnected)
.map(({ sessionId }) => sessionId);
this.queriedSessionIds = this.queriedSessionIds.concat(newSessionsIds);
return newSessionsIds;
}
public subscribed(sessionId: string) {
return waitUntil(() => {
const session = this.sessions[sessionId];
return Boolean(session && session.hasReceivedSubscription);
}, `Session ${sessionId} never subscribed to a topic`);
}
public scheduleMessage(
topic: string,
payload: any,
headers: {} = {
"content-type": "application/json;charset=UTF-8"
}
): string {
const body = JSON.stringify(payload);
const mockMessageId = uuid();
this.stompServer.send(`/${topic}`, { ...headers, mockMessageId }, body);
return mockMessageId;
}
public messageSent(messageId: string) {
return waitUntil(
() => this.sentMessageIds.includes(messageId),
`Message ${messageId} was never sent`
);
}
public disconnected(sessionId: string) {
return waitUntil(() => {
const session = this.sessions[sessionId];
return Boolean(session && session.hasDisconnected);
}, `Session ${sessionId} never disconnected`);
}
public kill() {
this.httpServer.close();
}
public getPort() {
return this.port;
}
private thereAreNewSessions(): boolean {
const numberOfSessions = Object.entries(this.sessions).length;
const numberOfSessionsQueried = this.queriedSessionIds.length;
return numberOfSessions - numberOfSessionsQueried > 0;
}
private setMiddleware([event, middlewareHook]: MiddlewareStrategy) {
this.stompServer.setMiddleware(
event,
(socket: Socket, args: MiddlewareArgs, next: CallNextMiddleWare) => {
process.nextTick(() =>
middlewareHook({ sessionId: socket.sessionId, frame: args.frame })
);
return next();
}
);
}
private registerMiddlewares() {
const strategies: MiddlewareStrategy[] = [
[
"connect",
({ sessionId }) =>
(this.sessions[sessionId] = {
...MockStompBroker.BASE_SESSION,
sessionId,
hasConnected: true
})
],
[
"subscribe",
({ sessionId }) =>
(this.sessions[sessionId].hasReceivedSubscription = true)
],
[
"send",
({ frame }) => this.sentMessageIds.push(frame.headers.mockMessageId)
],
[
"disconnect",
({ sessionId }) => (this.sessions[sessionId].hasDisconnected = true)
]
];
strategies.forEach(this.setMiddleware);
}
}
|