Merge pull request #8 from oroulet/master

high level subcsription interface, fix subscription bug

Makefile.am

@@ -209,6 +209,7 @@ libopcuaprotocol_la_SOURCES = \
   src/protocol/input_from_buffer.cpp \
   src/protocol/monitored_items.cpp \
   src/protocol/nodeid.cpp \
+  src/protocol/session.cpp \
   src/protocol/subscriptions.cpp \
   src/protocol/string_utils.cpp \
   src/protocol/types.cpp \
@@ -256,6 +257,7 @@ opcuainclude_HEADERS = \
   include/opc/ua/server.h \
   include/opc/ua/socket_channel.h \
   include/opc/ua/subscriptions.h \
+  include/opc/ua/subscription.h \
   include/opc/ua/view.h
 
 commondir = $(opcincludedir)/common
@@ -296,7 +298,8 @@ libopcuacore_la_SOURCES = \
                   src/core/common/common_errors.cpp \
                   src/core/node.cpp \
                   src/core/opcua_errors.cpp \
-                  src/core/socket_channel.cpp
+                  src/core/socket_channel.cpp \
+                  src/core/subscription.cpp
 
 libopcuacore_la_CPPFLAGS = -I$(top_srcdir)/include -I/usr/include/libxml2 $(GCOV_FLAGS)
 libopcuacore_la_LIBADD = libopcuaprotocol.la
@@ -386,15 +389,15 @@ example_client_SOURCES = \
 
 example_client_CPPFLAGS = -I$(top_srcdir)/include -I/usr/include/libxml2
 example_client_LDADD = libopcuaclient.la libopcuaprotocol.la libopcuacore.la
-example_client_LDFLAGS = -ldl -lpthread -lxml2
+example_client_LDFLAGS = -ldl -lpthread -lxml2 -lboost_system 
 
 noinst_PROGRAMS += example_server
 example_server_SOURCES = \
   src/examples/example_server.cpp
 
 example_server_CPPFLAGS = -I$(top_srcdir)/include -I/usr/include/libxml2
 example_server_LDADD = libopcuaserver.la libopcuaprotocol.la libopcuacore.la
-example_server_LDFLAGS = -ldl -lpthread -lxml2
+example_server_LDFLAGS = -ldl -lpthread -lxml2 -lboost_system 
 
 
 

include/opc/ua/client/client.h

@@ -22,6 +22,7 @@
 
 #include <opc/ua/node.h>
 #include <opc/ua/server.h>
+#include <opc/ua/subscription.h>
 #include <opc/ua/client/remote_server.h>
 
 
@@ -65,6 +66,8 @@ namespace OpcUa
     void Connect();
     void Disconnect();
 
+    Subscription CreateSubscription(uint period, SubscriptionClient& client);
+
   private:
     std::string Endpoint = "opc.tcp:://localhost:4841";
     std::string SessionName = "Open source OPC-UA Client Session";

include/opc/ua/protocol/expanded_object_ids.h

@@ -27,6 +27,9 @@ namespace OpcUa
     DataTypeAttribute = 372,
     ViewAttribute = 375,
 
+    DataChangeNotification = 811,
+    EventNotificationList = 812,
+    StatusChangeNotification = 813,
   };
 }
 

include/opc/ua/protocol/object_ids.h

@@ -122,9 +122,6 @@ namespace OpcUa
     StructureDeleteNodesItem = 382,
     StructureDeleteReferencesItem = 385,
 
-    DataChangeNotification = 811,
-    EventNotificationList = 812,
-    StatusChangeNotification = 813,
 
     RedundancySupport = 851,
     ServerState    = 852,

include/opc/ua/protocol/session.h

@@ -139,6 +139,9 @@ namespace OpcUa
     CloseSessionResponse();
   };
 
+  NodeID GenerateSessionId();
+
+
 } // namespace OpcUa
 
 #endif // __OPC_UA_BINARY_SESSIONS_H__

include/opc/ua/protocol/subscriptions.h

@@ -41,6 +41,8 @@ namespace OpcUa
     NodeID TypeID;
     RequestHeader Header;
     std::vector<IntegerID> SubscriptionsIds;
+
+    DeleteSubscriptionRequest();
   };
 
   struct DeleteSubscriptionResponse

include/opc/ua/protocol/types.h

