Merge pull request #620 from erjiaqing/wdm_events

[WDM] Add GetEvents() in WeaveDataManagementClient

Author: robszewczyk

src/device-manager/WeaveDataManagementClient.cpp

@@ -36,10 +36,14 @@
 #include <SystemLayer/SystemPacketBuffer.h>
 #include <Weave/Profiles/data-management/SubscriptionClient.h>
 #include <Weave/Profiles/data-management/Current/GenericTraitCatalogImpl.h>
+#include <Weave/Profiles/data-management/Current/EventProcessor.h>
 #include <map>
+#include <memory>
 #include <vector>
 #include "WeaveDeviceManager.h"
 
+#include <chrono>
+
 namespace nl {
 namespace Weave {
 namespace DeviceManager {
@@ -70,6 +74,7 @@ class NL_DLL_EXPORT BytesData
 
 class GenericTraitUpdatableDataSink;
 class WdmClient;
+class WdmEventProcessor;
 
 class WdmClientFlushUpdateStatus
 {
@@ -167,6 +172,7 @@ class NL_DLL_EXPORT GenericTraitUpdatableDataSink : public nl::Weave::Profiles::
 class NL_DLL_EXPORT WdmClient
 {
     friend class GenericTraitUpdatableDataSink;
+    friend class WdmEventProcessor;
 
 public:
     enum
@@ -189,7 +195,10 @@ class NL_DLL_EXPORT WdmClient
     WEAVE_ERROR FlushUpdate(void * apAppReqState, DMFlushUpdateCompleteFunct onComplete, DMErrorFunct onError);
 
     WEAVE_ERROR RefreshData(void * apAppReqState, DMCompleteFunct onComplete, DMErrorFunct onError,
-                            GetDataHandleFunct getDataHandleCb);
+                            GetDataHandleFunct getDataHandleCb, bool aFetchEvents = false);
+
+    WEAVE_ERROR GetEvents(BytesData * aBytes);
+    void SetEventFetchingTimeout(uint32_t aTimeoutSec);
 
     void * mpAppState;
 
@@ -216,9 +225,8 @@ class NL_DLL_EXPORT WdmClient
     static void ClientEventCallback(void * const aAppState, SubscriptionClient::EventID aEvent,
                                     const SubscriptionClient::InEventParam & aInParam,
                                     SubscriptionClient::OutEventParam & aOutParam);
-
     WEAVE_ERROR RefreshData(void * apAppReqState, void * apContext, DMCompleteFunct onComplete, DMErrorFunct onError,
-                            GetDataHandleFunct getDataHandleCb);
+                            GetDataHandleFunct getDataHandleCb, bool aWithEvents);
     WEAVE_ERROR GetDataSink(const ResourceIdentifier & aResourceId, uint32_t aProfileId, uint64_t aInstanceId,
                             GenericTraitUpdatableDataSink *& apGenericTraitUpdatableDataSink);
     WEAVE_ERROR SubscribePublisherTrait(const ResourceIdentifier & aResourceId, const uint64_t & aInstanceId,
@@ -230,6 +238,10 @@ class NL_DLL_EXPORT WdmClient
                                         PropertyPathHandle mPropertyPathHandle, uint32_t aReason, uint32_t aStatusProfileId,
                                         uint16_t aStatusCode);
 
+    WEAVE_ERROR ProcessEvent(nl::Weave::TLV::TLVReader inReader, const EventProcessor::EventHeader & inEventHeader);
+
+    WEAVE_ERROR PrepareLastObservedEventList(uint32_t & aEventListLen);
+
     GenericTraitSinkCatalog mSinkCatalog;
     TraitPath * mpPublisherPathList;
 
@@ -240,7 +252,39 @@ class NL_DLL_EXPORT WdmClient
     OpState mOpState;
     std::vector<std::string> mFailedPaths;
     std::vector<WdmClientFlushUpdateStatus> mFailedFlushPathStatus;
+
+    std::unique_ptr<WdmEventProcessor> mpWdmEventProcessor;
+    std::string mEventStrBuffer;
+
+    // Three flags for event fetching
+    // If EventFetching is enabled
+    // If we limit the event fetching time
+    // If event fetching time limit exceeded (TLE)
+    bool mEnableEventFetching, mLimitEventFetchTimeout, mEventFetchingTLE;
+
+    std::chrono::time_point<std::chrono::system_clock> mEventFetchTimeout;
+    std::chrono::seconds mEventFetchTimeLimit;
+    SubscriptionClient::LastObservedEvent mLastObservedEventByImportance[kImportanceType_Last - kImportanceType_First + 1];
+    SubscriptionClient::LastObservedEvent
+        mLastObservedEventByImportanceForSending[kImportanceType_Last - kImportanceType_First + 1];
 };
+
+class WdmEventProcessor : public EventProcessor
+{
+public:
+    WdmEventProcessor(uint64_t aNodeId, WdmClient * aWdmClient);
+    virtual ~WdmEventProcessor() = default;
+
+protected:
+    WEAVE_ERROR ProcessEvent(nl::Weave::TLV::TLVReader inReader, nl::Weave::Profiles::DataManagement::SubscriptionClient & inClient,
+                             const EventHeader & inEventHeader) override;
+
+    WEAVE_ERROR GapDetected(const EventHeader & inEventHeader) override;
+
+private:
+    WdmClient * mWdmClient;
+};
+
 } // namespace DeviceManager
 } // namespace Weave
 } // namespace nl

