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 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 | 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 159x 159x 159x 4x 4x 4x 4x 4x 4x 4x 4x 144x 144x 144x 144x 144x 144x 144x 6x 6x 6x 6x 6x 6x 6x 6x 6x 144x 144x 144x 144x 144x 144x 144x 144x 99x 18x 18x 1x 18x 18x 6x 18x 18x 2x 18x 18x 18x 99x 99x 99x 99x 99x 99x 99x 144x 27x 27x 1x 26x 26x 6x 144x 144x 144x 144x 144x 144x 3x 3x 144x 144x 144x 144x 144x 144x 144x 144x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 4x 2x 2x 4x 5x 5x 1x 1x 1x 1x 1x 5x 144x 93x 93x 93x 93x 93x 93x 93x 89x 89x 89x 89x 89x 89x 89x 89x 93x 93x 93x 93x 93x 93x 93x 93x 93x 75x 75x 75x 75x 75x 75x 1x 1x 1x 75x 93x 93x 93x 78x 1x 1x 77x 77x 77x 78x 73x 73x 73x 73x 77x 78x 75x 75x 75x 75x 75x 75x 75x 78x 78x 78x 144x 144x 221x 221x 221x 221x 221x 221x 221x 221x 144x 144x 144x 144x 144x 218x 218x 218x 74x 218x 218x 218x 218x 218x 144x 15x 15x 15x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x 144x | import { Centrifuge, Subscription, SubscriptionState } from 'centrifuge';
import { Watchers } from './Watchers';
/**
* Access to our realtime messaging socket (centrifugo). The server lives
* behind the same origin at /ws/connect and authenticates connections with
* the browser's session cookie (via a server-side connect proxy), so no
* token handling is needed here.
*
* The connection is owned by a SocketManager which is page-scoped - it hangs
* off `window` rather than any component, so it survives components mounting
* and unmounting, and vanilla js on the containing page can share the same
* connection:
*
* const sub = window.sockets.subscribe('notifications:<org>:<user>', (event) => {
* ...
* });
* sub.unsubscribe();
*
* Any number of subscribers (components or page js) can watch the same
* channel - the underlying centrifugo subscription is created on first use
* and torn down when the last subscriber leaves. The connection itself stays
* open for the life of the page. Each published event arrives as raw JSON.
*
* The connection's state is readable too, so a page can tell its user when it
* has gone quiet rather than silently showing stale data:
*
* window.sockets.onConnectionState((state) => { ... });
*/
export interface SocketSubscription {
unsubscribe(): void;
}
export type PublicationHandler = (data: any) => void;
/**
* Where the shared connection is. Mirrors centrifugo's own client states -
* connecting covers the first attempt and every reconnect after a drop, so a
* page showing a "reconnecting" hint wants that one.
*/
export enum ConnectionState {
Disconnected = 'disconnected',
Connecting = 'connecting',
Connected = 'connected'
}
export type ConnectionStateHandler = (state: ConnectionState) => void;
export interface SocketProvider {
subscribe(
channel: string,
onPublication: PublicationHandler,
onSubscribed?: () => void
): SocketSubscription;
publish(channel: string, data: any): Promise<void>;
getConnectionState(): ConnectionState;
onConnectionState(handler: ConnectionStateHandler): SocketSubscription;
}
interface ChannelEntry {
sub: Subscription;
count: number;
}
export class SocketManager implements SocketProvider {
private socket: Centrifuge = null;
private channels = new Map<string, ChannelEntry>();
private createSocket: () => Centrifuge;
// no connection exists until something asks for one, so that is where we
// start rather than pretending we know the server is unreachable
private state = ConnectionState.Disconnected;
private stateHandlers = new Watchers<ConnectionStateHandler>(
'socket connection handler'
);
constructor(createSocket?: () => Centrifuge) {
this.createSocket =
createSocket ||
(() => {
const protocol = window.location.protocol === 'https:' ? 'wss' : 'ws';
const socket = new Centrifuge(
`${protocol}://${window.location.host}/ws/connect`
);
socket.connect();
return socket;
});
}
/**
* Publishes data on a channel. The server proxies client publications
* (e.g. typing events on history channels) to mailroom for authorization
* and fan-out; a rejection means the publication was denied.
*/
public publish(channel: string, data: any): Promise<void> {
const socket = this.ensureSocket();
// prefer the channel's live subscription when we have one
const entry = this.channels.get(channel);
const published = entry
? entry.sub.publish(data)
: socket.publish(channel, data);
return published.then(() => undefined);
}
/**
* The shared connection, opened on first use. Its state is tracked from
* here on, seeded with whatever it is already in - creating it connects, so
* the first transition can happen before we are listening.
*/
private ensureSocket(): Centrifuge {
if (!this.socket) {
this.socket = this.createSocket();
this.socket.on('connecting', () =>
this.setState(ConnectionState.Connecting)
);
this.socket.on('connected', () =>
this.setState(ConnectionState.Connected)
);
this.socket.on('disconnected', () =>
this.setState(ConnectionState.Disconnected)
);
this.setState(this.socket.state as unknown as ConnectionState);
}
return this.socket;
}
// counts transitions, so an initial delivery can tell whether one beat it
// to the handler
private stateSeq = 0;
private setState(state: ConnectionState): void {
if (!state || state === this.state) {
return;
}
this.state = state;
this.stateSeq++;
this.stateHandlers.each((handler) => handler(state));
}
/**
* Where the shared connection is right now. Disconnected until something
* subscribes or publishes, since that is what opens it.
*/
public getConnectionState(): ConnectionState {
return this.state;
}
/**
* Watches the connection. The handler is called on every transition, and
* once up front with the current state so a caller never has to pair this
* with getConnectionState to render.
*/
public onConnectionState(
handler: ConnectionStateHandler
): SocketSubscription {
this.stateHandlers.add(handler);
// where we are now, rather than wherever we've got to by the time the
// prime lands - opening the connection in this same tick transitions us
// and delivers that to the handler ahead of it, and saying the same thing
// again is the repeat this is supposed to suppress
const initial = this.state;
const seq = this.stateSeq;
this.stateHandlers.prime(handler, () => {
if (this.stateSeq === seq) {
handler(initial);
}
});
return {
unsubscribe: () => {
this.stateHandlers.remove(handler);
}
};
}
public subscribe(
channel: string,
onPublication: PublicationHandler,
onSubscribed?: () => void
): SocketSubscription {
const socket = this.ensureSocket();
let entry = this.channels.get(channel);
if (!entry) {
entry = {
sub: socket.getSubscription(channel) || socket.newSubscription(channel),
count: 0
};
this.channels.set(channel, entry);
entry.sub.subscribe();
}
entry.count++;
const sub = entry.sub;
const pubHandler = (ctx: { data: any }) => onPublication(ctx.data);
sub.on('publication', pubHandler);
let subHandler: () => void = null;
if (onSubscribed) {
// fires on every (re)subscribe, including after reconnects, so
// subscribers can catch up on anything missed while offline
subHandler = () => onSubscribed();
sub.on('subscribed', subHandler);
// late joiners on an already-live channel won't see a subscribed
// event, so give them their initial one
if (sub.state === SubscriptionState.Subscribed) {
window.setTimeout(() => subHandler && subHandler(), 0);
}
}
let active = true;
return {
unsubscribe: () => {
if (!active) {
return;
}
active = false;
sub.off('publication', pubHandler);
if (subHandler) {
sub.off('subscribed', subHandler);
subHandler = null;
}
entry.count--;
if (entry.count === 0) {
this.channels.delete(channel);
sub.unsubscribe();
this.socket.removeSubscription(sub);
}
}
};
}
}
// the page-scoped manager, shared with vanilla js as window.sockets and
// reused if another copy of this module already created it
const getManager = (): SocketManager => {
const w = window as any;
if (!w.sockets) {
w.sockets = new SocketManager();
}
return w.sockets;
};
getManager();
// when set, components subscribe through this instead of the page manager
let provider: SocketProvider = null;
export const subscribeToSocket = (
channel: string,
onPublication: PublicationHandler,
onSubscribed?: () => void
): SocketSubscription => {
return (provider || getManager()).subscribe(
channel,
onPublication,
onSubscribed
);
};
export const publishToSocket = (channel: string, data: any): Promise<void> => {
return (provider || getManager()).publish(channel, data);
};
export const getSocketConnectionState = (): ConnectionState => {
return (provider || getManager()).getConnectionState();
};
export const onSocketConnectionState = (
handler: ConnectionStateHandler
): SocketSubscription => {
return (provider || getManager()).onConnectionState(handler);
};
// for tests to swap in a mock provider, returns the previous provider
export const setSocketProvider = (newProvider: SocketProvider) => {
const previous = provider;
provider = newProvider;
return previous;
};
|