Feature subscriptions

di/pubsub/VERSION

@@ -0,0 +1 @@
+0.2.0

di/pubsub/init.q

@@ -1,3 +1,9 @@
+/ publisher/subscriber management - the tickerplant side of a subscription: a registry of who wants
+/ which tables (optionally sym- or condition-filtered), and the fan-out that publishes to them
+
 \l ::pubsub.q
 
-export:([subscribe;subscribestr;subscribestrfilter;publish;setsubtables;callendofperiod;callendofday;closesub;pubclear;init])
+/ version string, read from the VERSION file - fails if missing or empty
+version:first read0`:::VERSION
+
+export:([subscribe;subscribestr;subscribestrfilter;publish;setsubtables;getsubtables;callendofperiod;callendofday;closesub;pubclear;init;version])

di/pubsub/pubsub.md

@@ -49,11 +49,13 @@ publish data with/without filters. The function takes two arguments: t and x, wh
 | Function                  | Description                                                                  |
 |---------------------------|------------------------------------------------------------------------------|
 | `pubsub.setsubtables`     | Set a specified list of tables that are available for subscription.          | 
-| `pubsub.callendofday`     | Broadcast an end-of-day event to all subscribers (requires `endofday`).      |
-| `pubsub.callendofperiod`  | Broadcast an end-of-period event to all subscribers (requires `endofperiod`).|
+| `pubsub.getsubtables`     | Read the list of tables currently available for subscription - the counterpart to `setsubtables`, which replaces it. Empty until `init` has run. |
+| `pubsub.callendofday`     | Broadcast an end-of-day event to all subscribers (requires `endofday`). **Unary**: `callendofday[date]`. |
+| `pubsub.callendofperiod`  | Broadcast an end-of-period event to all subscribers (requires `endofperiod`). **Ternary**: `callendofperiod[currentperiod;nextperiod;data]`. |
 | `pubsub.closesub`         | Remove handle upon connection close.                                         | 
 | `pubsub.subclear`         | Publish tables and clear up the contents.                                    |
-| `pubsub.init`             | Initialize variables  - run before calling pub/sub functions to populate required state (e.g., tables/schemas).  |
+| `pubsub.init`             | Initialize variables  - run before calling pub/sub functions to populate required state (e.g., tables/schemas). `deps` is OPTIONAL - `pubsub.init[]` still works unchanged. An OPTIONAL `` `handlers `` key (a `di.handlers` register dict) additionally registers `closesub` as a named `.z.pc` observer, so `di.handlers.list[`.z.pc]` lists `` `pubsub `` explicitly - see Notes below.  |
+| `pubsub.version`          | Module version string, read from the `VERSION` file. `di.depcheck` resolves a dependency's minimum from here. |
 ---
 
 ### Example: 
@@ -80,6 +82,76 @@ q)pubsub.subscribestrfilter["quote";"bid>50.0";"time,sym,bid"]
 ---
 ## Notes:
 