src/device-manager/WeaveDataManagementClient.h

@@ -59,6 +59,7 @@ typedef void (^WdmClientFailureBlock)(id owner, NSError * error);
  */
 - (void)setNodeId:(uint64_t)nodeId;
 
+
 /**
  * Create the new data newDataSink
  *
@@ -89,4 +90,38 @@ typedef void (^WdmClientFailureBlock)(id owner, NSError * error);
  */
 - (void)refreshData:(WdmClientCompletionBlock)completionHandler failure:(WdmClientFailureBlock)failureHandler;
 
+/**
+ * Begins a sync of all events. The result of this operation can be observed through the CompletionHandler and
+ * failureHandler, the procedure will be terminated after reaching timeout.
+ */
+- (void)fetchEvents:(WdmClientCompletionBlock)completionHandler failure:(WdmClientFailureBlock)failureHandler
+            timeoutSec:(uint32_t)timeoutSec;
+
+/**
+ * getEvents will return a list of event data in json array representation.
+ * If no events have been fetched, an empty array ("[]") will be returned.
+ * The internal buffer will be cleared when beginFetchEvents() is called.
+ * Each element in this array should be an object
+ *
+ * Field                  | Type        | Description
+ * -----------------------+-------------+--------------
+ * Source                 | uint64      | Event Header
+ * Importance             | int (enum)
+ * Id                     | uint64
+ * RelatedImportance      | uint64
+ * RelatedId              | uint64
+ * UTCTimestamp           | uint64
+ * ResourceId             | uint64
+ * TraitProfileId         | uint64
+ * TraitInstanceId        | uint64
+ * Type                   | uint64
+ * DeltaUTCTime           | int
+ * DeltaSystemTime        | int
+ * PresenceMask           | uint64
+ * DataSchemaVersionRange | Object{MinVersion: uint64, MaxVersion: uint64}
+ * Data                   | Object      | Event Trait Data
+ */
+- (void)getEvents:(NSString **)events;
+
+
 @end

src/device-manager/cocoa/NLWdmClient.h

@@ -547,5 +547,61 @@ - (void)refreshData:(WdmClientCompletionBlock)completionHandler failure:(WdmClie
     });
 }
 
+- (void)fetchEvents:(WdmClientCompletionBlock)completionHandler failure:(WdmClientFailureBlock)failureHandler timeoutSec:(uint32_t)timeoutSec
+{
+    WDM_LOG_METHOD_SIG();
+
+    NSString * taskName = @"FetchEvents";
+
+    dispatch_sync(_mWeaveWorkQueue, ^() {
+        _mWeaveCppWdmClient->SetEventFetchingTimeout(timeoutSec);
+    });
+
+    // we use async for the results are sent back to caller via async means also
+    dispatch_async(_mWeaveWorkQueue, ^() {
+        if (nil == _mRequestName) {
+            _mRequestName = taskName;
+            _mCompletionHandler = [completionHandler copy];
+            _mFailureHandler = [failureHandler copy];
+
+            WEAVE_ERROR err = _mWeaveCppWdmClient->RefreshData((__bridge void *) self, onWdmClientComplete, onWdmClientError, NULL, true);
+
+            if (WEAVE_NO_ERROR != err) {
+                [self dispatchAsyncDefaultFailureBlockWithCode:err];
+            }
+        } else {
+            WDM_LOG_ERROR(@"%@: Attemp to %@ while we're still executing %@, ignore", _name, taskName, _mRequestName);
+
+            // do not change _mRequestName, as we're rejecting this request
+            [self dispatchAsyncFailureBlock:WEAVE_ERROR_INCORRECT_STATE taskName:taskName handler:failureHandler];
+        }
+    });
+}
+
+- (WEAVE_ERROR)getEvents:(NSString **)events
+{
+    __block WEAVE_ERROR err = WEAVE_NO_ERROR;
+    __block nl::Weave::DeviceManager::BytesData bytesData;
+
+    WDM_LOG_METHOD_SIG();
+
+    VerifyOrExit(NULL != _mWeaveCppWdmClient, err = WEAVE_ERROR_INCORRECT_STATE);
+
+    // need this bracket to use Verify macros
+    {
+        // we use sync so the bytesData is immediately available to the caller upon return
+        dispatch_sync(_mWeaveWorkQueue, ^() {
+            err = _mWeaveCppWdmClient->GetEvents(&bytesData);
+        });
+    }
+
+exit:
+    if (WEAVE_NO_ERROR == err)
+    {
+        *events = [[NSString alloc] initWithBytes:bytesData.mpDataBuf length:bytesData.mDataLen encoding:NSUTF8StringEncoding];
+    }
+    return err;
+}
+
 @end
 #endif // WEAVE_CONFIG_DATA_MANAGEMENT_CLIENT_EXPERIMENTAL

