Feature subscriptions

di/pubsub/VERSION

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

di/pubsub/init.q

@@ -1,3 +1,13 @@
+/ 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])
+/ module version, read from the VERSION file rather than hardcoded, so a release bump touches one
+/ plain-text file. read module-relative at load (`:::` resolves to di/pubsub) and BEFORE the export
+/ line, since export:([...]) evaluates each name. NB `version` must STAY in the export: di.depcheck
+/ resolves a dependency's minimum version from the export dict, and reports "exports no version" -
+/ failing the dependency check - for any module that omits it
+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,36 @@ closesub:{[h]
   delete from .z.M.reqfilteredtbl where handle=h;
   };
 
-/ define .z.pc, add bespoke actions as needed
-.z.pc:{closesub[x]};
+/ define .z.pc, add bespoke actions as needed.
+/ CHAINS onto whatever already owns .z.pc rather than replacing it. a bare .z.pc:{closesub[x]} here
+/ silently destroyed every observer another module had already registered - measured: with a
+/ di.handlers registration in place first, loading this module stopped it firing while di.handlers
+/ went on listing it as registered, so the failure was invisible from the registry.
+/ this stays a raw assignment (not a di.handlers registration) at LOAD time, because di.pubsub is
+/ STANDALONE - it takes no REQUIRED injected dependency, and this code runs before init is ever
+/ called, so it cannot depend on anything init might later receive. it remains active even when a
+/ caller does pass a handlers dep to init (see init, below) - it is di.pubsub's own always-on safety
+/ net, chaining correctly either way
+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, client needs to define endofperiod function.
+/ TERNARY, matching legacy: TorQ's code/common/pubsub.q:19 broadcasts (`endofperiod;x;y;z) and both
+/ of its subscribers (code/rdb/endofperiod.q, code/wdb/writedown.q:52) are {[currp;nextp;data]}.
+/ it was unary, which failed two ways at once (both measured): callendofperiod[c;n;d] threw 'rank, so
+/ a caller following that contract could not call it at all, and the one-argument form left a ternary
+/ subscriber as a PROJECTION - the body never ran, and nothing threw, logged or was returned to say so.
+/ same defect class as the callendofday bug PR #118 fixed.
+/ NB callendofday stays UNARY on purpose. TorQ's producer sends (`endofday;x;y), but its second
+/ argument is processdata, which legacy's own rdb never reads and the shipped .u.end alias passes
+/ ()!() for - di.rdb's endofday is unary to match. Do not "fix" that one for symmetry with this
+callendofperiod:{[currentperiod;nextperiod;data](neg getallhandles[])@\:(`endofperiod;currentperiod;nextperiod;data)};
 
 / get table schema
 extractschema:{[table]0#value table};
@@ -103,25 +125,85 @@ pubclear:{[t]
   @[`.;;0#] each t;
   };
 
+raisenosub:{[res]
+  / internal - signal when a subscribe matched NOTHING, for the string entry points below.
+  / subscribe returns one of three shapes: (tables;schemas) when every requested table exists,
+  / (errmsg;(tables;schemas)) when only some do, or a bare errmsg SYMBOL when none do. the string
+  / entry points exist for non-kdb+ clients, which cannot inspect a q result shape - so a request
+  / that subscribed to nothing has to arrive as an error, not as a value that merely reads like one.
+  / the partial case deliberately still RETURNS: those tables really were subscribed, and signalling
+  / would tell the caller it failed while leaving it registered.
+  / NB this replaces a guard (10h~type last res) that could never fire - errmsg is built with `$ so it
+  / is a symbol, and `last` of either success shape is the schema list, never a 10h string
+  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 - a consumer that needs to ADD to the publish set has no other way to learn the
+  / current one, and reaching into module state from outside is not an interface.
+  / empty until init has run, rather than signalling on an unset name
+  / read .z.m.t EXPLICITLY - a bare t would resolve to the same module state, but the explicit form is
+  / the one qlint accepts and matches how every other module reads its own state
+  :@[{[] .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 DI'd module's init, di.pubsub has no REQUIRED injected dependency,
+  / so pubsub[`init][] (the original, still-supported call shape) keeps working: a unary function
+  / called with no args binds deps to the generic null ::, same as always.
+  / .
+  / an OPTIONAL `handlers` key (di.handlers' register dict) ADDITIONALLY registers closesub as a
+  / named observer on .z.pc, so di.handlers.list[`.z.pc] lists `pubsub explicitly - closing the
+  / actual gap the raw .z.pc block (above) used to leave: a di.handlers registration made elsewhere
+  / stayed correctly chained, but pubsub's own hook stayed invisible to the registry itself.
+  / the raw chaining block keeps running regardless of whether handlers is passed here - it fires at
+  / module LOAD time, before init is ever called, so it cannot be conditionally skipped from inside
+  / init. that is harmless, not a double-registration bug: 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 handlers dispatch in the same tick is a no-op on the second call.
+  / .
+  / re-registers on EVERY init call that supplies handlers, not just the first - there is no one-shot
+  / latch here. di.handlers.register is itself idempotent under a repeat call with the same name
+  / (registersimple's upsertphase deletes then re-adds by name, see di.handlers), so calling it again
+  / on a re-init costs nothing; a one-shot latch, by contrast, would silently swallow a LATER init
+  / call's registration attempt - including one meaning to re-point pubsub at a genuinely different
+  / handlers instance - with no error and no way to tell it never took effect. omitting `handlers on a
+  / later call 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
+  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;