it compiles with new TCP socket

reflector/BMProtocol.cpp

@@ -27,7 +27,7 @@
 
 bool CBMProtocol::Initialize(const char *type, const EProtocol ptype, const uint16_t port, const bool has_ipv4, const bool has_ipv6)
 {
-	m_HasTranscoder = g_Configure.IsString(g_Keys.modules.tcmodules);
+	m_HasTranscoder = (0 != g_Configure.GetUnsigned(g_Keys.tc.port));
 	if (! CProtocol::Initialize(type, ptype, port, has_ipv4, has_ipv6))
 		return false;
 

reflector/Buffer.h

@@ -20,6 +20,7 @@
 
 #include <vector>
 #include <fstream>
+#include <cstdint>
 
 ////////////////////////////////////////////////////////////////////////////////////////
 

reflector/CodecStream.cpp

@@ -20,13 +20,12 @@
 #include <string.h>
 #include <sys/select.h>
 
+#include "Global.h"
 #include "DVFramePacket.h"
 #include "PacketStream.h"
 #include "CodecStream.h"
 #include "Reflector.h"
 
-extern CReflector g_Reflector;
-
 ////////////////////////////////////////////////////////////////////////////////////////
 // constructor
 
@@ -110,6 +109,21 @@ void CCodecStream::Thread()
 
 void CCodecStream::Task(void)
 {
+	int fd = g_TCServer.GetFD(m_CSModule);
+	if (fd < 0) // log the situation
+		std::cout << "Lost connection to transcoder, module '" << m_CSModule << "', waiting..." << std::endl;
+	while (fd < 0)
+	{
+		if (g_TCServer.Accept()) // we will block until we have a new connection
+		{
+			std::cerr << "UNRECOVERABLE ERROR!" << std::endl;
+			exit(1);
+		}
+		// Accept was sucessful, but we need to make sure THIS MODULE was reestablished
+		// It's possile that other Transcoder ports were lost
+		fd = g_TCServer.GetFD(m_CSModule);
+	}
+
 	STCPacket pack;
     struct timeval tv;
     fd_set readfds;
@@ -117,15 +131,14 @@ void CCodecStream::Task(void)
     tv.tv_sec = 0;
     tv.tv_usec = 7000;
 
-	int fd = g_Reflector.tcServer.GetFD(m_CSModule);
-
     FD_ZERO(&readfds);
     FD_SET(fd, &readfds);
 
     // don't care about writefds and exceptfds:
     if (select(fd+1, &readfds, NULL, NULL, &tv))
 	{
-		g_Reflector.tcServer.Receive(fd, &pack);
+		if (g_TCServer.Receive(fd, &pack))
+			return;
 		// update statistics
 		double rt = pack.rt_timer.time();	// the round-trip time
 		if (0 == m_RTCount)
@@ -151,18 +164,17 @@ void CCodecStream::Task(void)
 		{
 			// pop the original packet
 			auto Packet = m_LocalQueue.Pop();
-			auto Frame = static_cast<CDvFramePacket *>(Packet.get());
 
 			// make sure this is the correct packet
-			if ((pack.streamid == Frame->GetCodecPacket()->streamid) && (pack.sequence == Frame->GetCodecPacket()->sequence))
+			if ((pack.streamid == Packet->GetCodecPacket()->streamid) && (pack.sequence == Packet->GetCodecPacket()->sequence))
 			{
 				// update content with transcoded data
-				Frame->SetCodecData(&pack);
+				Packet->SetCodecData(&pack);
 				// mark the DStar sync frames if the source isn't dstar
-				if (ECodecType::dstar!=Frame->GetCodecIn() && 0==Frame->GetPacketId()%21)
+				if (ECodecType::dstar!=Packet->GetCodecIn() && 0==Packet->GetPacketId()%21)
 				{
 					const uint8_t DStarSync[] = { 0x55, 0x2D, 0x16 };
-					Frame->SetDvData(DStarSync);
+					Packet->SetDvData(DStarSync);
 				}
 
 				// and push it back to client
@@ -172,10 +184,10 @@ void CCodecStream::Task(void)
 			{
 				// Not the correct packet! It will be ignored
 				// Report it
-				if (pack.streamid != Frame->GetCodecPacket()->streamid)
-					std::cerr << std::hex  << std::showbase << "StreamID mismatch: this voice frame=" << ntohs(Frame->GetCodecPacket()->streamid) << " returned transcoder packet=" << ntohs(pack.streamid) << std::dec << std::noshowbase << std::endl;
-				if (pack.sequence != Frame->GetCodecPacket()->sequence)
-					std::cerr << "Sequence mismatch: this voice frame=" << Frame->GetCodecPacket()->sequence << " returned transcoder packet=" << pack.sequence << std::endl;
+				if (pack.streamid != Packet->GetCodecPacket()->streamid)
+					std::cerr << std::hex  << std::showbase << "StreamID mismatch: this voice frame=" << ntohs(Packet->GetCodecPacket()->streamid) << " returned transcoder packet=" << ntohs(pack.streamid) << std::dec << std::noshowbase << std::endl;
+				if (pack.sequence != Packet->GetCodecPacket()->sequence)
+					std::cerr << "Sequence mismatch: this voice frame=" << Packet->GetCodecPacket()->sequence << " returned transcoder packet=" << pack.sequence << std::endl;
 			}
 		}
 		else
@@ -185,12 +197,10 @@ void CCodecStream::Task(void)
 		}
 	}
 
-	// anything in our queue
-	auto Packet = m_Queue.Pop();
-	while (Packet)
+	// anything in our queue, then get it to the transcoder!
+	while (! m_Queue.IsEmpty())
 	{
-		// we need a CDvFramePacket pointer to access Frame stuff
-		auto Frame = (CDvFramePacket *)Packet.get();
+		auto &Frame = m_Queue.Front();
 
 		if (m_IsOpen)
 		{
@@ -199,13 +209,22 @@ void CCodecStream::Task(void)
 			Frame->SetTCParams(m_uiTotalPackets++);
 
 			// now send to transcoder
-			g_Reflector.tcServer.Send(Frame->GetCodecPacket());
+			int fd = g_TCServer.GetFD(Frame->GetCodecPacket()->module);
+			if (fd < 0)
+			{
+				// Crap! We've lost connection to the transcoder!
+				// we'll try to fix this on the next pass
+				return;
+			}
 
-			// push to our local queue where it can wait for the transcoder
-			m_LocalQueue.Push(std::move(Packet));
+			if (g_TCServer.Send(fd, Frame->GetCodecPacket()))
+			{
+				// ditto, we'll try to fix this on the next pass
+				return;
+			}
+			// the fd was good and then the send was successful, so...
+			// push the frame to our local queue where it can wait for the transcoder
+			m_LocalQueue.Push(std::move(m_Queue.Pop()));
 		}
-
-		// get the next packet, if there is one
-		Packet = m_Queue.Pop();
 	}
 }

reflector/CodecStream.h

@@ -50,7 +50,7 @@ public:
 	void Task(void);
 
 	// pass-through
-	void Push(std::unique_ptr<CPacket> p) { m_Queue.Push(std::move(p)); }
+	void Push(std::unique_ptr<CDvFramePacket> p) { m_Queue.Push(std::move(p)); }
 
 protected:
 	// identity
@@ -67,7 +67,7 @@ protected:
 	CPacketStream  *m_PacketStream;
 
 	// queues
-	CSafePacketQueue<std::unique_ptr<CPacket>> m_LocalQueue, m_Queue;
+	CSafePacketQueue<std::unique_ptr<CDvFramePacket>> m_LocalQueue, m_Queue;
 
 	// thread
 	std::atomic<bool> keep_running;

reflector/Configure.cpp

@@ -305,6 +305,7 @@ bool CConfigure::ReadData(const std::string &path)
 				}
 				else
 					badParam(key);
