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 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 5x 5x 5x 5x 5x 5x 5x 7x 7x 7x 7x 7x 4x 2x 2x 4x 7x 7x 7x 7x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 5x 5x 5x 5x 5x 5x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x | import { firstValueFrom, of } from "rxjs";
import { DurableSocketChannel, RPCChannel } from "./channel";
import { DurableSocket } from "./durable-socket";
import { AnyConstructor, getRpcServiceName, getRpcUrl } from "./internal";
import { MethodsOf, Proxied, RemoteSubscription } from "./proxied";
import { Remotable } from "./remotable";
import { RPCSession } from "./session";
interface EventSubscription {
observer: Function;
remoteSubscription: RemoteSubscription;
}
const SERVICE_PROXY_SESSION = Symbol('SERVICE_PROXY_SESSION');
/**
* Creates an immediate proxy for a remote service without waiting for:
* - The channel to be created
* - The channel to be ready
* - The RPC session to be created
* - The RPC service to be acquired
*
* Automatically handles state loss by reacquiring the remote service once connection is ready again, and replaying
* any active subscription calls.
*/
export function createServiceProxy<T extends object, U extends object = {}>(sessionPromise: Promise<RPCSession>, klass: AnyConstructor<T>, target?: U): Proxied<T> & U {
let servicePromise: Promise<Proxied<T>> | undefined;
let methodTable = new Map<string | symbol, Function>();
let eventObservers = new Map<string | symbol, EventSubscription[]>();
let sessionReady: Promise<RPCSession>; // resolves only when the session is ready, value will change over time
async function acquireService() {
let service = await (await sessionReady).getRemoteService(klass);
// Resubscribe to events
try {
for (let [eventName, subscriptions] of eventObservers.entries()) {
for (let eventSub of subscriptions) {
eventSub.remoteSubscription = await service[eventName].subscribe(eventSub.observer);
}
}
} catch (e) {
throw new Error(`While restoring subscriptions to events after state loss: ${e.stack || e}`);
}
return service;
};
sessionReady = sessionPromise.then(async session => {
// Session initialization
// Needs to run exactly once, but needs to wait for the session to be ready (the first time).
// As the session transitions from ready to not-ready and back, we'll maintain the sessionReady
// promise so that new calls can correctly wait for the right state to execute.
let ready = true;
session.channel.stateLost?.subscribe(() => {
// Protect against channel types that emit multiple stateLost events before a ready event.
// If this were to occur without this check, we could resubscribe to events multiple times.
if (!ready)
return;
// Set ourselves up for the next time the connection is ready, including reacquiring the service
// as soon as possible (so that we can resubscribe to events).
ready = false;
sessionReady = (session.channel.ready ? firstValueFrom(session.channel.ready) : Promise.resolve(session)).then(() => session);
servicePromise = sessionReady.then(() => acquireService());
});
if (session.channel.ready)
session.channel.ready.subscribe(() => ready = true);
else
ready = true;
await session.channel.ready ? firstValueFrom(session.channel.ready) : Promise.resolve();
return session;
});
/**
* Acquire the service. It's important that this only be called when a method call or subscribe action has been
* initiated. If this occurs before that moment, then it is possible to produce a deadlock if the caller uses
* session.lock() before we are able to start our own getRemoteService() call.
* @returns
*/
let serviceProvider = async () => {
let service = await (servicePromise ??= acquireService());
if (!service)
throw new Error(`Service.proxy(): No such remote service with ID '${getRpcServiceName(klass)}' (for class ${klass.name})`);
return service;
};
return <Proxied<T> & U> new Proxy<any>(target ?? {}, {
get(target, p) {
if (p in target)
return target[p];
if (p === SERVICE_PROXY_SESSION)
return sessionPromise;
if (methodTable.has(p))
return methodTable.get(p);
let method = async (...args) => (await serviceProvider())[p](...args);
method['subscribe'] = async (observer, ...args): Promise<RemoteSubscription> => {
if (!eventObservers.has(p))
eventObservers.set(p, []);
let observerList = eventObservers.get(p);
let eventSub: EventSubscription = {
observer,
remoteSubscription: (await serviceProvider())[p].subscribe(...args)
};
observerList.push(eventSub);
return {
unsubscribe: async () => {
await eventSub.remoteSubscription.unsubscribe();
let index = observerList.indexOf(eventSub);
if (index >= 0)
observerList.splice(index, 1);
}
};
};
methodTable.set(p, method);
return method;
}
});
}
/**
* Provides a powerful and ergonomic way to consume well-known remote objects (services) over Conduit.
*
* ### Why is this necessary?
*
* When using the lower level RPCSession API, a caller must:
* - Choose and establish a communication channel (such as DurableSocket) for the communication to occur over
* - Wait for the communication channel to become ready
* - Request a remote service object via getRemoteService() and wait for the call to complete
*
* All of the above must occur before the first method call can be sent, and the caller must also maintain the
* state of the connection themselves. When the communication channel loses state (for example, when a network
* disconnect occurs), the caller must wait for the communication channel to become ready again, acquire a new
* service object and otherwise manually restore the state of the connection.
*
* While the lower level API is extremely powerful and allows for communication patterns that are typically not
* possible with other RPC systems, it is a lot of manual work to do a job that when using REST communication is
* extremely simple.
*
* The Service class takes care of all of this for you and more.
*
* ### Constraints of using Conduit Services
*
* Effective use requires committing to several assumptions that Conduit itself does *not* make:
*
* - You will communicate over HTTP/Websockets using DurableSocket
* (Conduit itself supports communicating over any arbitrary communication medium, including locally in the same
* process).
*
* - You will primarily adopt a client/server architecture, where the party initiating the connection is the client
* and the party receiving the connection is the server.
* (Conduit does not require constraining the roles of participants)
*
* - You will primarily make method calls on well-known (service) objects, not transient objects
* (Conduit supports leasing any object across the communication channel along with method calls to those
* transient objects)
*
* - You will use method calls as the primary method of sending data to the server participant, and you will use
* Observable events as your primary method of sending data to the client
* (Conduit itself allows either party to acquire services from the other party and make method calls or event
* subscriptions as they wish)
*
* ### Usage
*
* Servers providing Conduit Services should provide a library component (typically an NPM package)
* which contains abstract service classes deriving from the Service base class.
* - Each class should be decorated with `@Name()` to establish the well-known name for the service over Conduit.
* - Each class should be decorated with `@URL()` to establish the default URL endpoint to connect to.
*
* Clients will acquire a local proxy for a service by calling the static `proxy()` method on the service's class.
* Those objects can immediately be used to perform calls or subscribe to events.
*
* When the first call or event subscription occurs, the service proxy object will:
* - Acquire or create an appropriate DurableSocketChannel for the configured endpoint URL. If one already exists,
* it will be used (only one connection to the server will be established).
* - Acquire or create an appropriate RPCSession for the configured endpoing URL. If one already exists,
* it will be used (only one session will exist between client/server)
* - Acquire a Conduit remote proxy for the service
*
* The service object will monitor the underlying channel for stateLost/ready events, and automatically reacquire remote
* proxies on your behalf. The service object will also automatically maintain local state about event subscriptions and
* ensure that the subscriptions are recreated when the connection becomes ready after state loss.
*
* Thus the intricacies of managing a long-lived RPC session while your app is running are abstracted away, letting you
* simply make calls. Remember that Conduit will automatically reject any pending method calls that are outstanding
* when connection loss occurs, allowing you to address that within your normal business logic layer.
*/
@Remotable()
export class Service {
/**
* Construct a new proxy for this service pointing at the URL specified by the @URL() decorator.
* The connection will be established and re-established automatically, the returned service
* proxy is immediately available for use. Requests to the proxy will be automatically delayed
* while the connection is established and the service object is obtained from the remote endpoint.
* @param socketUrl The URL of the WebSocket server which supports Conduit.
*/
static proxy<T extends object>(this: AnyConstructor<T>): MethodsOf<T>;
/**
* Construct a new proxy for this service pointing at the given WebSocket URL.
* The connection will be established and re-established automatically, the returned service
* proxy is immediately available for use. Requests to the proxy will be automatically delayed
* while the connection is established and the service object is obtained from the remote endpoint.
* @param socketUrl The URL of the WebSocket server which supports Conduit.
*/
static proxy<T extends object>(this: AnyConstructor<T>, socketUrl: string): Proxied<T>;
/**
* Construct a new proxy for this service pointing at the given RPCChannel.
* The returned service proxy is immediately available for use. Requests to the
* proxy will be automatically delayed while the channel promise is resolved and the
* service object is obtained from the remote endpoint.
* @param channel A promise for obtaining the channel to use
*/
static proxy<T extends object>(this: AnyConstructor<T>, channel: Promise<RPCChannel>): Proxied<T>;
/**
* Construct a new proxy for the service identified by this class, which is running remotely on the
* other side of the given RPCChannel.
*
* If channelOrEndpoint is a string, it is treated as a WebSockets URL, and a new durable WebSocket channel
* connection will be created. If a connection to the endpoint already exists, the connection will be reused.
*
* The returned service proxy is immediately available for use without awaiting. Requests to the
* proxy will be automatically delayed while the service object is obtained from the
* remote endpoint.
*
* @param channel The channel to connect to
*/
static proxy<T extends object>(this: AnyConstructor<T>, channel: RPCChannel): Proxied<T>;
static proxy<T extends object>(this: AnyConstructor<T>, channelOrEndpoint?: string | Promise<RPCChannel> | RPCChannel): Proxied<T> {
channelOrEndpoint ??= getRpcUrl(this);
let channelPromise: Promise<RPCChannel>;
if (typeof channelOrEndpoint === 'string') {
let endpointChannel = Service.channelForEndpoint(channelOrEndpoint);
channelPromise = firstValueFrom(endpointChannel.ready ?? of()).then(() => endpointChannel);
} else {
channelPromise = Promise.resolve(channelOrEndpoint);
}
let proxy = createServiceProxy<T>(channelPromise.then(channel => Service.sessionForChannel(channel)), this);
if (typeof channelOrEndpoint === 'string') {
Reflect.defineMetadata('rpc:endpoint', channelOrEndpoint, proxy);
}
return proxy;
}
static sessionOf(service): RPCSession {
return this.sessionForChannel(this.channelOf(service));
}
static channelOf(service): RPCChannel {
return this.channelForEndpoint(this.endpointOf(service));
}
static endpointOf(service): string {
return Reflect.getMetadata('rpc:endpoint', service);
}
private static channelSessions = new WeakMap<RPCChannel, RPCSession>();
/**
* Retrieve the RPCSession for a given Service object, as returned by Service.proxy()
* @param service
* @returns
*/
static sessionForProxy(service: Service): Promise<RPCSession> {
return service[SERVICE_PROXY_SESSION];
}
/**
* Acquire the RPCSession object associated with the given channel. This will be the same session used
* by Service.proxy() when used in concert with the given channel.
* @param channel
* @returns
*/
static sessionForChannel(channel: RPCChannel) {
if (this.channelSessions.has(channel))
return this.channelSessions.get(channel);
const session = new RPCSession(channel);
this.channelSessions.set(channel, session);
return session;
}
private static endpointChannels = new Map<string, WeakRef<DurableSocketChannel>>();
/**
* Get or create the RPCChannel object associated with the given endpoint URL. This will be the same channel
* used by Service.proxy() when used in concert with the given endpoint URL.
* @param channel
* @returns
*/
static channelForEndpoint(endpoint: string): RPCChannel {
let channel = this.endpointChannels.get(endpoint)?.deref();
if (channel) {
return channel;
} else {
this.endpointChannels.set(
endpoint,
new WeakRef(channel = new DurableSocketChannel(new DurableSocket(endpoint)))
);
}
return channel;
}
} |