Synthetic notifications

AdsLib/AdsLib.h

@@ -128,6 +128,35 @@ long AdsSyncAddDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
 long AdsSyncDelDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
 				       uint32_t hNotification);
 
+/**
+ * A specific type of notification about the internal state of the library.
+ * When a state change as defined by nType parameter occurs the callback function is invoked
+ * @param[in] port port number of an Ads port that had previously been opened with AdsPortOpenEx().
+ * @param[in] pAddr Structure with NetId and port number of the ADS server.
+ * @param[in] nType The notification type.
+ * @param[in] pFunc Pointer to the structure describing the callback function.
+ * @param[in] hUser 32-bit value that is passed to the callback function.
+ * @param[out] pNotification Address of the variable that will receive the handle of the notification.
+ * @return [ADS Return
+ * Code](https://infosys.beckhoff.com/content/1031/tcadscommon/html/ads_returncodes.htm?id=1666172286265530469)
+ */
+long AdsAddSyntheticDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
+				       uint32_t nType,
+				       PAdsSyntheticNotificationFuncEx pFunc,
+				       uint32_t hUser,
+				       uint32_t *pNotification);
+
+/**
+ * A notification defined previously is deleted from an ADS server.
+ * @param[in] port port number of an Ads port that had previously been opened with AdsPortOpenEx().
+ * @param[in] pAddr Structure with NetId and port number of the ADS server.
+ * @param[in] hNotification Address of the variable that contains the handle of the notification.
+ * @return [ADS Return
+ * Code](https://infosys.beckhoff.com/content/1031/tcadscommon/html/ads_returncodes.htm?id=1666172286265530469)
+ */
+long AdsDelSyntheticDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
+				       uint32_t hNotification);
+
 /**
  * Read the configured timeout for the ADS functions. The standard value is 5000 ms.
  * @param[in] port port number of an Ads port that had previously been opened with AdsPortOpenEx().

AdsLib/AdsSyntheticNotification.h

@@ -0,0 +1,38 @@
+// SPDX-License-Identifier: MIT
+
+#pragma once
+
+#include "AdsDef.h"
+
+using VirtualConnection = std::pair<uint16_t, AmsAddr>;
+
+struct SyntheticNotification {
+	const VirtualConnection connection;
+	const uint32_t type;
+
+	SyntheticNotification(PAdsSyntheticNotificationFuncEx __func,
+			      uint32_t __hUser, AmsAddr __amsAddr,
+			      uint16_t __port, uint32_t __type)
+		: connection({ __port, __amsAddr })
+		, type(__type)
+		, callback(__func)
+		, hNotification(0)
+		, hUser(__hUser)
+	{
+	}
+
+	void Notify()
+	{
+		callback(&connection.second, hNotification, hUser);
+	}
+
+	void hNotify(uint32_t value)
+	{
+		hNotification = value;
+	}
+
+    private:
+	const PAdsSyntheticNotificationFuncEx callback;
+	uint32_t hNotification;
+	const uint32_t hUser;
+};

AdsLib/AmsConnection.h

@@ -74,6 +74,9 @@ struct AmsConnection {
 	SharedDispatcher
 	CreateNotifyMapping(uint32_t hNotify,
 			    std::shared_ptr<Notification> notification);
+	SharedDispatcher
+	CreateSyntheticNotifyMapping(uint32_t hNotify,
+			    std::shared_ptr<SyntheticNotification> notification);
 	long DeleteNotification(const AmsAddr &amsAddr, uint32_t hNotify,
 				uint32_t tmms, uint16_t port);
 	long AdsRequest(AmsRequest &request, uint32_t timeout);

AdsLib/AmsPort.h

@@ -19,9 +19,14 @@ struct AmsPort {
 			     SharedDispatcher dispatcher);
 	long DelNotification(AmsAddr ams, uint32_t hNotify);
 
+	void AddSyntheticNotification(AmsAddr ams, uint32_t hNotify,
+			     SharedDispatcher dispatcher);
+	long DelSyntheticNotification(AmsAddr ams, uint32_t hNotify);
+
     private:
 	using NotifyUUID = std::pair<const AmsAddr, const uint32_t>;
 	static const uint32_t DEFAULT_TIMEOUT = 5000;
 	std::map<NotifyUUID, SharedDispatcher> dispatcherList;
+	std::map<NotifyUUID, SharedDispatcher> syntheticDispatcherList;
 	std::mutex mutex;
 };

AdsLib/AmsRouter.h

@@ -6,6 +6,7 @@
 #pragma once
 
 #include "AmsConnection.h"
+#include <atomic>
 #include <unordered_set>
 
 struct AmsRouter : Router {
@@ -21,6 +22,11 @@ struct AmsRouter : Router {
 			     std::shared_ptr<Notification> notify);
 	long DelNotification(uint16_t port, const AmsAddr *pAddr,
 			     uint32_t hNotification);
+	long AddSyntheticNotification(uint16_t port, const AmsAddr &addr, uint32_t *pNotification,
+			     std::shared_ptr<SyntheticNotification> notify);
+	long DelSyntheticNotification(uint16_t port, const AmsAddr *pAddr,
+			     uint32_t hNotification);
+	uint32_t AllocateNotifyId();
 
 	[[deprecated]]
 	long AddRoute(AmsNetId ams, const IpV4 &ip);
@@ -43,4 +49,6 @@ struct AmsRouter : Router {
 	void DeleteIfLastConnection(const AmsConnection *conn);
 
 	std::array<AmsPort, NUM_PORTS_MAX> ports;
+
+	std::atomic<uint32_t> nextNotifyId;
 };

AdsLib/NotificationDispatcher.h

@@ -6,36 +6,46 @@
 #pragma once
 
 #include "AdsNotification.h"
+#include "AdsSyntheticNotification.h"
 #include "AmsHeader.h"
 #include "Semaphore.h"
 
 #include <atomic>
 #include <functional>
 #include <map>
 #include <thread>
+#include <vector>
 
 using DeleteNotificationCallback =
 	std::function<long(uint32_t hNotify, uint32_t tmms)>;
 
 struct NotificationDispatcher {
-	NotificationDispatcher(DeleteNotificationCallback callback);
+	NotificationDispatcher(VirtualConnection connection,
+		     DeleteNotificationCallback callback);
 	~NotificationDispatcher();
 	void Emplace(uint32_t hNotify,
 		     std::shared_ptr<Notification> notification);
 	long Erase(uint32_t hNotify, uint32_t tmms);
+	void EmplaceSynthetic(uint32_t hNotify,
+		     std::shared_ptr<SyntheticNotification> notification);
+	long EraseSynthetic(uint32_t hNotify);
 	void Notify();
 	void Run();
 
 	const DeleteNotificationCallback deleteNotification;
 	RingBuffer ring;
 
-    private:
+	private:
+	const VirtualConnection connection;
 	std::map<uint32_t, std::shared_ptr<Notification> > notifications;
+	std::map<uint32_t, std::shared_ptr<SyntheticNotification> >	syntheticNotifications;
 	std::recursive_mutex mutex;
 	Semaphore sem;
 	std::atomic<bool> stopExecution;
 	std::thread thread;
 
 	std::shared_ptr<Notification> Find(uint32_t hNotify);
+	std::vector<std::shared_ptr<SyntheticNotification> >
+	FindSynthetic(uint32_t type);
 };
 using SharedDispatcher = std::shared_ptr<NotificationDispatcher>;

AdsLib/RingBuffer.h

@@ -10,6 +10,8 @@
 #include <memory>
 
 struct RingBuffer {
+	friend struct RingBufferTransaction;
+
 	RingBuffer(size_t N)
 		: dataSize(N + 1)
 		, data(new uint8_t[N + 1])

AdsLib/RingBufferTransaction.h

@@ -0,0 +1,45 @@
+// SPDX-License-Identifier: MIT
+
+#pragma once
+
+#include "RingBuffer.h"
+
+#include <cassert>
+#include <cstdint>
+
+struct RingBufferTransaction {
+	RingBufferTransaction(RingBuffer &target)
+		: buffer(target)
+		, write(target.write)
+	{
+	}
+
+	size_t BytesFree() const
+	{
+		return (write < buffer.read) ?
+			       buffer.read - write - 1 :
+			       buffer.dataSize - 1 - (write - buffer.read);
+	}
+
+	size_t WriteChunk() const
+	{
+		return (write < buffer.read) ?
+			       buffer.read - write - 1 :
+			       buffer.data.get() + buffer.dataSize - write -
+				       (buffer.data.get() == buffer.read);
+	}
+
+	void Write(size_t n)
+	{
+		assert(n <= BytesFree());
+		write = buffer.Increment(write, n);
+	}
+
+	void Commit()
+	{
+		buffer.write = write;
+	}
+
+	RingBuffer &buffer;
+	uint8_t *write;
+};

AdsLib/standalone/AdsDef.h

@@ -279,6 +279,11 @@ enum AMSPORT : uint16_t {
 #define ADSERR_CLIENT_SYNCPORTLOCKED \
 	(0x55 + ERR_ADSERRS) /**< sync port is locked */
 