src/device-manager/cocoa/NLWdmClient.mm

@@ -184,6 +184,8 @@ extern "C" {
     NL_DLL_EXPORT jlong Java_nl_Weave_DataManagement_WdmClientImpl_newDataSink(JNIEnv *env, jobject self, jlong wdmClientPtr, jobject resourceIdentifierObj, jlong profileId, jlong instanceId, jstring path);
     NL_DLL_EXPORT void Java_nl_Weave_DataManagement_WdmClientImpl_beginFlushUpdate(JNIEnv *env, jobject self, jlong wdmClientPtr);
     NL_DLL_EXPORT void Java_nl_Weave_DataManagement_WdmClientImpl_beginRefreshData(JNIEnv *env, jobject self, jlong wdmClientPtr);
+    NL_DLL_EXPORT void Java_nl_Weave_DataManagement_WdmClientImpl_beginFetchEvents(JNIEnv *env, jobject self, jlong wdmClientPtr, jint timeoutSec);
+    NL_DLL_EXPORT jstring Java_nl_Weave_DataManagement_WdmClientImpl_getEvents(JNIEnv *env, jobject self, jlong wdmClientPtr);
     NL_DLL_EXPORT void Java_nl_Weave_DataManagement_GenericTraitUpdatableDataSinkImpl_init(JNIEnv *env, jobject self, jlong genericTraitUpdatableDataSinkPtr);
     NL_DLL_EXPORT void Java_nl_Weave_DataManagement_GenericTraitUpdatableDataSinkImpl_shutdown(JNIEnv *env, jobject self, jlong genericTraitUpdatableDataSinkPtr);
     NL_DLL_EXPORT void Java_nl_Weave_DataManagement_GenericTraitUpdatableDataSinkImpl_clear(JNIEnv *env, jobject self, jlong genericTraitUpdatableDataSinkPtr);
@@ -3995,6 +3997,47 @@ void Java_nl_Weave_DataManagement_WdmClientImpl_beginRefreshData(JNIEnv *env, jo
     }
 }
 