+				break;
 			case ESection::modules:
 				if (0 == key.compare(JMODULES))
 				{
@@ -669,6 +670,12 @@ bool CConfigure::ReadData(const std::string &path)
 				}
 			}
 		}
+		else
+		{
+			// there is no transcoder
+			data[g_Keys.tc.modules] = "";
+			data[g_Keys.tc.bind] = "";
+		}
 	}
 
 	// "simple" protocols with only a Port

reflector/Global.h

@@ -21,6 +21,7 @@
 #include "LookupDmr.h"
 #include "LookupNxdn.h"
 #include "LookupYsf.h"
+#include "TCSocket.h"
 #include "JsonKeys.h"
 
 extern CReflector  g_Reflector;
@@ -31,3 +32,4 @@ extern CLookupDmr  g_LDid;
 extern CLookupNxdn g_LNid;
 extern CLookupYsf  g_LYtr;
 extern SJsonKeys   g_Keys;
+extern CTCServer   g_TCServer;

reflector/Main.cpp

@@ -32,6 +32,7 @@ CVersion    g_Version(3,1,2); // The major byte should only change if the interl
 CLookupDmr  g_LDid;
 CLookupNxdn g_LNid;
 CLookupYsf  g_LYtr;
+CTCServer   g_TCServer;
 
 ////////////////////////////////////////////////////////////////////////////////////////
 