+////////////////////////////////////////////////////////////////////////////////
+// Synthetic notification types
+#define NOTIFY_NOTIFICATION_RCV 0x02 /**< notification received notification */
+#define NOTIFY_CONNECTION_LOST 0x01 /**< connection lost notification */
+
 #pragma pack(push, 1)
 
 /**
@@ -427,6 +432,16 @@ typedef void (*PAdsNotificationFuncEx)(
 	const AmsAddr *pAddr, const AdsNotificationHeader *pNotification,
 	uint32_t hUser);
 
+/**
+ * @brief Type definition of the callback function required by the AdsSyncAddSyntheticDeviceNotificationReqEx() function.
+ * @param[in] pAddr Structure with NetId and port number of the ADS server.
+ * @param[in] hNotification Handle for the notification. Is specified when the notification is defined.
+ * @param[in] hUser custom handle pass to AdsSyncAddSyntheticDeviceNotificationReqEx() during registration
+ */
+typedef void (*PAdsSyntheticNotificationFuncEx)(
+	const AmsAddr *pAddr, uint32_t hNotification,
+	uint32_t hUser);
+
 #define ADSSYMBOLFLAG_PERSISTENT ((uint32_t)(1 << 0))
 #define ADSSYMBOLFLAG_BITVALUE ((uint32_t)(1 << 1))
 #define ADSSYMBOLFLAG_REFERENCETO ((uint32_t)(1 << 2))

