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.version`          | Module version string, read from the `VERSION` file. `di.depcheck` resolves a dependency's minimum from here. |
 ---
 
 ### Example: 
@@ -80,6 +82,52 @@ 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.
+- It stays a raw assignment rather than a `di.handlers` registration because the modularisation plan
+  classifies `di.pubsub` as **standalone** — it takes no injected dependencies, so reaching
+  `di.handlers` would contradict its own tier.
+
 - 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,34 @@ 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 rather than a di.handlers registration because the modularisation
+/ plan classifies di.pubsub as STANDALONE - it takes no injected dependencies, so it cannot reach
+/ di.handlers without contradicting its own tier
+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,20 +123,44 @@ 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;