reflector/PacketStream.cpp

@@ -92,9 +92,11 @@ void CPacketStream::Push(std::unique_ptr<CPacket> Packet)
 	if ( m_CodecStream && Packet->IsDvFrame() && Packet->IsLocalOrigin())
 	{
 		// yes, push packet to trancoder queue
-		// trancoder will push it after transcoding
+		// first, recast to a CDvFramePacket
+		auto Frame = std::unique_ptr<CDvFramePacket>(static_cast<CDvFramePacket *>(Packet.release()));
+		// trancoder will push it to m_Queue after transcoding
 		// is completed
-		m_CodecStream->Push(std::move(Packet));
+		m_CodecStream->Push(std::move(Frame));
 	}
 	else
 	{

reflector/Reflector.cpp

@@ -70,7 +70,7 @@ bool CReflector::Start(void)
 
 	// init transcoder comms
 	if (port)
-		tcServer.Open(g_Configure.GetString(g_Keys.tc.bind), tcmods, port);
+		g_TCServer.Open(g_Configure.GetString(g_Keys.tc.bind), tcmods, port);
 
 	// init gate keeper. It can only return true!
 	g_GateKeeper.Init();
@@ -149,7 +149,7 @@ void CReflector::Stop(void)
 
 	// stop transcoder comms
 	// if it was never opened, then there is nothing to close;
-	tcServer.Close();
+	g_TCServer.Close();
 
 	// stop & delete report threads
 	if ( m_XmlReportFuture.valid() )

reflector/Reflector.h

@@ -25,7 +25,6 @@
 #include "Peers.h"
 #include "Protocols.h"
 #include "PacketStream.h"
-#include "TCTCPSocket.h"
 
 #ifndef NO_DHT
 #include "dht-values.h"
@@ -84,9 +83,6 @@ public:
 	void GetDHTConfig(const std::string &cs);
 #endif
 
-	// transcoder communication
-	CTCTCPServer tcServer;
-
 protected:
 #ifndef NO_DHT
 	// Publish DHT

reflector/SafePacketQueue.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <iostream>
 #include <queue>
 #include <mutex>
 #include <condition_variable>
@@ -43,6 +44,18 @@ public:
 		c.notify_one();
 	}
 