+- **The string entry points signal on failure.** `subscribestr` and `subscribestrfilter` exist so a
+  non-kdb+ client can subscribe, and such a client cannot inspect a q result shape. A request that
+  matched **no** table therefore signals rather than returning the error message as a value that
+  merely reads like one. (The guard that previously did this could never fire: `errmsg` is built with
+  `` `$ `` so it is a symbol, and `last` of either success shape is the schema list — never the `10h`
+  string it tested for.) A *partial* match still returns, because those tables really were subscribed
+  and signalling would report failure while leaving the client registered.
+- **`.z.pc` chains, it does not replace.** This module installs a `.z.pc` handler at load so a
+  dropped connection is deregistered (`closesub`). It captures whatever already owned the event and
+  calls it afterwards. This matters: a bare `.z.pc:{closesub[x]}` silently destroyed every observer
+  another module had already registered — measured against `di.handlers`, whose registry went on
+  reporting the registration as live while it no longer fired, so the loss was invisible. The guard
+  is asserted in `test.csv` by a child process that installs a handler *before* loading this module,
+  which is the only way to observe load-time ordering.
+- This raw assignment stays in place unconditionally (it runs at module **load** time, before `init`
+  is ever called, so it cannot depend on anything `init` might later receive) — `di.pubsub` remains
+  **standalone**: `init[]` with no argument, or with a `deps` dict that omits `` `handlers ``, behaves
+  exactly as before.
+- **`init`'s OPTIONAL `` `handlers `` key closes the one real gap the chaining above left**: chaining
+  correctly preserved whatever `di.handlers` had already wired, but `di.pubsub`'s own hook stayed
+  invisible to `di.handlers`' own registry (`di.handlers.list[`.z.pc]` never named `` `pubsub ``, even
+  though it correctly fired). Passing a `` `handlers `` dep (`di.handlers`'s register dict, or the
+  whole `di.handlers` module handle) additionally registers `closesub` as a named observer, so it now
+  shows up there. The raw chain keeps running regardless — `closesub` is idempotent under a repeat
+  call (`delhandle`/`delhandlef` are both remove-if-present), so a disconnect invoking it via both the
+  raw chain and a `di.handlers` dispatch in the same tick is a harmless no-op on the second call, not
+  a double-registration bug. Asserted in `test.csv` by a child process that inits `di.handlers` then
+  `di.pubsub` with a `handlers` dep (subscribing through the **public** `subscribe` entry point, using
+  the caller's real `.z.w` handle — not an internal `reqalldict` write and a fabricated handle, which
+  would silently stop testing anything observable if that private variable's name or shape ever
+  changed), and checks all three: registry visibility, `closesub` actually running, and the
+  repeat-call idempotency.
+- **`init` re-registers with `di.handlers` on *every* call that supplies `` `handlers ``, not just the
+  first.** There is no one-shot latch: `di.handlers.register` is itself idempotent under a repeat call
+  with the same name (`registersimple`'s `upsertphase` deletes then re-adds by name), so calling it
+  again on a re-init costs nothing. A one-shot latch was tried and removed — it silently swallowed a
+  *later* `init` call's registration attempt (including one meaning to re-point `di.pubsub` at a
+  genuinely different `handlers` instance) with no error and no way to tell it never took effect,
+  confirmed by a mock `handlers` dict that counts calls to `register` across two `init` calls. Omitting
+  `` `handlers `` on a later call still leaves any earlier registration exactly as it was — `di.pubsub`
+  never infers a deregistration from a bare re-init, the same as every other injected dep here.
+
 - By default, all tables on top level of the process are available for subscription.
 - The user should define the `.u.sub` and the `.u.pub` functions within the process.
 - The module initializes with defined list of tables to subscribe to and fetches their schemas and columns for use. This is done via calling `init` function.
+
+---
+
+### End-of-day and end-of-period arity
+
+These two broadcasts deliberately have **different arities**, which looks like an inconsistency and
+is not:
+
+| function | arity | broadcast |
+|---|---|---|
+| `callendofday` | unary | `` (`endofday;date) `` |
+| `callendofperiod` | ternary | `` (`endofperiod;currentperiod;nextperiod;data) `` |
+
+`callendofperiod` matches TorQ exactly - `code/common/pubsub.q:19` sends all three, and both shipped
+subscribers (`code/rdb/endofperiod.q`, `code/wdb/writedown.q:52`) are `{[currp;nextp;data]}`.
+
+`callendofday` deliberately **diverges** from TorQ, which sends `` (`endofday;x;y) ``. That second
+argument is `processdata`; legacy's own rdb never reads it, and the shipped `.u.end` alias passes
+`()!()` for it. Subscribers here are unary to match. Do not "fix" it for symmetry with
+`callendofperiod` - doing so would turn every unary `endofday` subscriber into a projection.
+
+That projection failure is the reason this matters, and it is completely silent. A subscriber whose
+arity does not match what is broadcast is **partially applied**: q returns a projection, the body
+never runs, and nothing throws, logs, or comes back to say so. `callendofperiod` was previously
+unary, which failed both ways at once - a `callendofperiod[c;n;d]` call threw `'rank`, so a caller
+following TorQ's contract could not call it at all, while the one-argument form silently no-opped
+every ternary subscriber. Both measured; both covered by the suite, whose two assertions fail
+against the unary implementation.

di/pubsub/pubsub.q

@@ -71,14 +71,20 @@ closesub:{[h]
   delete from .z.M.reqfilteredtbl where handle=h;
   };
 
-/ define .z.pc, add bespoke actions as needed
-.z.pc:{closesub[x]};
+/ chains onto any existing .z.pc rather than replacing it - a bare .z.pc:{closesub[x]} here
+/ silently destroyed other modules' registered observers (measured). raw assignment, not a
+/ di.handlers registration, because this runs at LOAD time before init exists to depend on
+priorpc:@[value;`.z.pc;{[e] (::)}];
+.z.pc:{[w]
+  closesub[w];
+  if[not (::)~priorpc;priorpc w];
+  };
 
 / broadcast to all subscribers upon end of day, client needs to define endofday function
 callendofday:{[d](neg getallhandles[])@\:(`endofday;d)};
 
