Merge pull request #649 from openweave/bug/improve_ble_endpoint_close
Improve woble connection close
diff --git a/Makefile-Android b/Makefile-Android
index f152356..e4f3c23 100644
--- a/Makefile-Android
+++ b/Makefile-Android
@@ -271,7 +271,7 @@
LDFLAGS="$(LDFLAGS)" \
JAVA_HOME="$(JAVA_HOME)" \
JNI_INCLUDE_DIRS="$(JNI_INCLUDE_DIRS)" \
---build=$(shell $(AbsTopSourceDir)/third_party/nlbuild-autotools/repo/autoconf/config.guess) \
+--build=$(shell $(AbsTopSourceDir)/third_party/nlbuild-autotools/repo/third_party/autoconf/config.guess) \
--host=$(ABI_CONFIG_TUPLE_$(1)) \
--with-sysroot=$(ABI_SYSROOT) \
--with-libtool-sysroot=$(ABI_SYSROOT) \
diff --git a/Makefile-iOS b/Makefile-iOS
index 2aebc79..4d8c33f 100644
--- a/Makefile-iOS
+++ b/Makefile-iOS
@@ -294,7 +294,7 @@
lt_cv_ld_exported_symbols_list=yes \
ac_cv_func_clock_gettime=no \
ac_cv_have_decl_clock_gettime=no \
---build=$(shell $(AbsTopSourceDir)/third_party/nlbuild-autotools/repo/autoconf/config.guess) \
+--build=$(shell $(AbsTopSourceDir)/third_party/nlbuild-autotools/repo/third_party/autoconf/config.guess) \
--host=$($(1)_arch_AUTOTOOLS)-$(TargetTupleStem) \
--target=$($(1)_arch_AUTOTOOLS)-$(TargetTupleStem) \
--prefix=/ \
diff --git a/src/device-manager/cocoa/NLWdmClient.h b/src/device-manager/cocoa/NLWdmClient.h
index 8d59868..4a690fe 100644
--- a/src/device-manager/cocoa/NLWdmClient.h
+++ b/src/device-manager/cocoa/NLWdmClient.h
@@ -121,7 +121,7 @@
* DataSchemaVersionRange | Object{MinVersion: uint64, MaxVersion: uint64}
* Data | Object | Event Trait Data
*/
-- (void)getEvents:(NSString **)events;
+- (WEAVE_ERROR)getEvents:(NSString **)events;
@end
diff --git a/src/lib/core/WeaveConfig.h b/src/lib/core/WeaveConfig.h
index bb510b6..b3990a2 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 edfdcdc..f26df90 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 3603532..a4319f1 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionClient.cpp
+++ b/src/lib/profiles/data-management/Current/SubscriptionClient.cpp
@@ -165,6 +165,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 5e3fb05..0ab3825 100644
--- a/src/lib/profiles/data-management/Current/SubscriptionClient.h
+++ b/src/lib/profiles/data-management/Current/SubscriptionClient.h
@@ -450,6 +450,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; }
@@ -495,6 +507,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 b37a1c5..8070e27 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 7ae183d..1dea155 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);
@@ -537,6 +563,11 @@
#endif // WDM_PUBLISHER_ENABLE_CUSTOM_COMMAND_HANDLER
#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);
diff --git a/src/lib/profiles/device-control/DeviceControl.cpp b/src/lib/profiles/device-control/DeviceControl.cpp
index 6c41f7d..3fca886 100644
--- a/src/lib/profiles/device-control/DeviceControl.cpp
+++ b/src/lib/profiles/device-control/DeviceControl.cpp
@@ -1193,6 +1193,13 @@
#if WEAVE_CONFIG_REQUIRE_AUTH_DEVICE_CONTROL
case kMsgType_ResetConfig:
+ if (authMode == kWeaveAuthMode_CASE_AccessToken || authMode == kWeaveAuthMode_CASE_ServiceEndPoint ||
+ (authMode == kWeaveAuthMode_PASE_PairingCode && !IsPairedToAccount()))
+ {
+ result = kAccessControlResult_Accepted;
+ }
+ break;
+
case kMsgType_ArmFailSafe:
if (authMode == kWeaveAuthMode_CASE_AccessToken ||
(authMode == kWeaveAuthMode_PASE_PairingCode && !IsPairedToAccount()))