+	// You will die if the queue is empty!
+	T &Front(void)
+	{
+		std::lock_guard<std::mutex> lock(m);
+		if (q.empty())
+		{
+			std::cerr << "ERROR: CSavePacketQueue::Front() called, but queue is EMPTY!" << std::endl;
+			exit(1);
+		}
+		return q.front();
+	}
+
 	T Pop(void)
 	{
 		std::lock_guard<std::mutex> lock(m);

reflector/TCSocket.cpp

@@ -0,0 +1,294 @@
+// urfd -- The universal reflector
+// Copyright © 2024 Thomas A. Early N7TAE
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+#include <iostream>
+#include <unistd.h>
+#include <thread>
+#include <chrono>
+#include <sys/types.h>
+#include <sys/socket.h>
+
+#include "IP.h"
+#include "TCSocket.h"
+
+void CTCSocket::Close()
+{
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	for (auto item : m_FD)
+		close(item.second);
+	m_FD.clear();
+}
+
+void CTCSocket::Close(char mod)
+{
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	auto item = m_FD.find(mod);
+	if (m_FD.end() == item)
+	{
+		std::cerr << "Could not find a file descriptor for module '" << mod << "'" << std::endl;
+		return;
+	}
+	close(item->second);
+	m_FD.erase(item);
+}
+
+void CTCSocket::Close(int fd)
+{
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	for (auto &p : m_FD)
+	{
+		if (fd == p.second)
+		{
+			close(fd);
+			m_FD.erase(p.first);
+			return;
+		}
+	}
+	std::cerr << "Could not find a file descriptor with a value of " << fd << std::endl;
+}
+
+int CTCSocket::GetFD(char module) const
+{
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	const auto item = m_FD.find(module);
+	if (m_FD.cend() == item)
+	{
+		return -1;
+	}
+	return item->second;
+}
+
+char CTCSocket::GetMod(int fd) const
+{
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	for (const auto &p : m_FD)
+	{
+		if (fd == p.second)
+			return p.first;
+	}
+	return '?';
+}
+
+bool CTCSocket::Send(int fd, const STCPacket *packet)
+{
+	unsigned count = 0;
+	auto data = (const unsigned char *)packet;
+	do {
+		auto n = send(fd, data+count, sizeof(STCPacket)-count, 0);
+		if (n <= 0)
+		{
+			if (0 == n)
+			{
+				std::cerr << "CTCSocket::Send: socket on module '" << packet->module << "' has been closed!" << std::endl;
+			}
+			else
+			{
+				perror("CTCSocket::Send");
+			}
+			Close(packet->module);
+			return true;
+		}
+		count += n;
+	} while (count < sizeof(STCPacket));
+	return false;
+}
+
+bool CTCSocket::Receive(int fd, STCPacket *packet)
+{
+	auto data = (unsigned char *)packet;
+	auto n = recv(fd, data, sizeof(STCPacket), MSG_WAITALL);
+	if (n < 0)
+	{
+		perror("CTCSocket::Receive");
+		Close(fd);
+		return true;
+	}
+	return n == sizeof(STCPacket);
+}
+
+bool CTCServer::Open(const std::string &address, const std::string &modules, uint16_t port)
+{
+	m_Modules.assign(modules);
+
+	CIp ip(address.c_str(), AF_UNSPEC, SOCK_STREAM, port);
+
+	int fd = socket(ip.GetFamily(), SOCK_STREAM, 0);
+	if (fd < 0)
+	{
+		perror("Open socket");
+		return true;
+	}
+
+	int yes = 1;
+	int rv = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
+	if (rv < 0)
+	{
+		close(fd);
+		perror("Open setsockopt");
+		return true;
+	}
+
+	rv = bind(fd, ip.GetCPointer(), ip.GetSize());
+	if (rv < 0)
+	{
+		close(fd);
+		perror("Open bind");
+		return true;
+	}
+
+	rv = listen(fd, 3);
+	if (rv < 0)
+	{
+		perror("Open listen");
+		close(fd);
+		Close();
+		return true;
+	}
+
+	auto n = m_Modules.size();
+	std::cout << "Waiting for " << n << " tC connection(s)..." << std::endl;
+
+	while (m_FD.size() < n)
+	{
+		if (Accept())
+			return true;
+	}
+
+	m_listenSock = fd;
+
+	return false;
+}
+
+bool CTCServer::Accept()
+{
+	CIp their_addr; // connector's address information
+
+	socklen_t sin_size = sizeof(struct sockaddr_storage);
+
+	auto newfd = accept(m_listenSock, their_addr.GetPointer(), &sin_size);
+	if (newfd < 0)
+	{
+		if (EAGAIN == errno || EWOULDBLOCK == errno)
+			return false;
+		perror("accept");
+		return true;
+	}
+
+	char mod;
+	auto rv = recv(newfd, &mod, 1, 0);	// retrieve the identification byte
+	if (rv != 1)
+	{
+		if (rv < 0)
+			perror("accept recv");
+		else
+			std::cerr << "recv got no identification byte!" << std::endl;
+		close(newfd);
+		return true;
+	}
+
+	if (std::string::npos == m_Modules.find(mod))
+	{
+		std::cerr << "New connection for module '" << mod << "', but it's not configured!" << std::endl;
+		std::cerr << "The transcoded modules need to be configured identically for both urfd and tcd." << std::endl;
+		close(newfd);
+		return true;
+	}
+
+	std::cout << "File descriptor " << newfd << " opened TCP port for module '" << mod << "' on " << their_addr << std::endl;
+
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	m_FD[mod] = newfd;
+
+	return false;
+}
+
+bool CTCClient::Initialize(const std::string &address, const std::string &modules, uint16_t port)
+{
+	m_Address.assign(address);
+	m_Modules.assign(modules);
+	m_Port = port;
+
+	std::cout << "Connecting to the TCP server..." << std::endl;
+
+	for (char c : modules)
+	{
+		if (Connect(c))
+		{
+			return true;
+		}
+	}
+	return false;
+}
+
+bool CTCClient::Connect(char module)
+{
+	CIp ip(m_Address.c_str(), AF_UNSPEC, SOCK_STREAM, m_Port);
+
+	auto fd = socket(ip.GetFamily(), SOCK_STREAM, 0);
+	if (fd < 0)
+	{
+		std::cerr << "Could not open socket for module '" << module << "'" << std::endl;
+		perror("TC client socket");
+		return true;
+	}
+
+	int yes = 1;
+	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)))
+	{
+		perror("TC client setsockopt");
+		close(fd);
+		return true;
+	}
+
+	unsigned count = 0;
+	while (connect(fd, ip.GetCPointer(), ip.GetSize()))
+	{
+		if (ECONNREFUSED == errno)
+		{
+			if (0 == ++count % 100) std::cout << "Connection refused! Restart the server." << std::endl;
+			std::this_thread::sleep_for(std::chrono::milliseconds(100));
+		}
+		else
+		{
+			std::cerr << "For module '" << module << "' : ";
+			perror("connect");
+			close(fd);
+			return true;
+		}
+	}
+
+	int sent = send(fd, &module, 1, 0); // send the identification byte
+	if (sent < 0)
+	{
+		std::cerr << "Error sending ID byte to module '" << module << "':" << std::endl;
+		perror("send");
+		close(fd);
+		return true;
+	}
+	else if (0 == sent)
+	{
+		std::cerr << "Could not set ID byte to module '" << module << "'" << std::endl;
+		close(fd);
+		return true;
+	}
+
+	std::cout << "File descriptor " << fd << " on " << ip << " opened for module '" << module << "'" << std::endl;
+
+	std::lock_guard<std::mutex> lck(m_Mutex);
+	m_FD[module] = fd;
+
+	return false;
+}

