Fix Wires-X reply when reflector is busy and buffer overflow due to long Wires-X (WIP)

YSF2DMR/WiresX.cpp

@@ -69,7 +69,10 @@ m_status(WXSI_NONE),
 m_start(0U),
 m_search(),
 m_category(),
-m_makeUpper(makeUpper)
+m_makeUpper(makeUpper),
+m_busy(false),
+m_busyTimer(3000U, 1U),
+m_bufferTX(10000U, "YSF Wires-X TX Buffer")
 {
 	assert(network != NULL);
 
@@ -130,6 +133,8 @@ m_makeUpper(makeUpper)
 
 		::fclose(fp);
 	}
+
+	m_txWatch.start();
 }
 
 CWiresX::~CWiresX()
@@ -377,6 +382,9 @@ WX_STATUS CWiresX::processConnect(const unsigned char* source, const unsigned ch
 {
 	//::LogDebug("Received Connect to %6.6s from %10.10s", data, source);
 
+	m_busy = true;
+	m_busyTimer.start();
+
 	std::string id = std::string((char*)data, 6U);
 
 	m_dstID = atoi(id.c_str());
@@ -391,6 +399,9 @@ WX_STATUS CWiresX::processConnect(const unsigned char* source, const unsigned ch
 
 void CWiresX::processConnect(int dstID)
 {
+	m_busy = true;
+	m_busyTimer.start();
+
 	m_dstID = dstID;
 
 	m_status = WXSI_CONNECT;
@@ -408,6 +419,8 @@ void CWiresX::processDisconnect(const unsigned char* source)
 
 void CWiresX::clock(unsigned int ms)
 {
+	unsigned char buffer[200U];
+
 	m_timer.clock(ms);
 	if (m_timer.isRunning() && m_timer.hasExpired()) {
 		switch (m_status) {
@@ -435,6 +448,24 @@ void CWiresX::clock(unsigned int ms)
 		m_status = WXSI_NONE;
 		m_timer.stop();
 	}
+
+	if (m_txWatch.elapsed() > 90U) {
+		if (!m_bufferTX.isEmpty() && m_bufferTX.dataSize() >= 155U) {
+			unsigned char len = 0U;
+			m_bufferTX.getData(&len, 1U);
+			if (len == 155U) {
+				m_bufferTX.getData(buffer, 155U);
+				m_network->write(buffer);
+			}
+		}
+		m_txWatch.start();
+	}
+
+	m_busyTimer.clock(ms);
+	if (m_busyTimer.isRunning() && m_busyTimer.hasExpired()) {
+		m_busy = false;
+		m_busyTimer.stop();
+	}
 }
 
 void CWiresX::createReply(const unsigned char* data, unsigned int length)
@@ -483,7 +514,7 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 	buffer[34U] = seqNo;
 	seqNo += 2U;
 
-	m_network->write(buffer);
+	writeData(buffer);
 
 	fich.setFI(YSF_FI_COMMUNICATIONS);
 
@@ -530,7 +561,7 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 		buffer[34U] = seqNo;
 		seqNo += 2U;
 
-		m_network->write(buffer);
+		writeData(buffer);
 
 		fn++;
 		if (fn >= 8U) {
@@ -550,7 +581,15 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 
 	buffer[34U] = seqNo | 0x01U;
 
-	m_network->write(buffer);
+	writeData(buffer);
+}
+
+void CWiresX::writeData(const unsigned char* buffer)
+{
+	// Send host Wires-X reply using ring buffer
+	unsigned char len = 155U;
+	m_bufferTX.addData(&len, 1U);
+	m_bufferTX.addData(buffer, len);
 }
 
 unsigned char CWiresX::calculateFT(unsigned int length, unsigned int offset) const
@@ -1086,3 +1125,8 @@ void CWiresX::sendCategoryReply()
 
 	m_seqNo++;
 }
+
+bool CWiresX::isBusy() const
+{
+	return m_busy;
+}

YSF2DMR/WiresX.h

@@ -25,6 +25,8 @@
 #include "DMRNetwork.h"
 #include "Thread.h"
 #include "Timer.h"
+#include "StopWatch.h"
+#include "RingBuffer.h"
 
 #include <vector>
 #include <string>
@@ -70,6 +72,7 @@ public:
 	~CWiresX();
 
 	bool start();
+	bool isBusy() const;
 
 	WX_STATUS process(const unsigned char* data, const unsigned char* source, unsigned char fi, unsigned char dt, unsigned char fn, unsigned char ft);
 
@@ -111,6 +114,10 @@ private:
 	std::vector<CTGReg*> m_TGSearch;
 	std::vector<CTGReg*> m_category;
 	bool                 m_makeUpper;
+	bool                 m_busy;
+	CTimer               m_busyTimer;
+	CStopWatch           m_txWatch;
+	CRingBuffer<unsigned char> m_bufferTX;
 
 	WX_STATUS processConnect(const unsigned char* source, const unsigned char* data);
 	void processDX(const unsigned char* source);
@@ -124,6 +131,7 @@ private:
 	void sendCategoryReply();
 
 	void createReply(const unsigned char* data, unsigned int length);
+	void writeData(const unsigned char* data);
 	unsigned char calculateFT(unsigned int length, unsigned int offset) const;
 };
 

YSF2NXDN/WiresX.cpp

@@ -69,7 +69,10 @@ m_status(WXSI_NONE),
 m_start(0U),
 m_search(),
 m_category(),
-m_makeUpper(makeUpper)
+m_makeUpper(makeUpper),
+m_busy(false),
+m_busyTimer(3000U, 1U),
+m_bufferTX(10000U, "YSF Wires-X TX Buffer")
 {
 	assert(network != NULL);
 
@@ -128,6 +131,8 @@ m_makeUpper(makeUpper)
 
 		::fclose(fp);
 	}
+
+	m_txWatch.start();
 }
 
 CWiresX::~CWiresX()
@@ -348,6 +353,9 @@ WX_STATUS CWiresX::processConnect(const unsigned char* source, const unsigned ch
 {
 	//::LogDebug("Received Connect to %6.6s from %10.10s", data, source);
 
+	m_busy = true;
+	m_busyTimer.start();
+
 	std::string id = std::string((char*)data, 6U);
 
 	m_dstID = atoi(id.c_str());
@@ -362,6 +370,9 @@ WX_STATUS CWiresX::processConnect(const unsigned char* source, const unsigned ch
 
 void CWiresX::processConnect(int dstID)
 {
+	m_busy = true;
+	m_busyTimer.start();
+
 	m_dstID = dstID;
 
 	m_status = WXSI_CONNECT;
@@ -379,6 +390,8 @@ void CWiresX::processDisconnect(const unsigned char* source)
 
 void CWiresX::clock(unsigned int ms)
 {
+	unsigned char buffer[200U];
+
 	m_timer.clock(ms);
 	if (m_timer.isRunning() && m_timer.hasExpired()) {
 		switch (m_status) {
@@ -407,6 +420,24 @@ void CWiresX::clock(unsigned int ms)
 		m_status = WXSI_NONE;
 		m_timer.stop();
 	}
+
+	if (m_txWatch.elapsed() > 90U) {
+		if (!m_bufferTX.isEmpty() && m_bufferTX.dataSize() >= 155U) {
+			unsigned char len = 0U;
+			m_bufferTX.getData(&len, 1U);
+			if (len == 155U) {
+				m_bufferTX.getData(buffer, 155U);
+				m_network->write(buffer);
+			}
+		}
+		m_txWatch.start();
+	}
+
+	m_busyTimer.clock(ms);
+	if (m_busyTimer.isRunning() && m_busyTimer.hasExpired()) {
+		m_busy = false;
+		m_busyTimer.stop();
+	}
 }
 
 void CWiresX::createReply(const unsigned char* data, unsigned int length)
@@ -455,7 +486,7 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 	buffer[34U] = seqNo;
 	seqNo += 2U;
 
-	m_network->write(buffer);
+	writeData(buffer);
 
 	fich.setFI(YSF_FI_COMMUNICATIONS);
 
@@ -502,7 +533,7 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 		buffer[34U] = seqNo;
 		seqNo += 2U;
 
-		m_network->write(buffer);
+		writeData(buffer);
 
 		fn++;
 		if (fn >= 8U) {
@@ -522,7 +553,15 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 
 	buffer[34U] = seqNo | 0x01U;
 
-	m_network->write(buffer);
+	writeData(buffer);
+}
+
+void CWiresX::writeData(const unsigned char* buffer)
+{
+	// Send host Wires-X reply using ring buffer
+	unsigned char len = 155U;
+	m_bufferTX.addData(&len, 1U);
+	m_bufferTX.addData(buffer, len);
 }
 
 unsigned char CWiresX::calculateFT(unsigned int length, unsigned int offset) const
@@ -1037,3 +1076,8 @@ void CWiresX::sendCategoryReply()
 
 	m_seqNo++;
 }
+
+bool CWiresX::isBusy() const
+{
+	return m_busy;
+}

YSF2NXDN/WiresX.h

@@ -24,6 +24,8 @@
 #include "YSFNetwork.h"
 #include "Thread.h"
 #include "Timer.h"
+#include "StopWatch.h"
+#include "RingBuffer.h"
 
 #include <vector>
 #include <string>
@@ -67,6 +69,7 @@ public:
 	~CWiresX();
 
 	bool start();
+	bool isBusy() const;
 
 	WX_STATUS process(const unsigned char* data, const unsigned char* source, unsigned char fi, unsigned char dt, unsigned char fn, unsigned char ft);
 
@@ -105,6 +108,10 @@ private:
 	std::vector<CTGReg*> m_TGSearch;
 	std::vector<CTGReg*> m_category;
 	bool                 m_makeUpper;
+	bool                 m_busy;
+	CTimer               m_busyTimer;
+	CStopWatch           m_txWatch;
+	CRingBuffer<unsigned char> m_bufferTX;
 
 	WX_STATUS processConnect(const unsigned char* source, const unsigned char* data);
 	void processDX(const unsigned char* source);
@@ -118,6 +125,7 @@ private:
 	void sendCategoryReply();
 
 	void createReply(const unsigned char* data, unsigned int length);
+	void writeData(const unsigned char* data);
 	unsigned char calculateFT(unsigned int length, unsigned int offset) const;
 };
 

YSF2P25/WiresX.cpp

@@ -69,7 +69,10 @@ m_status(WXSI_NONE),
 m_start(0U),
 m_search(),
 m_category(),
-m_makeUpper(makeUpper)
+m_makeUpper(makeUpper),
+m_busy(false),
+m_busyTimer(3000U, 1U),
+m_bufferTX(10000U, "YSF Wires-X TX Buffer")
 {
 	assert(network != NULL);
 
@@ -128,6 +131,8 @@ m_makeUpper(makeUpper)
 
 		::fclose(fp);
 	}
+
+	m_txWatch.start();
 }
 
 CWiresX::~CWiresX()
@@ -348,6 +353,9 @@ WX_STATUS CWiresX::processConnect(const unsigned char* source, const unsigned ch
 {
 	//::LogDebug("Received Connect to %6.6s from %10.10s", data, source);
 
+	m_busy = true;
+	m_busyTimer.start();
+
 	std::string id = std::string((char*)data, 6U);
 
 	m_dstID = atoi(id.c_str());
@@ -362,6 +370,9 @@ WX_STATUS CWiresX::processConnect(const unsigned char* source, const unsigned ch
 
 void CWiresX::processConnect(int dstID)
 {
+	m_busy = true;
+	m_busyTimer.start();
+
 	m_dstID = dstID;
 
 	m_status = WXSI_CONNECT;
@@ -379,6 +390,8 @@ void CWiresX::processDisconnect(const unsigned char* source)
 
 void CWiresX::clock(unsigned int ms)
 {
+	unsigned char buffer[200U];
+
 	m_timer.clock(ms);
 	if (m_timer.isRunning() && m_timer.hasExpired()) {
 		switch (m_status) {
@@ -407,6 +420,24 @@ void CWiresX::clock(unsigned int ms)
 		m_status = WXSI_NONE;
 		m_timer.stop();
 	}
+
+	if (m_txWatch.elapsed() > 90U) {
+		if (!m_bufferTX.isEmpty() && m_bufferTX.dataSize() >= 155U) {
+			unsigned char len = 0U;
+			m_bufferTX.getData(&len, 1U);
+			if (len == 155U) {
+				m_bufferTX.getData(buffer, 155U);
+				m_network->write(buffer);
+			}
+		}
+		m_txWatch.start();
+	}
+
+	m_busyTimer.clock(ms);
+	if (m_busyTimer.isRunning() && m_busyTimer.hasExpired()) {
+		m_busy = false;
+		m_busyTimer.stop();
+	}
 }
 
 void CWiresX::createReply(const unsigned char* data, unsigned int length)
@@ -455,7 +486,7 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 	buffer[34U] = seqNo;
 	seqNo += 2U;
 
-	m_network->write(buffer);
+	writeData(buffer);
 
 	fich.setFI(YSF_FI_COMMUNICATIONS);
 
@@ -502,7 +533,7 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 		buffer[34U] = seqNo;
 		seqNo += 2U;
 
-		m_network->write(buffer);
+		writeData(buffer);
 
 		fn++;
 		if (fn >= 8U) {
@@ -522,7 +553,15 @@ void CWiresX::createReply(const unsigned char* data, unsigned int length)
 
 	buffer[34U] = seqNo | 0x01U;
 
-	m_network->write(buffer);
+	writeData(buffer);
+}
+
+void CWiresX::writeData(const unsigned char* buffer)
+{
+	// Send host Wires-X reply using ring buffer
+	unsigned char len = 155U;
+	m_bufferTX.addData(&len, 1U);
+	m_bufferTX.addData(buffer, len);
 }
 
 unsigned char CWiresX::calculateFT(unsigned int length, unsigned int offset) const
@@ -1037,3 +1076,8 @@ void CWiresX::sendCategoryReply()
 
 	m_seqNo++;
 }
+
+bool CWiresX::isBusy() const
+{
+	return m_busy;
+}

YSF2P25/WiresX.h

@@ -24,6 +24,8 @@
 #include "YSFNetwork.h"
 #include "Thread.h"
 #include "Timer.h"
+#include "StopWatch.h"
+#include "RingBuffer.h"
 
 #include <vector>
 #include <string>
@@ -67,6 +69,7 @@ public:
 	~CWiresX();
 
 	bool start();
+	bool isBusy() const;
 
 	WX_STATUS process(const unsigned char* data, const unsigned char* source, unsigned char fi, unsigned char dt, unsigned char fn, unsigned char ft);
 
@@ -105,6 +108,10 @@ private:
 	std::vector<CTGReg*> m_TGSearch;
 	std::vector<CTGReg*> m_category;
 	bool                 m_makeUpper;
+	bool                 m_busy;
+	CTimer               m_busyTimer;
+	CStopWatch           m_txWatch;
+	CRingBuffer<unsigned char> m_bufferTX;
 
 	WX_STATUS processConnect(const unsigned char* source, const unsigned char* data);
 	void processDX(const unsigned char* source);
@@ -118,6 +125,7 @@ private:
 	void sendCategoryReply();
 
 	void createReply(const unsigned char* data, unsigned int length);
+	void writeData(const unsigned char* data);
 	unsigned char calculateFT(unsigned int length, unsigned int offset) const;
 };