Add APIs to persist and restore state of subscriptions
This commit enables a set of APIs to persist and restore the
subscription state. Such functionality may be used when the
Weave process needs to shutdown and restart in an orderly fashion
while preserving the illusion that the subscription remains
unchanged. In order to successfully accomplish this task, the
commit introduces the functionality to:
* serialize SubscriptionClient and SubscriptionHandler data.
* load SubscriptionClient and SubscriptionHandler from
serialized data and integrate it into the subscription engine
mechanisms.
diff --git a/src/lib/core/WeaveConfig.h b/src/lib/core/WeaveConfig.h
index be5eb53..132ba47 100644
--- a/src/lib/core/WeaveConfig.h
+++ b/src/lib/core/WeaveConfig.h
@@ -1881,6 +1881,22 @@
#define WEAVE_CONFIG_ENABLE_CONDITION_LOGGING 0
#endif // WEAVE_CONFIG_ENABLE_CONDITION_LOGGING
+/**
+ * @def WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ *
+ * @brief
+ * If set to (1), use of the persistent subscription state
+ * implementation is enabled. Default value is (0) or disabled.
+ *
+ * @note
+ * Enabling this profile allows applications using Weave to
+ * persist the mutual subscription states between device and service
+ * across device reboots.
+ *
+ */
+#ifndef WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+#define WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE 0
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
/**
* @def WEAVE_CONFIG_ENABLE_SERVICE_DIRECTORY
diff --git a/src/lib/profiles/data-management/Current/NotificationEngine.cpp b/src/lib/profiles/data-management/Current/NotificationEngine.cpp
index ae87b2b..6d1cdc9 100644
--- a/src/lib/profiles/data-management/Current/NotificationEngine.cpp
+++ b/src/lib/profiles/data-management/Current/NotificationEngine.cpp
@@ -1152,6 +1152,9 @@
size_t i = static_cast<size_t>(iterator - kImportanceType_First);
ImportanceType importance = (ImportanceType) iterator;
logger.NotifyEventsDelivered(importance, aSubHandler->mSelfVendedEvents[i] - 1, aSubHandler->GetPeerNodeId());
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ aSubHandler->UpdateDeliveredEvents(importance);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
}
}
diff --git a/src/lib/profiles/data-management/Current/SubscriptionClient.cpp b/src/lib/profiles/data-management/Current/SubscriptionClient.cpp
index 99942a3..a432120 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionClient.cpp
+++ b/src/lib/profiles/data-management/Current/SubscriptionClient.cpp
@@ -163,6 +163,153 @@
return err;
}
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+// load subscription id and liveness timeout
+// AddRef to Binding
+// store pointers to binding and delegate
+// move to kState_SubscriptionEstablished_Idle
+WEAVE_ERROR SubscriptionClient::LoadFromPersistedState(Binding * const apBinding, void * const apAppState, EventCallback const aEventCallback,
+ const TraitCatalogBase<TraitDataSink> * const apCatalog,
+ const uint32_t aInactivityTimeoutDuringSubscribingMsec, IWeaveWDMMutex * aUpdateMutex,
+ TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+ WeaveLogIfFalse(0 == mRefCount);
+
+ // AddRef for the duration of this method
+ _AddRef();
+
+ // add reference to the binding
+ apBinding->AddRef();
+
+ // set subscription id and liveness timeout
+ err = LoadSubscriptionState(reader);
+ SuccessOrExit(err);
+
+ // make a copy of the pointers
+ mBinding = apBinding;
+ mAppState = apAppState;
+ mEventCallback = aEventCallback;
+
+ // Set the protocol callback on the binding object so that the SubscriptionClient gets
+ // notified of changes in the binding's state.
+ mBinding->SetProtocolLayerCallback(BindingEventCallback, this);
+
+ err = _PrepareBinding();
+ SuccessOrExit(err);
+
+ if (NULL == apCatalog)
+ {
+ mDataSinkCatalog = NULL;
+ }
+ else
+ {
+ mDataSinkCatalog = const_cast <TraitCatalogBase<TraitDataSink>*>(apCatalog);
+ }
+
+ mInactivityTimeoutDuringSubscribingMsec = aInactivityTimeoutDuringSubscribingMsec;
+
+
+#if WEAVE_CONFIG_ENABLE_WDM_UPDATE
+ mUpdateMutex = aUpdateMutex;
+ mUpdateInFlight = false;
+ mMaxUpdateSize = 0;
+
+ err = mUpdateClient.Init(mBinding, this, UpdateEventCallback);
+ SuccessOrExit(err);
+
+ ConfigureUpdatableSinks();
+
+#endif // WEAVE_CONFIG_ENABLE_WDM_UPDATE
+
+ // Hold this instance until we clear the protocol state machine
+ _AddRef();
+ MoveToState(kState_SubscriptionEstablished_Idle);
+
+ WeaveLogDetail(DataManagement, "Client[%u] [%5.5s] %s Ref(%d)", SubscriptionEngine::GetInstance()->GetClientId(this),
+ GetStateStr(), __func__, mRefCount);
+
+ {
+ InEventParam inParam;
+ OutEventParam outParam;
+
+ // Emit an OnSubscriptionActivity event to the application.
+ inParam.mSubscriptionActivity.mClient = this;
+ mEventCallback(mAppState, kEvent_OnSubscriptionActivity, inParam, outParam);
+
+ // Emit an OnSubscriptionEstablished event to the application.
+ // Note that it's allowed to cancel or even abandon this subscription right inside this callback.
+ inParam.mSubscriptionEstablished.mSubscriptionId = mSubscriptionId;
+ inParam.mSubscriptionEstablished.mClient = this;
+ mEventCallback(mAppState, kEvent_OnSubscriptionEstablished, inParam, outParam);
+ }
+
+exit:
+ _Release();
+ return err;
+}
+
+WEAVE_ERROR SubscriptionClient::SerializeSubscriptionState(TLVWriter & writer)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ TLV::TLVType container;
+
+ err = writer.StartContainer(TLV::AnonymousTag, TLV::kTLVType_Structure, container);
+ SuccessOrExit(err);
+
+ err = writer.Put(TLV::ContextTag(kTag_PersistSubscriptionClient_SubscriptionId), mSubscriptionId);
+ SuccessOrExit(err);
+ err = writer.Put(TLV::ContextTag(kTag_PersistSubscriptionClient_LivenessTimeoutMsec), mLivenessTimeoutMsec);
+ SuccessOrExit(err);
+
+ err = writer.EndContainer(container);
+ SuccessOrExit(err);
+
+exit:
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Serialize persistent subscription data for subscription client error: %d", err);
+ }
+ return err;
+}
+
+WEAVE_ERROR SubscriptionClient::LoadSubscriptionState(TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ TLV::TLVType container;
+
+ err = reader.Next(TLV::kTLVType_Structure, TLV::AnonymousTag);
+ SuccessOrExit(err);
+ err = reader.EnterContainer(container);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistSubscriptionClient_SubscriptionId));
+ SuccessOrExit(err);
+ err = reader.Get(mSubscriptionId);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistSubscriptionClient_LivenessTimeoutMsec));
+ SuccessOrExit(err);
+ err = reader.Get(mLivenessTimeoutMsec);
+ SuccessOrExit(err);
+
+ err = reader.ExitContainer(container);
+ SuccessOrExit(err);
+
+
+exit:
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Load persistent subscription data for subscription client error: %d", err);
+ }
+ return err;
+}
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
#if WEAVE_DETAIL_LOGGING
const char * SubscriptionClient::GetStateStr() const
{
diff --git a/src/lib/profiles/data-management/Current/SubscriptionClient.h b/src/lib/profiles/data-management/Current/SubscriptionClient.h
index c718532..849e692 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionClient.h
+++ b/src/lib/profiles/data-management/Current/SubscriptionClient.h
@@ -448,6 +448,18 @@
kConfig_CounterSubscriber /**< Start a "counter subscription" */
};
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ /**
+ * @brief
+ * Tags for the persistent subscription data
+ */
+ enum
+ {
+ kTag_PersistSubscriptionClient_SubscriptionId = 1,
+ kTag_PersistSubscriptionClient_LivenessTimeoutMsec = 2
+ };
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
bool IsInitiator() { return mConfig == kConfig_Initiator; }
bool IsCounterSubscriber() { return mConfig == kConfig_CounterSubscriber; }
bool ShouldSubscribe() { return mConfig > kConfig_Down; }
@@ -493,6 +505,20 @@
const TraitCatalogBase<TraitDataSink> * const apCatalog,
const uint32_t aInactivityTimeoutDuringSubscribingMsec, IWeaveWDMMutex * aUpdateMutex);
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ // load subscription id and liveness timeout
+ // AddRef to Binding
+ // store pointers to binding and delegate
+ // move to kState_SubscriptionEstablished_Idle
+ WEAVE_ERROR LoadFromPersistedState(Binding * const apBinding, void * const apAppState, EventCallback const aEventCallback,
+ const TraitCatalogBase<TraitDataSink> * const apCatalog,
+ const uint32_t aInactivityTimeoutDuringSubscribingMsec, IWeaveWDMMutex * aUpdateMutex,
+ TLVReader & reader);
+
+ WEAVE_ERROR SerializeSubscriptionState(TLVWriter & writer);
+ WEAVE_ERROR LoadSubscriptionState(TLVReader & reader);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
void _InitiateSubscription(void);
WEAVE_ERROR SendSubscribeRequest(void);
diff --git a/src/lib/profiles/data-management/Current/SubscriptionEngine.cpp b/src/lib/profiles/data-management/Current/SubscriptionEngine.cpp
index c2f4233..00f028e 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionEngine.cpp
+++ b/src/lib/profiles/data-management/Current/SubscriptionEngine.cpp
@@ -283,6 +283,104 @@
return NewClient(appClient, apBinding, apAppState, aEventCallback, apCatalog, aInactivityTimeoutDuringSubscribingMsec, NULL);
}
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+WEAVE_ERROR SubscriptionEngine::NewClientFromPersistedState(SubscriptionClient ** const appClient, Binding * const apBinding, void * const apAppState,
+ SubscriptionClient::EventCallback const aEventCallback,
+ const TraitCatalogBase<TraitDataSink> * const apCatalog,
+ const uint32_t aInactivityTimeoutDuringSubscribingMsec, TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_ERROR_NO_MEMORY;
+
+#if WEAVE_CONFIG_ENABLE_WDM_UPDATE
+ uint32_t maxSize = WDM_MAX_UPDATE_SIZE;
+#endif // WEAVE_CONFIG_ENABLE_WDM_UPDATE
+
+ WEAVE_FAULT_INJECT(FaultInjection::kFault_WDM_SubscriptionClientNew, ExitNow());
+
+ *appClient = NULL;
+
+ for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i)
+ {
+ if (SubscriptionClient::kState_Free == mClients[i].mCurrentState)
+ {
+ *appClient = &mClients[i];
+ err =
+ (*appClient)
+ ->LoadFromPersistedState(apBinding, apAppState, aEventCallback, apCatalog, aInactivityTimeoutDuringSubscribingMsec, NULL, reader);
+
+ if (WEAVE_NO_ERROR != err)
+ {
+ *appClient = NULL;
+ ExitNow();
+ }
+#if WEAVE_CONFIG_ENABLE_WDM_UPDATE
+ mClients[i].SetMaxUpdateSize(maxSize);
+#endif // WEAVE_CONFIG_ENABLE_WDM_UPDATE
+ SYSTEM_STATS_INCREMENT(nl::Weave::System::Stats::kWDM_NumSubscriptionClients);
+ break;
+ }
+ }
+
+exit:
+ if (WEAVE_NO_ERROR != err)
+ {
+ WeaveLogError(DataManagement, "Load persistent subscription client failed with error: %d", err);
+ }
+ return err;
+}
+
+WEAVE_ERROR SubscriptionEngine::NewSubscriptionHandlerFromPersistedState(Binding * const apBinding, void * const apAppState,
+ SubscriptionHandler::EventCallback const aEventCallback, TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+ SubscriptionHandler * handler = NULL;
+
+ err = NewSubscriptionHandler(&handler);
+ SuccessOrExit(err);
+
+ handler->SetMaxNotificationSize(WDM_MAX_NOTIFICATION_SIZE);
+ err = handler->LoadFromPersistedState(apBinding, apAppState, aEventCallback, reader);
+ SuccessOrExit(err);
+
+exit:
+ WeaveLogFunctError(err);
+
+ return err;
+}
+
+WEAVE_ERROR SubscriptionEngine::SaveClient(uint64_t aPeerNodeID, TLVWriter &aWriter)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+ SubscriptionClient * client = NULL;
+
+ client = FindEstablishedIdleClient(aPeerNodeID);
+ VerifyOrExit(client != NULL, err = WEAVE_ERROR_INCORRECT_STATE);
+
+ err = client->SerializeSubscriptionState(aWriter);
+ SuccessOrExit(err);
+
+exit:
+
+ return err;
+}
+
+WEAVE_ERROR SubscriptionEngine::SaveSubscriptionHandler(uint64_t aPeerNodeID, TLVWriter &aWriter)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+ SubscriptionHandler * handler = NULL;
+
+ handler = FindEstablishedIdleHandler(aPeerNodeID);
+ VerifyOrExit(handler != NULL, err = WEAVE_ERROR_INCORRECT_STATE);
+
+ err = handler->SerializeSubscriptionState(aWriter);
+ SuccessOrExit(err);
+
+exit:
+
+ return err;
+}
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
/**
* Reply to a request with a StatuReport message.
*
@@ -800,6 +898,27 @@
return result;
}
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+SubscriptionClient * SubscriptionEngine::FindEstablishedIdleClient(const uint64_t aPeerNodeId)
+{
+ SubscriptionClient * result = NULL;
+
+ for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i)
+ {
+ if (mClients[i].mCurrentState == SubscriptionClient::kState_SubscriptionEstablished_Idle)
+ {
+ if (aPeerNodeId == mClients[i].mBinding->GetPeerNodeId())
+ {
+ result = &mClients[i];
+ break;
+ }
+ }
+ }
+
+ return result;
+}
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
bool SubscriptionEngine::UpdateClientLiveness(const uint64_t aPeerNodeId, const uint64_t aSubscriptionId, const bool aKill)
{
WEAVE_ERROR err = WEAVE_NO_ERROR;
@@ -919,6 +1038,27 @@
return result;
}
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+SubscriptionHandler * SubscriptionEngine::FindEstablishedIdleHandler(const uint64_t aPeerNodeId)
+{
+ SubscriptionHandler * result = NULL;
+
+ for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i)
+ {
+ if (mHandlers[i].mCurrentState == SubscriptionHandler::kState_SubscriptionEstablished_Idle)
+ {
+ if (aPeerNodeId == mHandlers[i].GetPeerNodeId())
+ {
+ result = &mHandlers[i];
+ break;
+ }
+ }
+ }
+
+ return result;
+}
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
WEAVE_ERROR SubscriptionEngine::GetMinEventLogPosition(size_t & outLogPosition) const
{
WEAVE_ERROR err = WEAVE_NO_ERROR;
diff --git a/src/lib/profiles/data-management/Current/SubscriptionEngine.h b/src/lib/profiles/data-management/Current/SubscriptionEngine.h
index 3e3652e..633fe96 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionEngine.h
+++ b/src/lib/profiles/data-management/Current/SubscriptionEngine.h
@@ -300,8 +300,34 @@
const TraitCatalogBase<TraitDataSink> * const apCatalog,
const uint32_t aInactivityTimeoutDuringSubscribingMsec, IWeaveWDMMutex * aUpdateMutex);
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ /**
+ * @brief This is the default event handler to be called by application layer for any ignored or unrecognized event
+ *
+ * @param[in] appClient A pointer to pointer for the new subscription client object
+ * @param[in] apBinding A pointer to Binding to be used for this subscription client
+ * @param[in] apAppState A pointer to application layer supplied state object
+ * @param[in] aEventCallback A function pointer for event call back
+ * @param[in] apCatalog A pointer to data sink catalog object
+ * @param[in] aTimeoutMsecBeforeSubscribeResponse Max number of milliseconds before subscribe
+ * response must be received after subscribe request is sent
+ * @param[in] reader A reference to TLVReader to load subscription client
+ */
+ WEAVE_ERROR NewClientFromPersistedState(SubscriptionClient ** const appClient, Binding * const apBinding, void * const apAppState,
+ SubscriptionClient::EventCallback const aEventCallback,
+ const TraitCatalogBase<TraitDataSink> * const apCatalog,
+ const uint32_t aInactivityTimeoutDuringSubscribingMsec, TLVReader & reader);
+ WEAVE_ERROR SaveClient(uint64_t aPeerNodeID, TLVWriter &aWriter);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
WEAVE_ERROR NewSubscriptionHandler(SubscriptionHandler ** const subHandler);
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ WEAVE_ERROR NewSubscriptionHandlerFromPersistedState(Binding * const apBinding, void * const apAppState,
+ SubscriptionHandler::EventCallback const aEventCallback, TLVReader & reader);
+ WEAVE_ERROR SaveSubscriptionHandler(uint64_t aPeerNodeID, TLVWriter &aWriter);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
uint16_t GetClientId(const SubscriptionClient * const apClient) const;
SubscriptionClient * FindClient(const uint64_t aPeerNodeId, const uint64_t aSubscriptionId);
@@ -536,6 +562,11 @@
#endif // WDM_PUBLISHER_ENABLE_CUSTOM_COMMANDS
#endif // WDM_ENABLE_SUBSCRIPTION_PUBLISHER
+
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ SubscriptionClient * FindEstablishedIdleClient(const uint64_t aPeerNodeId);
+ SubscriptionHandler * FindEstablishedIdleHandler(const uint64_t aPeerNodeId);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
};
}; // namespace WeaveMakeManagedNamespaceIdentifier(DataManagement, kWeaveManagedNamespaceDesignation_Current)
diff --git a/src/lib/profiles/data-management/Current/SubscriptionHandler.cpp b/src/lib/profiles/data-management/Current/SubscriptionHandler.cpp
index 12edb52..273f372 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionHandler.cpp
+++ b/src/lib/profiles/data-management/Current/SubscriptionHandler.cpp
@@ -75,6 +75,9 @@
memset(mSelfVendedEvents, 0, sizeof(mSelfVendedEvents));
memset(mLastScheduledEventId, 0, sizeof(mLastScheduledEventId));
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ memset(mDeliveredEvents, 0, sizeof(mDeliveredEvents));
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
}
WEAVE_ERROR SubscriptionHandler::AcceptSubscribeRequest(const uint32_t aLivenessTimeoutSec)
@@ -510,8 +513,10 @@
// mSelfVendedEvents should point to the next event ID that
// we publish. Otherwise, we would publish an event that
// the subscriber already received.
- mSelfVendedEvents[static_cast<uint32_t>(importance) - static_cast<uint32_t>(kImportanceType_First)] =
- static_cast<event_id_t>(eventId + 1);
+ uint32_t i = static_cast<uint32_t>(importance) - static_cast<uint32_t>(kImportanceType_First);
+ WeaveLogProgress(DataManagement, "Update mSelfVendedEvents[%d] from %d to %d using service data",
+ i, mSelfVendedEvents[i], eventId + 1);
+ mSelfVendedEvents[i] = static_cast<event_id_t>(eventId + 1);
}
else
{
@@ -635,6 +640,14 @@
return err;
}
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+void SubscriptionHandler::UpdateDeliveredEvents(ImportanceType importance)
+{
+ uint32_t i = static_cast<uint32_t>(importance) - static_cast<uint32_t>(kImportanceType_First);
+ mDeliveredEvents[i] = mSelfVendedEvents[i] - 1;
+}
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
void SubscriptionHandler::InitWithIncomingRequest(Binding * const aBinding, const uint64_t aRandomNumber,
nl::Weave::ExchangeContext * aEC, const nl::Inet::IPPacketInfo * aPktInfo,
const nl::Weave::WeaveMessageInfo * aMsgInfo, PacketBuffer * aPayload)
@@ -1753,6 +1766,402 @@
mMaxNotificationSize = aMaxSize;
}
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+WEAVE_ERROR SubscriptionHandler::LoadFromPersistedState(Binding * const apBinding, void * const apAppState, EventCallback const aEventCallback, TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)", SubscriptionEngine::GetInstance()->GetHandlerId(this),
+ GetStateStr(), __func__, mRefCount);
+
+ WeaveLogIfFalse(0 == mRefCount);
+
+ // AddRef for the duration of this method
+ _AddRef();
+
+ mBinding = apBinding;
+ mAppState = apAppState;
+ mEventCallback = aEventCallback;
+
+ // Capture the binding and arrange to receive event callbacks.
+ mBinding->AddRef();
+ mBinding->SetProtocolLayerCallback(BindingEventCallback, this);
+
+ mBytesOffloaded = 0;
+
+ err = LoadSubscriptionState(reader);
+ SuccessOrExit(err);
+
+ // Hold this instance until we clear the protocol state machine
+ _AddRef();
+ MoveToState(kState_SubscriptionEstablished_Idle);
+
+ {
+ InEventParam inParam;
+ OutEventParam outParam;
+ inParam.mSubscriptionEstablished.mSubscriptionId = mSubscriptionId;
+ inParam.mSubscriptionEstablished.mHandler = this;
+
+ // Note we could be aborted in this callback
+ mEventCallback(mAppState, kEvent_OnSubscriptionEstablished, inParam, outParam);
+ }
+
+ err = ReplaceExchangeContext();
+ SuccessOrExit(err);
+
+ SubscriptionEngine::GetInstance()->GetNotificationEngine()->Run();
+
+exit:
+ WeaveLogFunctError(err);
+
+ // Run termination if subscription is marked established
+ if (WEAVE_NO_ERROR != err && mCurrentState == kState_SubscriptionEstablished_Idle)
+ {
+ TerminateSubscription(err, NULL, false);
+ }
+
+ _Release();
+
+ return err;
+}
+
+WEAVE_ERROR SubscriptionHandler::LoadSubscriptionState(TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ TLV::TLVType container;
+
+ err = reader.Next(TLV::kTLVType_Structure, TLV::AnonymousTag);
+ SuccessOrExit(err);
+ err = reader.EnterContainer(container);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistSubscriptionHandler_PeerNodeId));
+ SuccessOrExit(err);
+ err = reader.Get(mPeerNodeId);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistSubscriptionHandler_SubscriptionId));
+ SuccessOrExit(err);
+ err = reader.Get(mSubscriptionId);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistSubscriptionHandler_LivenessTimeoutMsec));
+ SuccessOrExit(err);
+ err = reader.Get(mLivenessTimeoutMsec);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_Boolean, TLV::ContextTag(kTag_PersistSubscriptionHandler_IsInitiator));
+ SuccessOrExit(err);
+ err = reader.Get(mIsInitiator);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_Boolean, TLV::ContextTag(kTag_PersistSubscriptionHandler_SubscribeToAllEvents));
+ SuccessOrExit(err);
+ err = reader.Get(mSubscribeToAllEvents);
+ SuccessOrExit(err);
+
+ err = LoadTraitInstances(reader);
+ SuccessOrExit(err);
+
+ err = LoadDeliveredEvents(reader);
+ SuccessOrExit(err);
+
+ err = reader.ExitContainer(container);
+ SuccessOrExit(err);
+
+ // Load SelfVendedEvents from DeliveredEvents
+ for (int importance = kImportanceType_First; importance <= kImportanceType_Last; importance++)
+ {
+ uint32_t i = static_cast<uint32_t>(importance) - static_cast<uint32_t>(kImportanceType_First);
+ WeaveLogProgress(DataManagement, "Update mSelfVendedEvents[%d] from %d to %d using persisted data",
+ i, mSelfVendedEvents[i], 1 + mDeliveredEvents[i]);
+ mSelfVendedEvents[i] = 1 + mDeliveredEvents[i];
+ }
+
+exit:
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Load persistent subscription data for subscription handler error: %d", err);
+ }
+ return err;
+}
+
+
+WEAVE_ERROR SubscriptionHandler::LoadTraitInstances(TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ Platform::CriticalSectionEnter();
+
+ TLV::TLVType traitInstanceArrayContainer;
+
+ err = reader.Next(TLV::kTLVType_Array, TLV::ContextTag(kTag_PersistTraitInstances));
+ SuccessOrExit(err);
+ err = reader.EnterContainer(traitInstanceArrayContainer);
+ SuccessOrExit(err);
+
+ while ((err = reader.Next(TLV::kTLVType_Structure, TLV::AnonymousTag)) == WEAVE_NO_ERROR)
+ {
+ TLV::TLVType traitInstanceContainer;
+ TraitDataSource * dataSource;
+ TraitInstanceInfo * traitInstance = NULL;
+ TraitDataHandle traitDataHandle;
+ uint16_t requestedVersion;
+ bool isDirty;
+
+ err = reader.EnterContainer(traitInstanceContainer);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistTraitInstances_TraitDataHandle));
+ SuccessOrExit(err);
+ err = reader.Get(traitDataHandle);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistTraitInstances_RequestedVersion));
+ SuccessOrExit(err);
+ err = reader.Get(requestedVersion);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_Boolean, TLV::ContextTag(kTag_PersistTraitInstances_IsDirty));
+ SuccessOrExit(err);
+ err = reader.Get(isDirty);
+ SuccessOrExit(err);
+
+ err = reader.ExitContainer(traitInstanceContainer);
+ SuccessOrExit(err);
+
+ // allocate a new trait instance
+ WEAVE_FAULT_INJECT(FaultInjection::kFault_WDM_TraitInstanceNew, ExitNow(err = WEAVE_ERROR_NO_MEMORY));
+
+ if (SubscriptionEngine::GetInstance()->mNumTraitInfosInPool < SubscriptionEngine::kMaxNumPathGroups)
+ {
+ traitInstance =
+ SubscriptionEngine::GetInstance()->mTraitInfoPool + SubscriptionEngine::GetInstance()->mNumTraitInfosInPool;
+ ++mNumTraitInstances;
+ ++(SubscriptionEngine::GetInstance()->mNumTraitInfosInPool);
+ SYSTEM_STATS_INCREMENT(nl::Weave::System::Stats::kWDM_NumTraits);
+
+ traitInstance->Init();
+ }
+ else
+ {
+ // we run out of trait instances, abort
+ // Note it might help the client understanding what's going on with an error status like
+ // "out of memory" or "internal error", but it's pretty common that a server doesn't disclose
+ // too much internal status to clients
+ SuccessOrExit(err = WEAVE_ERROR_NO_MEMORY);
+ }
+
+
+ traitInstance->mTraitDataHandle = traitDataHandle;
+ traitInstance->mRequestedVersion = requestedVersion;
+
+ if (NULL == mTraitInstanceList)
+ {
+ // this the first trait instance for this subscription
+ // mNumTraitInstanceList has already be incremented
+ mTraitInstanceList = traitInstance;
+ }
+
+ // TODO (didis) Check if data source is persisted, set to dirty if trait data changed from persisted data
+ WeaveLogDetail(DataManagement, "Handler[%u] Syncing is requested for trait[%u].path[%u]",
+ SubscriptionEngine::GetInstance()->GetHandlerId(this), traitDataHandle, kRootPropertyPathHandle);
+ err = SubscriptionEngine::GetInstance()->mPublisherCatalog->Locate(traitDataHandle, &dataSource);
+ SuccessOrExit(err);
+ dataSource->SetRootDirty();
+ traitInstance->SetDirty();
+ }
+
+ if (err != WEAVE_END_OF_TLV)
+ {
+ goto exit;
+ }
+
+ err = reader.ExitContainer(traitInstanceArrayContainer);
+ SuccessOrExit(err);
+
+exit:
+ Platform::CriticalSectionExit();
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Load persistent trait instances error: %d", err);
+ }
+ return err;
+}
+
+WEAVE_ERROR SubscriptionHandler::SerializeSubscriptionState(TLVWriter & writer)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ TLV::TLVType container;
+
+ err = writer.StartContainer(TLV::AnonymousTag, TLV::kTLVType_Structure, container);
+ SuccessOrExit(err);
+
+
+ err = writer.Put(TLV::ContextTag(kTag_PersistSubscriptionHandler_PeerNodeId), mPeerNodeId);
+ SuccessOrExit(err);
+ err = writer.Put(TLV::ContextTag(kTag_PersistSubscriptionHandler_SubscriptionId), mSubscriptionId);
+ SuccessOrExit(err);
+ err = writer.Put(TLV::ContextTag(kTag_PersistSubscriptionHandler_LivenessTimeoutMsec), mLivenessTimeoutMsec);
+ SuccessOrExit(err);
+ err = writer.PutBoolean(TLV::ContextTag(kTag_PersistSubscriptionHandler_IsInitiator), mIsInitiator);
+ SuccessOrExit(err);
+ err = writer.PutBoolean(TLV::ContextTag(kTag_PersistSubscriptionHandler_SubscribeToAllEvents), mSubscribeToAllEvents);
+ SuccessOrExit(err);
+
+ err = SerializeTraitInstances(writer);
+ SuccessOrExit(err);
+
+ err = SerializeDeliveredEvents(writer);
+ SuccessOrExit(err);
+
+ err = writer.EndContainer(container);
+ SuccessOrExit(err);
+
+exit:
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Serialize subscription state error: %d", err);
+ }
+ return err;
+}
+
+WEAVE_ERROR SubscriptionHandler::SerializeTraitInstances(TLVWriter & writer)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ Platform::CriticalSectionEnter();
+
+ TLV::TLVType traitInstanceArrayContainer;
+ err = writer.StartContainer(TLV::ContextTag(kTag_PersistTraitInstances), TLV::kTLVType_Array, traitInstanceArrayContainer);
+ SuccessOrExit(err);
+
+ for (size_t i = 0; i < mNumTraitInstances; ++i)
+ {
+ TLV::TLVType traitInstanceContainer;
+
+ err = writer.StartContainer(TLV::AnonymousTag, TLV::kTLVType_Structure, traitInstanceContainer);
+ SuccessOrExit(err);
+ err = writer.Put(TLV::ContextTag(kTag_PersistTraitInstances_TraitDataHandle), mTraitInstanceList[i].mTraitDataHandle);
+ SuccessOrExit(err);
+ err = writer.Put(TLV::ContextTag(kTag_PersistTraitInstances_RequestedVersion), mTraitInstanceList[i].mRequestedVersion);
+ SuccessOrExit(err);
+ err = writer.PutBoolean(TLV::ContextTag(kTag_PersistTraitInstances_IsDirty), mTraitInstanceList[i].mDirty);
+ SuccessOrExit(err);
+
+ err = writer.EndContainer(traitInstanceContainer);
+ SuccessOrExit(err);
+ }
+
+ err = writer.EndContainer(traitInstanceArrayContainer);
+ SuccessOrExit(err);
+
+exit:
+ Platform::CriticalSectionExit();
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Serialize trait instances error: %d", err);
+ }
+ return err;
+}
+
+WEAVE_ERROR SubscriptionHandler::SerializeDeliveredEvents(TLVWriter & writer)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ Platform::CriticalSectionEnter();
+
+ TLV::TLVType container;
+
+ err = writer.StartContainer(TLV::ContextTag(kTag_PersistDeliveredEvent), TLV::kTLVType_Array, container);
+ SuccessOrExit(err);
+
+ for (int importance = kImportanceType_First; importance <= kImportanceType_Last; importance++)
+ {
+ TLV::TLVType deliveredEventContainer;
+ err = writer.StartContainer(TLV::AnonymousTag, TLV::kTLVType_Structure, deliveredEventContainer);
+ SuccessOrExit(err);
+
+ err = writer.Put(TLV::ContextTag(kTag_PersistDeliveredEvent_ImportanceLevel), (uint8_t)importance);
+ SuccessOrExit(err);
+
+ err = writer.Put(TLV::ContextTag(kTag_PersistDeliveredEvent_EventId), mDeliveredEvents[static_cast<uint32_t>(importance) - static_cast<uint32_t>(kImportanceType_First)]);
+ SuccessOrExit(err);
+
+ err = writer.EndContainer(deliveredEventContainer);
+ SuccessOrExit(err);
+ }
+
+ err = writer.EndContainer(container);
+ SuccessOrExit(err);
+
+exit:
+ Platform::CriticalSectionExit();
+
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Serialize delivered event id error: %d", err);
+ }
+ return err;
+}
+
+WEAVE_ERROR SubscriptionHandler::LoadDeliveredEvents(TLVReader & reader)
+{
+ WEAVE_ERROR err = WEAVE_NO_ERROR;
+
+ Platform::CriticalSectionEnter();
+
+ TLV::TLVType container;
+
+ err = reader.Next(TLV::kTLVType_Array, TLV::ContextTag(kTag_PersistDeliveredEvent));
+ SuccessOrExit(err);
+ err = reader.EnterContainer(container);
+ SuccessOrExit(err);
+
+ for (int importance = kImportanceType_First; importance <= kImportanceType_Last; importance++)
+ {
+ TLV::TLVType deliveredEventContainer;
+ uint8_t persistedImportance;
+
+ err = reader.Next(TLV::kTLVType_Structure, TLV::AnonymousTag);
+ SuccessOrExit(err);
+ err = reader.EnterContainer(deliveredEventContainer);
+ SuccessOrExit(err);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistDeliveredEvent_ImportanceLevel));
+ SuccessOrExit(err);
+ err = reader.Get(persistedImportance);
+ SuccessOrExit(err);
+ VerifyOrExit(persistedImportance == importance, err = WEAVE_ERROR_PERSISTED_STORAGE_FAIL);
+
+ err = reader.Next(TLV::kTLVType_UnsignedInteger, TLV::ContextTag(kTag_PersistDeliveredEvent_EventId));
+ SuccessOrExit(err);
+ err = reader.Get(mDeliveredEvents[static_cast<uint32_t>(importance) - static_cast<uint32_t>(kImportanceType_First)]);
+ SuccessOrExit(err);
+
+ err = reader.ExitContainer(deliveredEventContainer);
+ SuccessOrExit(err);
+ }
+
+ err = reader.ExitContainer(container);
+ SuccessOrExit(err);
+
+exit:
+ Platform::CriticalSectionExit();
+ if (err != WEAVE_NO_ERROR)
+ {
+ WeaveLogError(DataManagement, "Load persistent delivered events error: %d", err);
+ }
+ return err;
+}
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
}; // namespace WeaveMakeManagedNamespaceIdentifier(DataManagement, kWeaveManagedNamespaceDesignation_Current)
}; // namespace Profiles
}; // namespace Weave
diff --git a/src/lib/profiles/data-management/Current/SubscriptionHandler.h b/src/lib/profiles/data-management/Current/SubscriptionHandler.h
index ccd09d4..90fdd72 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionHandler.h
+++ b/src/lib/profiles/data-management/Current/SubscriptionHandler.h
@@ -217,6 +217,30 @@
kState_SubscriptionInfoValid_End = kState_Canceling,
};
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ /**
+ * @brief
+ * Tags for the persistent subscription data
+ */
+ enum
+ {
+ kTag_PersistSubscriptionHandler_PeerNodeId = 1,
+ kTag_PersistSubscriptionHandler_SubscriptionId = 2,
+ kTag_PersistSubscriptionHandler_LivenessTimeoutMsec = 3,
+ kTag_PersistSubscriptionHandler_IsInitiator = 4,
+ kTag_PersistSubscriptionHandler_SubscribeToAllEvents = 5,
+
+ kTag_PersistTraitInstances = 6,
+ kTag_PersistTraitInstances_TraitDataHandle = 7,
+ kTag_PersistTraitInstances_RequestedVersion = 8,
+ kTag_PersistTraitInstances_IsDirty = 9,
+
+ kTag_PersistDeliveredEvent = 10,
+ kTag_PersistDeliveredEvent_ImportanceLevel = 11,
+ kTag_PersistDeliveredEvent_EventId = 12
+ };
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
HandlerState mCurrentState;
bool IsNotifiable(void)
@@ -260,6 +284,9 @@
// TODO: WEAV-1426 in this incarnation, we do not account for event aggregation.
event_id_t mSelfVendedEvents[kImportanceType_Last - kImportanceType_First + 1];
event_id_t mLastScheduledEventId[kImportanceType_Last - kImportanceType_First + 1];
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ event_id_t mDeliveredEvents[kImportanceType_Last - kImportanceType_First + 1];
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
ImportanceType mCurrentImportance;
// The counter here tracks the number of event bytes offloaded to
@@ -298,6 +325,16 @@
ImportanceType FindNextImportanceForTransfer(void);
WEAVE_ERROR SetEventLogEndpoint(LoggingManagement & logger);
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ WEAVE_ERROR LoadFromPersistedState(Binding * const apBinding, void * const apAppState, EventCallback const aEventCallback, TLVReader & reader);
+ WEAVE_ERROR SerializeSubscriptionState(TLVWriter & writer);
+ WEAVE_ERROR LoadSubscriptionState(TLVReader & reader);
+ WEAVE_ERROR SerializeTraitInstances(TLVWriter & writer);
+ WEAVE_ERROR LoadTraitInstances(TLVReader & reader);
+ WEAVE_ERROR SerializeDeliveredEvents(TLVWriter & writer);
+ WEAVE_ERROR LoadDeliveredEvents(TLVReader & reader);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
#if WDM_ENABLE_SUBSCRIPTION_CANCEL
WEAVE_ERROR Cancel(void);
void CancelRequestHandler(nl::Weave::ExchangeContext * aEC, const nl::Inet::IPPacketInfo * aPktInfo,
@@ -312,6 +349,10 @@
inline WEAVE_ERROR ParseSubscriptionId(SubscribeRequest::Parser & aRequest, uint32_t & aRejectReasonProfileId,
uint16_t & aRejectReasonStatusCode, const uint64_t aRandomNumber);
+#if WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+ void UpdateDeliveredEvents(ImportanceType importance);
+#endif // WEAVE_CONFIG_PERSIST_SUBSCRIPTION_STATE
+
static void BindingEventCallback(void * const apAppState, const Binding::EventType aEvent,
const Binding::InEventParam & aInParam, Binding::OutEventParam & aOutParam);
static void OnTimerCallback(System::Layer * aSystemLayer, void * aAppState, System::Error aErrorCode);
