From 53b95dd336e9ca187c8d112daa4639086be46350 Mon Sep 17 00:00:00 2001
From: Michael Davidsaver <mdavidsaver@gmail.com>
Date: Sat, 6 Oct 2018 16:34:14 -0700
Subject: [PATCH] more SharedPV rpc state tracking

handle re-open() with RPC
---
 src/server/sharedstate_channel.cpp |  1 +
 src/server/sharedstate_pv.cpp      | 21 ++++++++++++++++++---
 src/server/sharedstate_rpc.cpp     |  1 +
 src/server/sharedstateimpl.h       |  2 ++
 4 files changed, 22 insertions(+), 3 deletions(-)

diff --git a/src/server/sharedstate_channel.cpp b/src/server/sharedstate_channel.cpp
index 9379a9e..f499ee1 100644
--- a/src/server/sharedstate_channel.cpp
+++ b/src/server/sharedstate_channel.cpp
@@ -158,6 +158,7 @@ pva::ChannelRPC::shared_pointer SharedChannel::createChannelRPC(
         pvd::PVStructure::shared_pointer const & pvRequest)
 {
     std::tr1::shared_ptr<SharedRPC> ret(new SharedRPC(shared_from_this(), requester, pvRequest));
+    ret->connected = true;
     requester->channelRPCConnect(pvd::Status(), ret);
     {
         Guard G(owner->mutex);
diff --git a/src/server/sharedstate_pv.cpp b/src/server/sharedstate_pv.cpp
index 8f06029..6ccebc0 100644
--- a/src/server/sharedstate_pv.cpp
+++ b/src/server/sharedstate_pv.cpp
@@ -124,6 +124,7 @@ struct PutInfo { // oh to be able to use std::tuple ...
 void SharedPV::open(const pvd::PVStructure &value, const epics::pvData::BitSet& valid)
 {
     typedef std::vector<PutInfo> xputs_t;
+    typedef std::vector<std::tr1::shared_ptr<SharedRPC> > xrpcs_t;
     typedef std::vector<std::tr1::shared_ptr<pva::MonitorFIFO> > xmonitors_t;
     typedef std::vector<std::tr1::shared_ptr<pva::GetFieldRequester> > xgetfields_t;
 
@@ -132,6 +133,7 @@ void SharedPV::open(const pvd::PVStructure &value, const epics::pvData::BitSet&
     newvalue->copyUnchecked(value, valid);
 
     xputs_t p_put;
+    xrpcs_t p_rpc;
     xmonitors_t p_monitor;
     xgetfields_t p_getfield;
     {
@@ -141,6 +143,7 @@ void SharedPV::open(const pvd::PVStructure &value, const epics::pvData::BitSet&
             throw std::logic_error("Already open()");
 
         p_put.reserve(puts.size());
+        p_rpc.reserve(rpcs.size());
         p_monitor.reserve(monitors.size());
         p_getfield.reserve(getfields.size());
 
@@ -161,6 +164,12 @@ void SharedPV::open(const pvd::PVStructure &value, const epics::pvData::BitSet&
                 //racing destruction
             }
         }
+        FOR_EACH(rpcs_t::const_iterator, it, end, rpcs) {
+            if((*it)->connected) continue;
+            try {
+                p_rpc.push_back((*it)->shared_from_this());
+            }catch(std::tr1::bad_weak_ptr&) {}
+        }
         FOR_EACH(monitors_t::const_iterator, it, end, monitors) {
             try {
                 (*it)->open(newtype);
@@ -184,6 +193,10 @@ void SharedPV::open(const pvd::PVStructure &value, const epics::pvData::BitSet&
             requester->channelPutConnect(it->status, it->put, it->type);
         }
     }
+    FOR_EACH(xrpcs_t::iterator, it, end, p_rpc) {
+        SharedRPC::requester_type::shared_pointer requester((*it)->requester.lock());
+        if(requester) requester->channelRPCConnect(pvd::Status(), *it);
+    }
     FOR_EACH(xmonitors_t::iterator, it, end, p_monitor) {
         (*it)->notify();
     }
@@ -219,6 +232,11 @@ void SharedPV::close(bool destroy)
     {
         Guard I(mutex);
 
+        FOR_EACH(rpcs_t::const_iterator, it, end, rpcs) {
+            if(!(*it)->connected) continue;
+            p_rpc.push_back((*it)->requester.lock());
+        }
+
         if(type) {
 
             p_put.reserve(puts.size());
@@ -230,9 +248,6 @@ void SharedPV::close(bool destroy)
                 (*it)->mapper.reset();
                 p_put.push_back((*it)->requester.lock());
             }
-            FOR_EACH(rpcs_t::const_iterator, it, end, rpcs) {
-                p_rpc.push_back((*it)->requester.lock());
-            }
             FOR_EACH(monitors_t::const_iterator, it, end, monitors) {
                 (*it)->close();
                 try {
diff --git a/src/server/sharedstate_rpc.cpp b/src/server/sharedstate_rpc.cpp
index 3f192f6..336d083 100644
--- a/src/server/sharedstate_rpc.cpp
+++ b/src/server/sharedstate_rpc.cpp
@@ -84,6 +84,7 @@ SharedRPC::SharedRPC(const std::tr1::shared_ptr<SharedChannel>& channel,
     :channel(channel)
     ,requester(requester)
     ,pvRequest(pvRequest)
+    ,connected(false)
 {
     REFTRACE_INCREMENT(num_instances);
 }
diff --git a/src/server/sharedstateimpl.h b/src/server/sharedstateimpl.h
index de12eca..8fbf362 100644
--- a/src/server/sharedstateimpl.h
+++ b/src/server/sharedstateimpl.h
@@ -107,6 +107,8 @@ struct SharedRPC : public pva::ChannelRPC,
 
     static size_t num_instances;
 
+    bool connected; // have I called requester->channelRPCConnect(Ok) ?
+
     SharedRPC(const std::tr1::shared_ptr<SharedChannel>& channel,
               const requester_type::shared_pointer& requester,
               const pvd::PVStructure::const_shared_pointer &pvRequest);