@@ -61,7 +61,7 @@ namespace OpcUa
   {
   public:
     IntegerID();
-    explicit IntegerID(const IntegerID& id);
+    IntegerID(const IntegerID& id);
     explicit IntegerID(uint32_t num);
     IntegerID& operator= (const IntegerID& id);
     IntegerID& operator= (uint32_t value);

include/opc/ua/server/opcuaserver.h

@@ -8,6 +8,7 @@
 #include <opc/common/addons_core/dynamic_addon_factory.h>
 #include <opc/ua/node.h>
 #include <opc/ua/server.h>
+#include <opc/ua/subscription.h>
 #include <opc/ua/server/tcp_server.h>
 #include <opc/ua/server/address_space.h>
 #include <opc/ua/server/endpoints_services.h>
@@ -37,6 +38,8 @@ namespace OpcUa
       Node GetNodeFromPath(const std::vector<QualifiedName>& path) {return GetRootNode().GetChild(path);}
       Node GetNodeFromPath(const std::vector<std::string>& path) {return GetRootNode().GetChild(path);}
 
+      Subscription CreateSubscription(uint period, SubscriptionClient& callback);
+
     protected:
       std::vector<std::string> xml_address_spaces;
       std::string config_path = "";

include/opc/ua/subscription.h

@@ -0,0 +1,78 @@
+/******************************************************************************
+ *   Copyright (C) 2014-2014 Olivier Roulet-Dubonnet          *
+ *   olivier.roulet@gmail.com          *
+ *                      *
+ *   This library is free software; you can redistribute it and/or modify   *
+ *   it under the terms of the GNU Lesser General Public License as      *
+ *   published by the Free Software Foundation; version 3 of the License.   *
+ *                      *
+ *   This library is distributed in the hope that it will be useful,      *
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of      *
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the      *
+ *   GNU Lesser General Public License for more details.        *
+ *                      *
+ *   You should have received a copy of the GNU Lesser General Public License *
+ *   along with this library; if not, write to the          *
+ *   Free Software Foundation, Inc.,              *
+ *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.        *
+ ******************************************************************************/
+
+#pragma once
+
+#include <opc/ua/subscriptions.h>
+#include <opc/ua/protocol/subscriptions.h>
+#include <opc/ua/node.h>
+
+#include <boost/asio.hpp>
+#include <sstream>
+#include <map>
+
+
+namespace OpcUa
+{
+  typedef std::map<IntegerID, AttributeValueID> AttValMap;
+
+  class SubscriptionClient
+  {
+    public:
+      virtual void DataChangeEvent(const Node& node, const Variant& val, AttributeID attribute){};
+      //virtual void Event(Event result); FIXME: Not implemented
+      //virtual void StatusChange(Event result); FIXME: Not implemented
+  };
+
+
+  class Subscription
+  {
+    public:
+      //Create a new subscription on server
+      //callback will be called everytime an event is received from the server
+      //FIXME: should we use interface or std::function for callback???? std::function syntax is ugly but is more flexible
+      Subscription(Remote::Server::SharedPtr server, const SubscriptionParameters& params, SubscriptionClient& callback); 
+      IntegerID GetId() const { return Data.ID; } 
+      Duration GetPeriode() const { return Data.RevisedPublishingInterval; } 
+      //Delete the subscription from server
+      void Delete();
+
+      //Subscribe to a Node attribute for its value to change
+      uint32_t Subscribe(const Node& node, AttributeID attr=AttributeID::VALUE);
+      // Subscribe to nodes for specified attribute change
+      std::vector<CreateMonitoredItemsResult> Subscribe(const std::vector<AttributeValueID>& attributes);
+      void UnSubscribe(std::vector<uint32_t> handles){}; //Not implemented in interface and server
+      //Monitor for events FIXME: Event support not implemented
+      //void SubscribeEvents(Node node); //As far as I remember the only allowed node is Server in most SDKs
+      //void Unsubscribe(Node node);
+      void PublishCallback(PublishResult); //Not sure it needs to be public
+
+    private:
+      void Publish();
+
+      Remote::Server::SharedPtr Server;
+      SubscriptionData Data;
+      std::vector<uint32_t> Acknowledgments;
+      SubscriptionClient& Client;
+      uint32_t LastMonitoredItemHandle = 1;
+      AttValMap Map; //I do not understand why I need this but event only send handles..
+  };
+}
+
+

include/opc/ua/subscriptions.h

@@ -20,12 +20,6 @@ namespace OpcUa
 {
   namespace Remote
   {
-    /*
-    class SubscriptionClient
-    {
-      virtual void newPublishResponse(PublishResult result);
-    };
-    */
 
     class SubscriptionServices : private Common::Interface
     {

src/client/client.cpp

@@ -71,5 +71,11 @@ namespace OpcUa
     return Node(Server, OpcUa::ObjectID::ObjectsFolder);
   }
 
+  Subscription RemoteClient::CreateSubscription(uint period, SubscriptionClient& callback)
+  {
+    SubscriptionParameters params;
+    params.RequestedPublishingInterval = period;
+    return Subscription(Server, params, callback);
+  }
 } // namespace OpcUa
 

src/client/stream_subscription.h

@@ -16,6 +16,8 @@
 #include <opc/ua/protocol/monitored_items.h>
 #include <opc/ua/protocol/binary/stream.h>
 
+#include <opc/ua/protocol/string_utils.h>
+
 namespace OpcUa
 {
   namespace Internal
@@ -65,9 +67,11 @@ namespace OpcUa
 
         Stream << request << OpcUa::Binary::flush;
 
-        CreateMonitoredItemsResponse response;
-        Stream >> response;
-        return response.Data;
+        ProcessPublishResults();
+
+        MonitoredItemsData data;
+        Stream >> data;
+        return data;
       }
 
       virtual std::vector<PublishResult> PopPublishResults(const std::vector<IntegerID>& subscriptionsIds)
@@ -83,6 +87,27 @@ namespace OpcUa
       }
 
     private:
+      void ProcessPublishResults()
+      {
+        NodeID typeId;
+        ResponseHeader header;
+        for(;;)
+        {
+          Stream >> typeId;
+          Stream >> header;
+          std::cout << " got header with type: " << typeId << std::endl;
+          if (typeId == NodeID(829, 0) )
+          {
+            PublishResult result;
+            Stream >> result;
+            std::cout << " got one publish result " << typeId << std::endl;
+          }
+          else
+          {
+            break;
+          }
+        }
+      }
       mutable StreamType Stream;
       NodeID AuthenticationToken;
     };