AdsLib/standalone/AdsLib.cpp

@@ -272,6 +272,39 @@ long AdsSyncDelDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
 					   hNotification);
 }
 
+long AdsAddSyntheticDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
+				       uint32_t nType,
+				       PAdsSyntheticNotificationFuncEx pFunc,
+				       uint32_t hUser, uint32_t *pNotification)
+{
+	ASSERT_PORT_AND_AMSADDR(port, pAddr);
+	if (!pFunc || !pNotification) {
+		return ADSERR_CLIENT_INVALIDPARM;
+	}
+
+	if (nType != NOTIFY_NOTIFICATION_RCV &&
+	    nType != NOTIFY_CONNECTION_LOST) {
+		return ADSERR_CLIENT_INVALIDPARM;
+	}
+
+	try {
+		auto notify = std::make_shared<SyntheticNotification>(
+			pFunc, hUser, *pAddr, (uint16_t)port, nType);
+		return GetRouter().AddSyntheticNotification(
+			(uint16_t)port, *pAddr, pNotification, notify);
+	} catch (const std::bad_alloc &) {
+		return GLOBALERR_NO_MEMORY;
+	}
+}
+
+long AdsDelSyntheticDeviceNotificationReqEx(long port, const AmsAddr *pAddr,
+				       uint32_t hNotification)
+{
+	ASSERT_PORT_AND_AMSADDR(port, pAddr);
+	return GetRouter().DelSyntheticNotification((uint16_t)port, pAddr,
+						    hNotification);
+}
+
 long AdsSyncGetTimeoutEx(long port, uint32_t *timeout)
 {
 	ASSERT_PORT(port);

AdsLib/standalone/AmsConnection.cpp

@@ -5,6 +5,7 @@
 
 #include "AmsConnection.h"
 #include "Log.h"
+#include "RingBufferTransaction.h"
 
 AmsResponse::AmsResponse()
 	: request(nullptr)
@@ -47,10 +48,13 @@ AmsConnection::DispatcherListAdd(const VirtualConnection &connection)
 	std::lock_guard<std::recursive_mutex> lock(dispatcherListMutex);
 	return dispatcherList
 		.emplace(connection,
-			 std::make_shared<NotificationDispatcher>(std::bind(
-				 &AmsConnection::DeleteNotification, this,
-				 connection.second, std::placeholders::_1,
-				 std::placeholders::_2, connection.first)))
+			 std::make_shared<NotificationDispatcher>(
+				 connection,
+				 std::bind(&AmsConnection::DeleteNotification,
+					   this, connection.second,
+					   std::placeholders::_1,
+					   std::placeholders::_2,
+					   connection.first)))
 		.first->second;
 }
 
@@ -93,6 +97,16 @@ AmsConnection::CreateNotifyMapping(uint32_t hNotify,
 	return dispatcher;
 }
 
+SharedDispatcher
+AmsConnection::CreateSyntheticNotifyMapping(uint32_t hNotify,
+				   std::shared_ptr<SyntheticNotification> notification)
+{
+	auto dispatcher = DispatcherListAdd(notification->connection);
+	notification->hNotify(hNotify);
+	dispatcher->EmplaceSynthetic(hNotify, notification);
+	return dispatcher;
+}
+
 long AmsConnection::DeleteNotification(const AmsAddr &amsAddr, uint32_t hNotify,
 				       uint32_t tmms, uint16_t port)
 {
@@ -291,7 +305,7 @@ bool AmsConnection::ReceiveNotification(const AoEHeader &header)
 		return false;
 	}
 
-	auto &ring = dispatcher->ring;
+	auto ring = RingBufferTransaction(dispatcher->ring);
 	auto bytesLeft = header.length();
 	if (bytesLeft + sizeof(bytesLeft) > ring.BytesFree()) {
 		ReceiveJunk(bytesLeft);
@@ -316,6 +330,8 @@ bool AmsConnection::ReceiveNotification(const AoEHeader &header)
 	}
 	Receive(ring.write, bytesLeft);
 	ring.Write(bytesLeft);
+
+	ring.Commit();
 	dispatcher->Notify();
 	return true;
 }
@@ -325,6 +341,9 @@ void AmsConnection::TryRecv()
 	try {
 		Recv();
 	} catch (const std::runtime_error &e) {
+		for (auto &dispatcher : dispatcherList) {
+			dispatcher.second->Notify();
+		}
 		LOG_INFO(e.what());
 	}
 }