diff --git a/example/Makefile b/example/Makefile
index ee93585..4478232 100644
--- a/example/Makefile
+++ b/example/Makefile
@@ -17,9 +17,6 @@ simplesrv_SRCS += simplesrv.cpp
 TESTPROD_HOST += mailbox
 mailbox_SRCS += mailbox.cpp
 
-TESTPROD_HOST += spam
-spam_SRCS += spam.cpp
-
 TESTPROD_HOST += ticker
 ticker_SRCS += ticker.cpp
 
diff --git a/example/spam.cpp b/example/spam.cpp
deleted file mode 100644
index b895c94..0000000
--- a/example/spam.cpp
+++ /dev/null
@@ -1,121 +0,0 @@
-/**
- * Copyright - See the COPYRIGHT that is included with this distribution.
- * pvxs is distributed subject to a Software License Agreement found
- * in file LICENSE that is included with this distribution.
- */
-
-#include <iostream>
-
-#include <epicsTime.h>
-
-#include <pvxs/source.h>
-#include <pvxs/server.h>
-#include <pvxs/nt.h>
-#include <pvxs/log.h>
-
-using namespace pvxs;
-
-DEFINE_LOGGER(app, "spam");
-
-namespace {
-
-struct SpamSource : public server::Source
-{
-    std::shared_ptr<std::set<std::string>> names;
-    Value initial;
-
-    SpamSource()
-        :names(std::make_shared<decltype (names)::element_type>())
-        ,initial(nt::NTScalar{TypeCode::UInt32}.create())
-    {}
-
-    // Source interface
-public:
-    virtual void onSearch(Search &op) override final
-    {
-        for(auto& pv :op) {
-            if(names->find(pv.name())!=names->end())
-                pv.claim();
-        }
-    }
-    virtual void onCreate(std::unique_ptr<server::ChannelControl> &&chan) override final
-    {
-        chan->onOp([this](std::unique_ptr<server::ConnectOp>&& cop) {
-            cop->onGet([](std::unique_ptr<server::ExecOp>&& op) {
-                op->error("Only monitor implemented");
-            });
-            cop->connect(initial);
-        });
-
-        chan->onSubscribe([this](std::unique_ptr<server::MonitorSetupOp>&& setup) {
-
-            std::shared_ptr<server::MonitorControlOp> sub(setup->connect(initial));
-
-            auto counter = std::make_shared<uint32_t>(0u);
-
-            auto fill = [this, sub, counter]() mutable {
-                Value update;
-                do {
-                    auto cnt = (*counter)++;
-                    update = initial.cloneEmpty();
-                    update["value"] = cnt;
-
-                    log_debug_printf(app, "%s count %u\n", sub->peerName().c_str(), unsigned(cnt));
-
-                }while(sub->tryPost(update));
-            };
-
-            sub->onHighMark(fill);
-
-            sub->onStart([fill](bool start) mutable {
-                if(start)
-                    fill();
-            });
-
-            log_info_printf(app, "%s Subscribing\n", setup->peerName().c_str());
-        });
-    }
-    virtual List onList() override final
-    {
-        return List{names, false};
-    }
-};
-
-} // namespace
-
-int main(int argc, char* argv[])
-{
-    if(argc<=1) {
-        std::cerr<<"Usage: "<<argv[0]<<" <pvname>\n";
-        return 1;
-    }
-
-    // Read $PVXS_LOG from process environment and update
-    // logging configuration.  eg.
-    //    export PVXS_LOG=*=DEBUG
-    // makes a lot of noise.
-    logger_level_set(app.name, Level::Info);
-    logger_config_env();
-
-    auto src = std::make_shared<SpamSource>();
-    src->names->insert(argv[1]);
-
-    // Build server which will serve this PV
-    // Configure using process environment.
-    server::Server serv = server::Server::fromEnv()
-            .addSource("spamsrc", src);
-
-    // (optional) Print the configuration this server is using
-    // with any auto-address list expanded.
-    std::cout<<"Effective config\n"<<serv.config();
-
-    std::cout<<"Running\n";
-
-    // Start server and run forever, or until Ctrl+c is pressed.
-    // Returns on SIGINT or SIGTERM
-    serv.run();
-
-    std::cout<<"Done\n";
-
-    return 0;
-}
diff --git a/test/Makefile b/test/Makefile
index b51f28d..94eedb6 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -125,6 +125,14 @@ TESTPROD_HOST += mcat
 mcat_SRCS += mcat.cpp
 # not a unittest
 