+void Java_nl_Weave_DataManagement_WdmClientImpl_beginFetchEvents(JNIEnv *env, jobject self, jlong wdmClientPtr, jint timeoutSec)
+{
+    WEAVE_ERROR err = WEAVE_NO_ERROR;
+    WdmClient *wdmClient = (WdmClient *)wdmClientPtr;
+
+    WeaveLogProgress(DeviceManager, "beginFetchEvents() called");
+
+    pthread_mutex_lock(&sStackLock);
+    wdmClient->SetEventFetchingTimeout((uint32_t)timeoutSec);
+    err = wdmClient->RefreshData((void *)"FetchEvents", HandleWdmClientComplete, HandleWdmClientError, NULL, true);
+    pthread_mutex_unlock(&sStackLock);
+
+    if (err != WEAVE_NO_ERROR && err != WDM_JNI_ERROR_EXCEPTION_THROWN)
+    {
+        ThrowError(env, err);
+    }
+}
+
+
+jstring Java_nl_Weave_DataManagement_WdmClientImpl_getEvents(JNIEnv *env, jobject self, jlong wdmClientPtr)
+{
+    WEAVE_ERROR err = WEAVE_NO_ERROR;
+    WdmClient *wdmClient = (WdmClient *)wdmClientPtr;
+    BytesData bytesData;
+
+    WeaveLogProgress(DeviceManager, "getEvents() called");
+
+    err = wdmClient->GetEvents(&bytesData);
+    SuccessOrExit(err);
+
+exit:
+
+    if (err != WEAVE_NO_ERROR && err != WDM_JNI_ERROR_EXCEPTION_THROWN)
+    {
+        ThrowError(env, err);
+    }
+
+    // bytesData.mpDataBuf should be a pointer get by mEventBuffer.c_str()
+    return env->NewStringUTF(reinterpret_cast<const char *>(bytesData.mpDataBuf));
+}
+
 void Java_nl_Weave_DataManagement_GenericTraitUpdatableDataSinkImpl_init(JNIEnv *env, jobject self, jlong genericTraitUpdatableDataSinkPtr)
 {
     GenericTraitUpdatableDataSink *pDataSink = NULL;

src/device-manager/java/WeaveDeviceManager-JNI.cpp

@@ -67,15 +67,51 @@ public interface WdmClient
      */
     public void beginRefreshData();
 
+    /**
+     * The result of this operation can be observed through the {@link CompletionHandler}
+     * that has been assigned via {@link #setCompletionHandler}.
+     * The fetched events can be accessed via getEvents()
+     *
+     * @param timeoutSec operation timeoutSec, this operation might take a long time if the number of events is too large.
+     */
+    public void beginFetchEvents(int timeoutSec);
+
     public CompletionHandler getCompletionHandler();
     public void setCompletionHandler(CompletionHandler compHandler);
 
     public GenericTraitUpdatableDataSink getDataSink(long traitInstancePtr);
 
+    /**
+     * getEvents will return a list of event data in json array representation.
+     * If no events have been fetched, an empty array ("[]") will be returned.
+     * The internal buffer will be cleared when beginFetchEvents() is called.
+     * Each element in this array should be an object
+     *
+     * Field                  | Type        | Description
+     * -----------------------+-------------+--------------
+     * Source                 | uint64      | Event Header
+     * Importance             | int (enum)
+     * Id                     | uint64
+     * RelatedImportance      | uint64
+     * RelatedId              | uint64
+     * UTCTimestamp           | uint64
+     * ResourceId             | uint64
+     * TraitProfileId         | uint64
+     * TraitInstanceId        | uint64
+     * Type                   | uint64
+     * DeltaUTCTime           | int
+     * DeltaSystemTime        | int
+     * PresenceMask           | uint64
+     * DataSchemaVersionRange | Object{MinVersion: uint64, MaxVersion: uint64}
+     * Data                   | Object      | Event Trait Data
+     */
+    public String getEvents();
+
     public interface CompletionHandler
     {
         void onFlushUpdateComplete(Throwable[] exceptions, WdmClient wdmClient);
         void onRefreshDataComplete();
+        void onFetchEventsComplete();
         void onError(Throwable err);
     }
 };

src/device-manager/java/src/nl/Weave/DataManagement/WdmClient.java

@@ -123,6 +123,13 @@ public void beginRefreshData()
         beginRefreshData(mWdmClientPtr);
     }
 
+    @Override
+    public void beginFetchEvents(int timeoutSec)
+    {
+        ensureNotClosed();
+        beginFetchEvents(mWdmClientPtr, timeoutSec);
+    }
+
     @Override
     public CompletionHandler getCompletionHandler()
     {
@@ -135,6 +142,13 @@ public void setCompletionHandler(CompletionHandler compHandler)
         mCompHandler = compHandler;
     }
 
+    @Override
+    public String getEvents()
+    {
+        ensureNotClosed();
+        return getEvents(mWdmClientPtr);
+    }
+
     public GenericTraitUpdatableDataSink getDataSink(long traitInstancePtr)
     {
         GenericTraitUpdatableDataSink trait = null;
@@ -179,6 +193,11 @@ private void onRefreshDataComplete()
         requireCompletionHandler().onRefreshDataComplete();
     }
 
+    private void onFetchEventsComplete()
+    {
+        requireCompletionHandler().onFetchEventsComplete();
+    }
+
     private void ensureNotClosed() {
       if (mWdmClientPtr == 0) {
         throw new IllegalStateException("This WdmClient has already been closed.");
@@ -205,4 +224,6 @@ private CompletionHandler requireCompletionHandler() {
     private native long newDataSink(long wdmClientPtr, ResourceIdentifier resourceIdentifier, long profileId, long instanceId, String path);
     private native void beginFlushUpdate(long wdmClientPtr);
     private native void beginRefreshData(long wdmClientPtr);
+    private native void beginFetchEvents(long wdmClientPtr, int timeoutSec);
+    private native String getEvents(long wdmClientPtr);
 };