-/ broadcast to all subscribers upon end of period, client needs to define endofperiod function
-callendofperiod:{(neg getallhandles[])@\:(`endofperiod;x)};
+/ broadcast to all subscribers upon end of period - TERNARY (currentperiod;nextperiod;data), matching legacy and its real subscribers (rdb.q, wdb.q)
+callendofperiod:{[currentperiod;nextperiod;data](neg getallhandles[])@\:(`endofperiod;currentperiod;nextperiod;data)};
 
 / get table schema
 extractschema:{[table]0#value table};
@@ -103,25 +109,65 @@ pubclear:{[t]
   @[`.;;0#] each t;
   };
 
+raisenosub:{[res]
+  / internal - signal when a subscribe matched NOTHING, for the string entry points below. subscribe
+  / returns (tables;schemas), (errmsg;(tables;schemas)) for a partial match, or a bare errmsg symbol
+  / for none - a non-kdb+ caller cannot inspect a q result shape, so "nothing subscribed" must arrive
+  / as an error. the partial case still RETURNS: those tables really were subscribed
+  if[-11h=type res;'string res];
+  :res;
+  };
+
 subscribestr:{[table;syms]
   / allow non-kdb+ process to subscribe to tables with/without symbols
   res:subscribe[`$table;$[count syms;`$vs[csv;syms];`]];
-  :$[10h~type last res;'last res;res];
+  :raisenosub res;
   };
 
 subscribestrfilter:{[table;filters;columns]
   / allow non-kdb+ process to subscribe to tables with custom conditions
   res:subscribe[`$table;1!enlist `table`filts`columns!(`$table;filters;columns)];
-  :$[10h~type last res;'last res;res];
+  :raisenosub res;
   };
 
 / create a list of tables for subscription, allow users to set subtables, otherwise set to null
 setsubtables:{.z.m.subtables:$[x~`;0#x;x]};
+
+getsubtables:{[]
+  / the tables currently available for subscription - the read counterpart to setsubtables, which
+  / REPLACES the list rather than adding to it. empty until init has run, rather than signalling
+  / on an unset name
+  :@[{[] .z.m.t};::;{[e] `symbol$()}];
+  };
 setsubtables`;
 
 initialized:0b;
 
-init:{
+iscallable:{[x]
+  / internal - is x a genuinely callable value? 100 112h spans every callable form, but 101h - the
+  / generic null :: - sits INSIDE that range while being callable in no useful sense, and :: is
+  / exactly what a dep dict hands back for a missing key. see di.servers' identical helper
+  t:type x;
+  (t within 100 112h) and 101h<>t
+  };
+
+init:{[deps]
+  / deps: OPTIONAL, unlike every other DI'd module - pubsub[`init][] (unary, no args) keeps working.
+  / an OPTIONAL `handlers` key additionally registers closesub as a NAMED .z.pc observer, so
+  / di.handlers.list[`.z.pc] lists `pubsub - closing the gap where the raw chain above stayed
+  / correctly wired but invisible to the registry.
+  / the raw chain keeps running regardless of `handlers` - it fires at LOAD time, before init exists
+  / to gate it - which is safe because closesub is idempotent, so a double dispatch on one disconnect
+  / is a no-op.
+  / re-registers on EVERY call that supplies handlers - no one-shot latch. di.handlers.register is
+  / itself idempotent by name, so this costs nothing on a re-init; a latch would instead silently
+  / swallow a later call meaning to re-point pubsub at a different handlers instance
+  if[not (::)~deps;
+    if[99h<>type deps;'"di.pubsub: deps, if given, must be a dict"];
+    if[`handlers in key deps;
+      if[not iscallable deps[`handlers]`register;
+        '"di.pubsub: handlers`register must be a function [event;phase;name;priority;func] - see di.handlers"];
+      (deps[`handlers][`register])[`.z.pc;`;`pubsub;0j;closesub]]];
   .z.m.t:$[count subtables;subtables;tables[]except`reqfilteredtbl];
   .z.m.schemas:t!extractschema each t;
   .z.m.tabcols:t!cols each t;