reflector/TCSocket.h

@@ -18,43 +18,53 @@
 
 #include <string>
 #include <cstdint>
+#include <mutex>
 #include <unordered_map>
 
 #include "TCPacketDef.h"
 
-class CTCTCPSocket
+class CTCSocket
 {
 public:
-	CTCTCPSocket() {}
-	virtual ~CTCTCPSocket() { Close(); }
+	CTCSocket() {}
+	virtual ~CTCSocket() { Close(); }
 
 	void Close(); // close all open sockets
+	void Close(char module); // close a specific module
+	void Close(int fd); // close a specific file descriptor
 
 	// bool functions return true on failure
-	virtual bool Open(const std::string &address, const std::string &modules, uint16_t port) = 0;
-	bool Send(const STCPacket *packet);
+	bool Send(int fd, const STCPacket *packet);
 	bool Receive(int fd, STCPacket *packet);
 
 	int GetFD(char module) const; // can return -1!
+	char GetMod(int fd) const;
 
 protected:
-	void Close(char); // close a specific module
-
 	std::unordered_map<char, int> m_FD;
+	mutable std::mutex m_Mutex;
 };
 
-class CTCTCPServer : public CTCTCPSocket
+class CTCServer : public CTCSocket
 {
 public:
-	CTCTCPServer() : CTCTCPSocket() {}
-	~CTCTCPServer() {}
+	CTCServer() : CTCSocket() {}
+	~CTCServer() {}
 	bool Open(const std::string &address, const std::string &modules, uint16_t port);
+	bool Accept();
+private:
+	int m_listenSock;
+	std::string m_Modules;
 };
 