+TESTPROD_HOST += spam
+spam_SRCS += spam.cpp
+# not a unittest
+
+TESTPROD_HOST += eatspam
+eatspam_SRCS += eatspam.cpp
+# not a unittest
+
 TESTSCRIPTS_HOST += $(TESTS:%=%.t)
 ifdef BASE_3_15
 ifneq ($(filter $(T_A),$(CROSS_COMPILER_RUNTEST_ARCHS)),)
diff --git a/test/eatspam.cpp b/test/eatspam.cpp
new file mode 100644
index 0000000..b87a699
--- /dev/null
+++ b/test/eatspam.cpp
@@ -0,0 +1,246 @@
+/**
+ * Copyright - See the COPYRIGHT that is included with this distribution.
+ * pvxs is distributed subject to a Software License Agreement found
+ * in file LICENSE that is included with this distribution.
+ */
+
+#include <iostream>
+#include <sstream>
+#include <vector>
+#include <atomic>
+
+#include <epicsTime.h>
+#include <epicsGetopt.h>
+#include <epicsEvent.h>
+#include <epicsTime.h>
+
+#define PVXS_ENABLE_EXPERT_API
+
+#include <pvxs/client.h>
+#include <pvxs/nt.h>
+#include <pvxs/log.h>
+
+#if EPICS_VERSION_INT < VERSION_INT(7,0,1,0)
+#define epicsMonotonicGet epicsTime::getCurrent
+#endif
+
+using namespace pvxs;
+
+DEFINE_LOGGER(app, "eatspam");
+
+namespace {
+
+template<typename T>
+bool parse_as(T& out, const char *s)
+{
+    std::istringstream strm(s);
+    return (strm>>out).fail() || !strm.eof();
+}
+
+struct Counter {
+    std::string name;
+    std::vector<Value> scratch;
+    std::shared_ptr<client::Subscription> sub;
+    uint32_t prev;
+    size_t nwake = 0;
+    size_t nupdate = 0;
+    size_t nskip = 0;
+    bool first = true;
+};
+
+} // namespace
+
+int main(int argc, char* argv[])
+{
+    logger_level_set(app.name, Level::Warn);
+    logger_config_env();
+    size_t queueSize = 0;
+    int pipeline = 0; // tri-bool
+
+    int opt;
+    {
+        while((opt = getopt(argc, argv, "hpPQ:")) != -1) {
+            switch (opt) {
+            case 'h':
+                std::cerr<<"Usage: "<<argv[0]<<" [-w <sec>] pvname..."<<std::endl;
+                return 0;
+            default:
+                std::cerr<<"Unknown argument -"<<char(opt)<<std::endl;
+                return 1;
+            case 'p':
+                pipeline = -1;
+                break;
+            case 'P':
+                pipeline = 1;
+                break;
+            case 'Q':
+                if(parse_as<size_t>(queueSize, optarg)) {
+                    std::cerr<<"Invalid queueSize: "<<optarg<<std::endl;
+                    return 1;
+                }
+                break;
+            }
+        }
+    }
+
+    std::string pvRequest;
+    {
+        std::ostringstream strm;
+        strm<<"record[";
+        if(pipeline==1)
+            strm<<"pipeline=true";
+        if(pipeline==-1)
+            strm<<"pipeline=false";
+        if(queueSize) {
+            if(pipeline!=0)
+                strm<<',';
+            strm<<"queueSize="<<queueSize;
+        }
+        strm<<']';
+        pvRequest = strm.str();
+    }
+
+    auto ctxt(client::Context::fromEnv());
+
+    std::vector<Counter> ctrs(argc - optind);
+
+    MPMCFIFO<size_t> todo(ctrs.size()+1);
+
+    auto t0(epicsMonotonicGet());
+
+    for(int i=0; i<argc-optind; i++) {
+
+        auto ctr = &ctrs[i];
+        ctr->name = argv[optind+i];
+        ctr->sub = ctxt.monitor(ctr->name)
+                .pvRequest(pvRequest.c_str())
+                .event([i, &todo](client::Subscription&) {
+                    todo.emplace(1+i);
+                })
+                .exec();
+    }
+
+    SigInt sig([&todo](){
+        todo.emplace(0);
+    });
+
+    while(auto i = todo.pop()) {
+        auto& ctr = ctrs[i-1];
+
+        ctr.nwake++;
+        try {
+            bool notdone;
+            if(!!(notdone = ctr.sub->pop(ctr.scratch, queueSize))) {
+                todo.emplace(i);
+            }
+            if(ctr.scratch.empty() && notdone)
+                log_warn_printf(app, "%s pointless wakeup. %c\n", ctr.name.c_str(),
+                                notdone ? 'N' : 'D');
+            else
+                log_debug_printf(app, "%s wake with %zu, %c\n", ctr.name.c_str(), ctr.scratch.size(),
+                                 notdone ? 'N' : 'D');
+        }  catch (client::Disconnect&) {
+            ctr.first = true;
+            continue;
+        }
+
+        for(auto& v : ctr.scratch) {
+            bool gotit = false;
+            uint32_t sval;
+
+            auto val(v["value"]);
+            if(!val.isMarked()) {
+                continue;
+
+            } else if(val.type()==TypeCode::UInt32A) {
+                auto aval(val.as<shared_array<const uint32_t>>());
+                if(!aval.empty()) {
+                    sval = aval[0];
+                    gotit = true;
+                } else {
+                    continue;
+                }
+
+            } else if(val.type().code <= TypeCode::Float64 && !val.type().isarray()) {
+                sval = val.as<uint32_t>();
+                gotit = true;
+            }
+
+            if(gotit) {
+                if(!ctr.first) {
+                    auto diff = sval-ctr.prev;
+                    if(diff != 1) {
+                        log_info_printf(app, "%s skip %u -> %u, %u\n",
+                                        ctr.name.c_str(), ctr.prev, sval, diff);
+                        ctr.nskip++;
+                    }
+
+                } else {
+                    ctr.first = false;
+                    log_info_printf(app, "%s initial %u\n", ctr.name.c_str(), sval);
+                }
+                ctr.nupdate++;
+                ctr.prev = sval;
+
+            } else {
+                std::cerr<<ctr.name<<": "<<val.format().arrayLimit(10)<<"\n"
+                        "Error: no compatible \".value\" "<<val.type()<<std::endl;
+                ctr.sub->cancel();
+            }
+        }
+        ctr.scratch.clear();
+    }
+
+    // cancel subscriptions
+    for(auto& ctr : ctrs) {
+        ctr.sub->cancel();
+    }
+
+    // final stats and cleanup
+    for(auto& ctr : ctrs) {
+        client::SubscriptionStat stats;
+        ctr.sub->stats(stats);
+        std::cout<<' '<<ctr.name<<" Q used "
+                <<stats.maxQueue<<'/'<<stats.limitQueue<<" w/ "
+                <<stats.nSrvSquash<<" server overflows "
+                <<stats.nCliSquash<<" client overflows for "
+                <<ctr.nupdate<<" updates"
+                "\n";
+        ctr.sub.reset();
+    }
+
+    auto t1(epicsMonotonicGet());
+    double dT = (t1-t0) * 1e-9;
+    std::cout<<"# run time "<<dT<<" sec.\n";
+
+    for(const auto& ctr : ctrs) {
+        std::cout<<ctr.name<<" "
+                 <<(ctr.nwake/dT)<<" wakes/s "
+                 <<(ctr.nupdate/dT)<<" update/s "
+                 <<(ctr.nskip/dT)<<" skips/s "
+                 <<(double(ctr.nupdate)/ctr.nwake)<<" updates/wake "
+                 <<(double(ctr.nskip) / (ctr.nskip + ctr.nupdate) * 100.0)<<" % skip"
+                 "\n";
+    }
+
+    ctrs.clear();
+    ctxt.close();
+    ctxt = client::Context();
+
+    bool header=false;
+    int ret = 0;
+    for(const auto& p : instanceSnapshot()) {
+        if(p.second!=0) {
+            if(!header) {
+                header = true;
+                std::cout<<"Trailing refs...\n";
+                ret = 1;
+            }
+            std::cout<<" #"<<p.first<<" = "<<p.second<<"\n";
+        }
+    }
+
+    std::cout<<std::endl;
+
+    return ret;
+}
diff --git a/test/spam.cpp b/test/spam.cpp
new file mode 100644
index 0000000..de59c0b
--- /dev/null
+++ b/test/spam.cpp
@@ -0,0 +1,225 @@
+/**
+ * Copyright - See the COPYRIGHT that is included with this distribution.
+ * pvxs is distributed subject to a Software License Agreement found
+ * in file LICENSE that is included with this distribution.
+ */
+
+#include <iostream>
+#include <sstream>
+
+#include <epicsTime.h>
+#include <epicsGetopt.h>
+#include <epicsEvent.h>
+
+#include <pvxs/source.h>
+#include <pvxs/server.h>
+#include <pvxs/sharedpv.h>
+#include <pvxs/nt.h>
+#include <pvxs/log.h>
+
+using namespace pvxs;
+
+DEFINE_LOGGER(app, "spam");
+
+namespace {
+
+struct SpamSource : public server::Source
+{
+    std::shared_ptr<std::set<std::string>> names;
+    Value initial;
+    size_t nelem = 1u;
+
+    SpamSource()
+        :names(std::make_shared<decltype (names)::element_type>())
+        ,initial(nt::NTScalar{TypeCode::UInt32}.create())
+    {}
+
+    void set_nelem(size_t n)
+    {
+        initial = nt::NTScalar{TypeCode::UInt32A}.create();
+        nelem = n;
+    }
+
+    // Source interface
+    virtual void onSearch(Search &op) override final
+    {
+        for(auto& pv :op) {
+            if(names->find(pv.name())!=names->end())
+                pv.claim();
+        }
+    }
+    virtual void onCreate(std::unique_ptr<server::ChannelControl> &&chan) override final
+    {
+        chan->onOp([this](std::unique_ptr<server::ConnectOp>&& cop) {
+            cop->onGet([](std::unique_ptr<server::ExecOp>&& op) {
+                op->error("Only monitor implemented");
+            });
+            cop->connect(initial);
+        });
+
+        chan->onSubscribe([this](std::unique_ptr<server::MonitorSetupOp>&& setup) {
+
+            std::shared_ptr<server::MonitorControlOp> sub(setup->connect(initial));
+
+            uint32_t counter = 0;
+
+            auto fill = [this, sub, counter]() mutable {
+                Value update;
+                size_t nposted = 0;
+                server::MonitorStat stats{};
+
+                sub->stats(stats);
+
+                do {
+                    auto cnt = counter++;
+                    update = initial.cloneEmpty();
+                    auto value = update["value"];
+                    if(value.type().isarray()) {
+                        shared_array<uint32_t> arr(nelem, cnt);
+
+                        value = arr.freeze();
+
+                    } else {
+                        value = cnt;
+                    }
+
+                    nposted++;
+
+                }while(sub->tryPost(update));
+
+                log_debug_printf(app, "%s %s counted %zu, %zu, %zu/%zu -> %u\n",
+                                 sub->peerName().c_str(), sub->name().c_str(),
+                                 nposted, stats.window, stats.nQueue, stats.limitQueue,
+                                 unsigned(counter));
+            };
+
+            sub->onHighMark(fill);
+
+            sub->onStart([fill](bool start) mutable {
+                if(start)
+                    fill();
+            });
+
+            log_info_printf(app, "%s Subscribing\n", setup->peerName().c_str());
+        });
+    }
+    virtual List onList() override final
+    {
+        return List{names, false};
+    }
+};
+
+template<typename T>
+bool parse_as(T& out, const char *s)
+{
+    std::istringstream strm(s);
+    return (strm>>out).fail() || !strm.eof();
+}
+
+int help(int ret, const char* argv0)
+{
+    std::cerr<<
+    "Usage: "<<argv0<<" [-h] [-T <period>] [-# <count>] [-S <spam:pv:name>] ... [-H <ham:pv:name>] ...\n"
+    "\n"
+    "    -h \n"
+    ;
+    std::cerr.flush();
+    return ret;
+}
+
+} // namespace
+
+int main(int argc, char* argv[])
+{
+    // Read $PVXS_LOG from process environment and update
+    // logging configuration.  eg.
+    //    export PVXS_LOG=*=DEBUG
+    // makes a lot of noise.
+    logger_level_set(app.name, Level::Info);
+    logger_config_env();
+
+    auto spamsrc = std::make_shared<SpamSource>();
+    auto hamsrc = server::StaticSource::build();
+
+    auto hampv(server::SharedPV::buildReadonly());
+    auto hamval(nt::NTScalar{TypeCode::UInt32}.create());
+    hampv.open(hamval);
+
+    double ham_period = 1.0;
+    size_t nelem = 1;
+
+    int opt;
+    {
+        while((opt = getopt(argc, argv, "hS:H:T:#:")) != -1) {
+            switch (opt) {
+            case 'h':
+                return help(0, argv[0]);
+            default:
+                std::cerr<<"Unknown argument -"<<char(opt)<<std::endl;
+                return 1;
+            case 'S':
+                spamsrc->names->insert(optarg);
+                break;
+            case 'H':
+                hamsrc.add(optarg, hampv);
+                break;
+            case 'T':
+                if(parse_as(ham_period, optarg)) {
+                    std::cerr<<"Unable to parse period: "<<optarg<<std::endl;
+                    return 1;
+                }
+                break;
+            case '#':
+                if(parse_as(nelem, optarg)) {
+                    std::cerr<<"Unable to parse array count: "<<optarg<<std::endl;
+                    return 1;
+                }
+                break;
+            }
+        }
+    }
+
+    if(ham_period <= 0.0)
+        ham_period = 1.0;
+
+    if(nelem==0)
+        nelem = 1;
+
+    spamsrc->set_nelem(nelem);
+
+    // Build server which will serve this PV
+    // Configure using process environment.
+    server::Server serv = server::Server::fromEnv()
+            .addSource("spamsrc", spamsrc)
+            .addSource("hamsrc", hamsrc.source());
+
+    // (optional) Print the configuration this server is using
+    // with any auto-address list expanded.
+    {
+        Detailed d(std::cout, 1);
+        std::cout<<serv;
+    }
+
+    serv.start();
+    std::cout<<"Running\n";
+
+    bool run = true;
+    epicsEvent evt;
+
+    SigInt sig([&run, &evt]{
+        run = false;
+        evt.signal();
+    });
+
+    uint32_t ham_count = 0;
+    while(run) {
+        hamval["value"] = ham_count++;
+        hampv.post(hamval);
+
+        evt.wait(ham_period);
+    }
+
+    std::cout<<"Done\n";
+
+    return 0;
+}