Skip to content

Commit eefa79c

Browse files
skiplabsdanielskiplabsdaniel
committed
[skiprutime] Add notifier callbacks
1 parent 176c098 commit eefa79c

File tree

8 files changed

+186
-45
lines changed

8 files changed

+186
-45
lines changed

‎skiplang/prelude/src/skstore/Context.sk‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,8 +54,10 @@ base class CmdKind {
5454
| NNotify(String)
5555
| NTail()
5656
| NWatch(
57+
identifier: String,
5758
start: Tick,
5859
(DirName, Array<(Key, Array<File>)>, Tick, Bool) ~> void,
60+
() ~> void,
5961
changes: Bool,
6062
)
6163
}
@@ -862,7 +864,7 @@ mutable class Context{
862864
| NTail() ->
863865
cond = unfreezeCond(sub.cond);
864866
_ = condBroadcast(cond)
865-
| NWatch(initTick, fn, changes) ->
867+
| NWatch(_, initTick, fn, _, changes) ->
866868
for (dirSub in sub.dirSubs) {
867869
dirName = dirSub.dirName;
868870
this.unsafeMaybeGetDir(dirName) match {

‎skipruntime-ts/native/src/BaseTypes.sk‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,4 +97,16 @@ base class Checker extends Request {
9797
fun check(request: String): void;
9898
}
9999

100+
base class Notifier {
101+
fun subscribed(): void;
102+
103+
fun notify(
104+
values: Array<(SKJSON.CJSON, Array<SKJSON.CJSON>)>,
105+
watermark: String,
106+
updates: Bool,
107+
): void;
108+
109+
fun close(): void;
110+
}
111+
100112
module end;

‎skipruntime-ts/native/src/Extern.sk‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -498,6 +498,10 @@ fun getUniqueOfLazyCollection(lazy: String, key: SKJSON.CJSON): SKJSON.CJSON {
498498

499499
/************ Notifier ****************/
500500

501+
@cpp_extern("SkipRuntime_Notifier__subscribed")
502+
@debug
503+
native fun subscribedOfNotifier(notifier: UInt32): void;
504+
501505
@cpp_extern("SkipRuntime_Notifier__notify")
502506
@debug
503507
native fun notifyOfNotifier(
@@ -507,16 +511,24 @@ native fun notifyOfNotifier(
507511
updates: Int32,
508512
): void;
509513

514+
@cpp_extern("SkipRuntime_Notifier__close")
515+
@debug
516+
native fun closeOfNotifier(notifier: UInt32): void;
517+
510518
@cpp_extern("SkipRuntime_deleteNotifier")
511519
@debug
512520
native fun deleteNotifier(notifier: UInt32): void;
513521

514522
@export("SkipRuntime_createNotifier")
515523
fun createNotifier(notifier: UInt32): Notifier {
516-
Notifier(SKStore.ExternalPointer::create(notifier, deleteNotifier))
524+
ExternNotifier(SKStore.ExternalPointer::create(notifier, deleteNotifier))
517525
}
518526

519-
class Notifier(eptr: SKStore.ExternalPointer) {
527+
class ExternNotifier(eptr: SKStore.ExternalPointer) extends Notifier {
528+
fun subscribed(): void {
529+
subscribedOfNotifier(this.eptr.value)
530+
}
531+
520532
fun notify(
521533
values: Array<(SKJSON.CJSON, Array<SKJSON.CJSON>)>,
522534
watermark: String,
@@ -531,6 +543,10 @@ class Notifier(eptr: SKStore.ExternalPointer) {
531543
Int32::truncate(if (updates) 1 else 0),
532544
)
533545
}
546+
547+
fun close(): void {
548+
closeOfNotifier(this.eptr.value)
549+
}
534550
}
535551

536552
/************ Reducer ****************/
@@ -683,7 +699,7 @@ fun subscribeOfRuntime(
683699
watermark: ?String,
684700
): Int {
685701
SKStore.runWithResult(context ~> {
686-
subscribe(context, reactiveId, notifier.notify, watermark)
702+
subscribe(context, reactiveId, notifier, watermark)
687703
}) match {
688704
| Success(id) -> id
689705
| Failure(err) -> -getErrorHdl(err).toInt()

‎skipruntime-ts/native/src/Runtime.sk‎

Lines changed: 80 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,12 @@ class ExistingResourceException() extends Exception {
1616
}
1717
}
1818

19+
class ExistingSubscriptionException() extends Exception {
20+
fun getMessage(): String {
21+
"A resource instance with specified identifier already subscribed."
22+
}
23+
}
24+
1925
class Params(value: Map<String, PValue>) extends SKStore.File uses Orderable {
2026
fun compare(other: Params): Order {
2127
keys = this.value.keys().collect(Array).sorted();
@@ -475,9 +481,7 @@ fun closeService(): Result<void, .Exception> {
475481
resourceHdl.items(context).each(kf -> {
476482
kf.i1.next().each(_f -> todestroy.push(SKStore.SID::keyType(kf.i0)))
477483
});
478-
todestroy.each(key -> {
479-
resourceHdl.writeArray(context, key, Array[]);
480-
});
484+
todestroy.each(key -> destroyReactiveResource(context, key));
481485
context
482486
.getPersistent(kRemoteSpecifiers)
483487
.map(RemoteSpecifiers::type)
@@ -880,19 +884,22 @@ class Collection(
880884

881885
fun subscribe(
882886
context: mutable SKStore.Context,
887+
identifier: String,
883888
session: Int,
884889
from: SKStore.Tick,
885890
notify: (
886891
Array<(SKJSON.CJSON, Array<SKJSON.CJSON>)>,
887892
SKStore.Tick,
888893
Bool,
889894
) ~> void,
895+
close: () ~> void,
890896
): void {
891897
keyConverter = this.keyConverter;
892898
fileConverter = this.fileConverter;
893899
context.subscribe(
894900
session,
895901
SKStore.NWatch(
902+
identifier,
896903
from,
897904
(_dirName, values, tick, update) ~> {
898905
notify(
@@ -903,6 +910,7 @@ class Collection(
903910
update,
904911
)
905912
},
913+
close,
906914
true,
907915
),
908916
None(),
@@ -1241,27 +1249,61 @@ fun getForKey(
12411249
res;
12421250
}
12431251

1252+
fun destroyReactiveResource(
1253+
context: mutable SKStore.Context,
1254+
sid: SKStore.SID,
1255+
): void {
1256+
context
1257+
.getPersistent(`subscription.${sid}`)
1258+
.map(SKStore.IntFile::type) match {
1259+
| Some(subId) ->
1260+
session = subId.value;
1261+
context.sessions.maybeGet(session).each(sub -> {
1262+
close = sub.cmd match {
1263+
| SKStore.NWatch(_, _, _, close, _) -> close
1264+
| _ -> invariant_violation("Not manage session kind")
1265+
};
1266+
context.removePersistent(`subscription.${sid}`);
1267+
close();
1268+
context.unsubscribe(session);
1269+
})
1270+
| _ -> void
1271+
};
1272+
resourceHdl = SKStore.EHandle(
1273+
SKStore.SID::keyType,
1274+
ResourceDef::type,
1275+
kResourceSessionDir,
1276+
);
1277+
resourceHdl.writeArray(context, sid, Array[]);
1278+
}
1279+
12441280
fun closeReactiveResource(
12451281
context: mutable SKStore.Context,
12461282
identifier: String,
12471283
update: Bool = true,
12481284
): void {
1249-
garbageHdl = SKStore.EHandle(
1250-
SKStore.SID::keyType,
1251-
SKStore.IntFile::type,
1252-
kResourceGarbageDir,
1253-
);
1254-
sid = SKStore.SID(identifier);
1255-
if (garbageHdl.maybeGet(context, sid).isSome()) return void;
1256-
time = Time.time_ms();
1257-
garbageHdl.writeArray(context, sid, Array[SKStore.IntFile(time)]);
1258-
if (update) updateContext(context);
1285+
context
1286+
.getPersistent(`subscription.${identifier}`)
1287+
.map(SKStore.IntFile::type) match {
1288+
| Some(subId) -> unsubscribe(context, subId.value, update)
1289+
| _ ->
1290+
garbageHdl = SKStore.EHandle(
1291+
SKStore.SID::keyType,
1292+
SKStore.IntFile::type,
1293+
kResourceGarbageDir,
1294+
);
1295+
sid = SKStore.SID(identifier);
1296+
if (garbageHdl.maybeGet(context, sid).isSome()) return void;
1297+
time = Time.time_ms();
1298+
garbageHdl.writeArray(context, sid, Array[SKStore.IntFile(time)]);
1299+
if (update) updateContext(context)
1300+
};
12591301
}
12601302

12611303
fun subscribe(
12621304
context: mutable SKStore.Context,
12631305
identifier: String,
1264-
notify: (Array<(SKJSON.CJSON, Array<SKJSON.CJSON>)>, String, Bool) ~> void,
1306+
notifier: Notifier,
12651307
optWatermark: ?String,
12661308
): Int {
12671309
garbageHdl = SKStore.EHandle(
@@ -1282,6 +1324,10 @@ fun subscribe(
12821324
ResourceInfo::type,
12831325
kResourceCollectionsDir,
12841326
);
1327+
subId = `subscription.${identifier}`;
1328+
if (context.getPersistent(subId).isSome()) {
1329+
throw ExistingSubscriptionException()
1330+
};
12851331
resourceHdl.maybeGet(context, sid) match {
12861332
| Some(definition) ->
12871333
info = resourcesCollectionsHdl.get(context, definition);
@@ -1291,21 +1337,38 @@ fun subscribe(
12911337
watermark.stripPrefix(start).toInt()
12921338
| _ -> 0
12931339
};
1340+
notifier.subscribed();
12941341
info.collection.subscribe(
12951342
context,
1343+
identifier,
12961344
session,
12971345
SKStore.Tick(from),
12981346
(values, tick, update) ~> {
1299-
notify(values, `${info.session}/${tick}`, update)
1347+
notifier.notify(values, `${info.session}/${tick}`, update)
13001348
},
1349+
notifier.close,
13011350
);
1351+
context.setPersistent(subId, SKStore.IntFile(session));
13021352
session
13031353
| _ -> -1
13041354
}
13051355
}
13061356

1307-
fun unsubscribe(context: mutable SKStore.Context, session: Int): void {
1308-
context.unsubscribe(session)
1357+
fun unsubscribe(
1358+
context: mutable SKStore.Context,
1359+
session: Int,
1360+
update: Bool = true,
1361+
): void {
1362+
context.sessions.maybeGet(session).each(sub -> {
1363+
(identifier, close) = sub.cmd match {
1364+
| SKStore.NWatch(identifier, _, _, close, _) -> (identifier, close)
1365+
| _ -> invariant_violation("Not manage session kind")
1366+
};
1367+
context.removePersistent(`subscription.${identifier}`);
1368+
close();
1369+
context.unsubscribe(session);
1370+
closeReactiveResource(context, identifier, update);
1371+
});
13091372
}
13101373

13111374
// WRITES

‎skipruntime-ts/native/src/Utils.sk‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,10 @@ const kResourceGarbageDir: SKStore.DirName = SKStore.DirName::create(
1818
"/sk_prv/resources/garbage/",
1919
);
2020

21+
const kResourceInstance2SubIdDir: SKStore.DirName = SKStore.DirName::create(
22+
"/sk_prv/resources/instance2subid/",
23+
);
24+
2125
const kRemoteSpecifiers: String = "SkipRuntime.RemoteSpecifiers";
2226

2327
class JSONKeyConverter() extends KeyConverter {

‎skipruntime-ts/server/src/rest.ts‎

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -121,17 +121,18 @@ export function streamingService(service: ServiceInstance): express.Express {
121121
res.sendStatus(406);
122122
return;
123123
}
124-
res.writeHead(200, {
125-
"Content-Type": "text/event-stream",
126-
Connection: "keep-alive",
127-
"Cache-Control": "no-cache",
128-
});
129-
res.flushHeaders();
130-
131124
try {
132-
const subscriptionID = service.subscribe(
133-
req.params.uuid,
134-
(update: CollectionUpdate<string, Json>) => {
125+
const uuid = req.params.uuid;
126+
const subscriptionID = service.subscribe(uuid, {
127+
subscribed: () => {
128+
res.writeHead(200, {
129+
"Content-Type": "text/event-stream",
130+
Connection: "keep-alive",
131+
"Cache-Control": "no-cache",
132+
});
133+
res.flushHeaders();
134+
},
135+
notify: (update: CollectionUpdate<string, Json>) => {
135136
if (update.isInitial) {
136137
res.write(`event: init\n`);
137138
} else {
@@ -140,15 +141,20 @@ export function streamingService(service: ServiceInstance): express.Express {
140141
res.write(`id: ${update.watermark}\n`);
141142
res.write(`data: ${JSON.stringify(update.values)}\n\n`);
142143
},
143-
// TODO: React upon resource instance deletion.
144-
);
144+
close: () => {
145+
res.end();
146+
},
147+
});
145148
req.on("close", () => {
146149
service.unsubscribe(subscriptionID);
147-
res.end();
148150
});
149151
} catch (e: unknown) {
150-
res.end();
151152
console.log(e);
153+
if (e instanceof UnknownCollectionError) {
154+
res.sendStatus(404);
155+
} else {
156+
res.sendStatus(500);
157+
}
152158
}
153159
});
154160

‎skipruntime-ts/server/src/server.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,8 +118,8 @@ export async function runService<
118118
return {
119119
close: () => {
120120
controlHttpServer.close();
121-
streamingHttpServer.close();
122121
runtime.close();
122+
streamingHttpServer.close();
123123
},
124124
};
125125
}

0 commit comments

Comments
 (0)