-class CTCTCPClient : public CTCTCPSocket
+class CTCClient : public CTCSocket
 {
 public:
-	CTCTCPClient() : CTCTCPSocket() {}
-	~CTCTCPClient() {}
-	bool Open(const std::string &address, const std::string &modules, uint16_t port);
+	CTCClient() : CTCSocket(), m_Port(0) {}
+	~CTCClient() {}
+	bool Initialize(const std::string &address, const std::string &modules, uint16_t port);
+	bool Connect(char module);
+private:
+	std::string m_Address, m_Modules;
+	uint16_t m_Port;
 };

reflector/TCTCPSocket.cpp

@@ -1,223 +0,0 @@
-// urfd -- The universal reflector
-// Copyright © 2024 Thomas A. Early N7TAE
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// This program 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 General Public License for more details.
-//
-// You should have received a copy of the GNU General Public License
-// along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-#include <iostream>
-#include <unistd.h>
-#include <thread>
-#include <chrono>
-#include <sys/types.h>
-#include <sys/socket.h>
-
-#include "IP.h"
-#include "TCTCPSocket.h"
-
-void CTCTCPSocket::Close()
-{
-	for (auto item : m_FD)
-		close(item.second);
-	m_FD.clear();
-}
-
-void CTCTCPSocket::Close(char mod)
-{
-	auto item = m_FD.find(mod);
-	if (m_FD.end() == item)
-		return;
-
-	close(item->second);
-	m_FD.erase(item);
-}
-
-bool CTCTCPSocket::Send(const STCPacket *packet)
-{
-	int fd = GetFD(packet->module);
-	if (fd < 0)
-	{
-		return true;
-	}
-
-	long count = 0;
-	auto data = (const unsigned char *)packet;
-	do {
-		auto n = send(fd, data+count, sizeof(STCPacket)-count, 0);
-		if (n <= 0)
-		{
-			if (0 == n)
-			{
-				std::cerr << "CTCTCPSocket::Send: socket on module '" << packet->module << "' has been closed!" << std::endl;
-			}
-			else
-			{
-				perror("CTCTCPSocket::Send");
-			}
-			Close(packet->module);
-			return true;
-		}
-		count += n;
-	} while (count < sizeof(STCPacket));
-	return false;
-}
-
-bool CTCTCPSocket::Receive(int fd, STCPacket *packet)
-{
-	auto data = (unsigned char *)packet;
-	auto n = recv(fd, data, sizeof(STCPacket), MSG_WAITALL);
-	if (n < 0)
-	{
-		perror("CTCTCPSocket::Receive");
-		return true;
-	}
-	return n == sizeof(STCPacket);
-}
-
-int CTCTCPSocket::GetFD(char module) const
-{
-	const auto item = m_FD.find(module);
-	if (m_FD.cend() == item)
-	{
-		return -1;
-	}
-	return item->second;
-}
-
-bool CTCTCPServer::Open(const std::string &address, const std::string &tcmodules, uint16_t port)
-{
-	int fd;
-	CIp ip(address.c_str(), AF_UNSPEC, SOCK_STREAM, port);
-
-	fd = socket(ip.GetFamily(), SOCK_STREAM, 0);
-	if (fd < 0)
-	{
-		perror("Open socket");
-		return true;
-	}
-
-	int yes = 1;
-	int rv = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
-	if (rv < 0)
-	{
-		close(fd);
-		perror("Open setsockopt");
-		return true;
-	}
-
-	rv = bind(fd, ip.GetCPointer(), ip.GetSize());
-	if (rv < 0)
-	{
-		close(fd);
-		perror("Open bind");
-		return true;
-	}
-
-	rv = listen(fd, 3);
-	if (rv < 0)
-	{
-		perror("Open listen");
-		close(fd);
-		Close();
-		return true;
-	}
-
-	std::cout << "Waiting for " << tcmodules.size() << " transcoder connection(s)..." << std::endl;
-	for (unsigned x=0; x<tcmodules.size(); x++)
-	{
-		CIp their_addr; // connector's address information
-
-		socklen_t sin_size = sizeof(struct sockaddr_storage);
-
-		auto newfd = accept(fd, their_addr.GetPointer(), &sin_size);
-		if (newfd < 0)
-		{
-			perror("Open accept");
-			Close();
-			return true;
-		}
-
-		char mod;
-		rv = recv(newfd, &mod, 1, 0);	// retrieve the identification byte
-		if (rv != 1)
-		{
-			if (rv < 0) perror("Open recv");
-			else        std::cerr << "recv got no identification byte!" << std::endl;
-			close(newfd);
-			Close();
-			return true;
-		}
-		if (std::string::npos == tcmodules.find(mod))
-		{
-			std::cerr << "New connection for module '" << mod << "', but it's not config'ed!" << std::endl;
-			std::cerr << "The transcoded modules need to be configured identically for both urfd and tcd." << std::endl;
-			close(newfd);
-			Close();
-			return true;
-		}
-
-		std::cout << "File descriptor " << newfd << " opened TCP port for module '" << mod << "' on " << their_addr << std::endl;
-		m_FD[mod] = newfd;
-	}
-	close(fd); // we don't need this anymore.
-	return false;
-}
-
-bool CTCTCPClient::Open(const std::string &address, const std::string &tcmodules, uint16_t port)
-{
-	std::cout << "Connecting to the TCP server..." << std::endl;
-	for (auto c : tcmodules)
-	{
-		CIp ip(address.c_str(), AF_UNSPEC, SOCK_STREAM, port);
-
-		auto fd = socket(ip.GetFamily(), SOCK_STREAM, 0);
-		if (fd < 0)
-		{
-			std::cout << "errno=" << errno << std::endl;
-			perror("Open socket");
-			Close();
-			return true;
-		}
-
-		int yes = 1;
-		if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)))
-		{
-			perror("Open setsockopt");
-			Close();
-			close(fd);
-			return true;
-		}
-
-		unsigned count = 0;
-		while (connect(fd, ip.GetCPointer(), ip.GetSize()))
-		{
-			if (ECONNREFUSED == errno)
-			{
-				if (0 == count++ % 15) std::cout << "Connection refused! Restart the server." << std::endl;
-				std::this_thread::sleep_for(std::chrono::seconds(2));
-			}
-			else
-			{
-				perror("Open connect");
-				Close();
-				close(fd);
-				return true;
-			}
-		}
-
-		std::cout << "File descriptor " << fd << " on " << ip << " opened for module '" << c << "'" << std::endl;
-		send(fd, &c, 1, 0); // send the identification byte
-		m_FD[c] = fd;
-	}
-
-	return false;
-}

reflector/URFProtocol.cpp

@@ -336,7 +336,7 @@ void CURFProtocol::HandlePeerLinks(void)
 							ok = false;
 							std::cerr << it->first << " has no module '" << c << "'" << std::endl;
 						}
-						else if ((std::string::npos == it->second.GetTCMods().find(c)) != (std::string::npos == g_Configure.GetString(g_Keys.modules.tcmodules).find(c)))
+						else if ((std::string::npos == it->second.GetTCMods().find(c)) != (std::string::npos == g_Configure.GetString(g_Keys.tc.modules).find(c)))
 						{	// are the transcoding states on both sides mismatched?
 							ok = false;
 							std::cerr << "The transcode states for module '" << c << "' don't match for this reflector and " << it->first << std::endl;