uvos/pipewirestreamwatcher
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: uvos:4b0b5847a7deec0c3a9d77afd62f85b16f126725
Branches Tags
uvos:master
..
compare: uvos:7511e68c912ab8bfe4d53f94d35c7e73f1e1d3fb
Branches Tags
uvos:master

No commits in common. "4b0b5847a7deec0c3a9d77afd62f85b16f126725" and "7511e68c912ab8bfe4d53f94d35c7e73f1e1d3fb" have entirely different histories.
4b0b5847a7 ... 7511e68c91

3 changed files with 487 additions and 592 deletions
Download patch file Download diff file Expand all files Collapse all files

162
Socket.cpp
View file

@ -19,34 +19,28 @@ using namespace std;
// SocketException Code
SocketException::SocketException(const string &message, bool inclSysMsg)
: userMessage(message)
{
if (inclSysMsg)
{
: userMessage(message) {
if (inclSysMsg) {
userMessage.append(": ");
userMessage.append(strerror(errno));
}
}
SocketException::~SocketException() noexcept (true)
{
SocketException::~SocketException() noexcept (true) {
}
const char *SocketException::what()
{
const char *SocketException::what(){
return userMessage.c_str();
}
// Function to fill in address structure given an address and port
static void fillAddr(const string &address, unsigned short port,
sockaddr_in &addr)
{
sockaddr_in &addr) {
memset(&addr, 0, sizeof(addr)); // Zero out address structure
addr.sin_family = AF_INET; // Internet address
hostent *host; // Resolve name
if ((host = gethostbyname(address.c_str())) == NULL)
{
if ((host = gethostbyname(address.c_str())) == NULL) {
// strerror() will not work for gethostbyname() and hstrerror()
// is supposedly obsolete
throw SocketException("Failed to resolve name (gethostbyname())");
@ -58,18 +52,15 @@ static void fillAddr(const string &address, unsigned short port,
// Socket Code
Socket::Socket(int type, int protocol)
{
Socket::Socket(int type, int protocol) {
// Make a new socket
if ((sockDesc = socket(PF_INET, type, protocol)) < 0)
{
if ((sockDesc = socket(PF_INET, type, protocol)) < 0) {
throw SocketException("Socket creation failed (socket())", true);
}
}
Socket::Socket(int sockDesc)
{
Socket::Socket(int sockDesc) {
this->sockDesc = sockDesc;
}
@ -79,32 +70,27 @@ Socket::~Socket()
sockDesc = -1;
}
string Socket::getLocalAddress()
{
string Socket::getLocalAddress() {
sockaddr_in addr;
unsigned int addr_len = sizeof(addr);
if (getsockname(sockDesc, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0)
{
if (getsockname(sockDesc, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0) {
throw SocketException("Fetch of local address failed (getsockname())", true);
}
return inet_ntoa(addr.sin_addr);
}
unsigned short Socket::getLocalPort()
{
unsigned short Socket::getLocalPort() {
sockaddr_in addr;
unsigned int addr_len = sizeof(addr);
if (getsockname(sockDesc, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0)
{
if (getsockname(sockDesc, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0) {
throw SocketException("Fetch of local port failed (getsockname())", true);
}
return ntohs(addr.sin_port);
}
void Socket::setLocalPort(unsigned short localPort)
{
void Socket::setLocalPort(unsigned short localPort) {
// Bind the socket to its port
sockaddr_in localAddr;
memset(&localAddr, 0, sizeof(localAddr));
@ -112,21 +98,18 @@ void Socket::setLocalPort(unsigned short localPort)
localAddr.sin_addr.s_addr = htonl(INADDR_ANY);
localAddr.sin_port = htons(localPort);
if (bind(sockDesc, (sockaddr *) &localAddr, sizeof(sockaddr_in)) < 0)
{
if (bind(sockDesc, (sockaddr *) &localAddr, sizeof(sockaddr_in)) < 0) {
throw SocketException("Set of local port failed (bind())", true);
}
}
void Socket::setLocalAddressAndPort(const string &localAddress,
unsigned short localPort)
{
unsigned short localPort) {
// Get the address of the requested host
sockaddr_in localAddr;
fillAddr(localAddress, localPort, localAddr);
if (bind(sockDesc, (sockaddr *) &localAddr, sizeof(sockaddr_in)) < 0)
{
if (bind(sockDesc, (sockaddr *) &localAddr, sizeof(sockaddr_in)) < 0) {
throw SocketException("Set of local address and port failed (bind())", true);
}
}
@ -149,21 +132,17 @@ void Socket::setKeepalive()
void Socket::setBlocking(bool flag)
{
int flags = fcntl(sockDesc, F_GETFL, 0);
if( !flag )
flags = flags | O_NONBLOCK;
else
flags = flags & ~O_NONBLOCK;
if( !flag ) flags = flags | O_NONBLOCK;
else flags = flags & ~O_NONBLOCK;
fcntl(sockDesc, F_SETFL, flags);
}
void Socket::cleanUp()
{
void Socket::cleanUp() {
}
unsigned short Socket::resolveService(const string &service,
const string &protocol)
{
const string &protocol) {
struct servent *serv; /* Structure containing service information */
if ((serv = getservbyname(service.c_str(), protocol.c_str())) == NULL)
@ -175,32 +154,27 @@ unsigned short Socket::resolveService(const string &service,
// CommunicatingSocket Code
CommunicatingSocket::CommunicatingSocket(int type, int protocol)
: Socket(type, protocol)
{
: Socket(type, protocol) {
}
CommunicatingSocket::CommunicatingSocket(int newConnSD) : Socket(newConnSD)
{
CommunicatingSocket::CommunicatingSocket(int newConnSD) : Socket(newConnSD) {
}
void CommunicatingSocket::connect(const string &foreignAddress,
unsigned short foreignPort)
{
unsigned short foreignPort) {
// Get the address of the requested host
sockaddr_in destAddr;
fillAddr(foreignAddress, foreignPort, destAddr);
// Try to connect to the given port
if (::connect(sockDesc, (sockaddr *) &destAddr, sizeof(destAddr)) < 0)
{
if (::connect(sockDesc, (sockaddr *) &destAddr, sizeof(destAddr)) < 0) {
throw SocketException("Connect failed (connect())", true);
}
}
void CommunicatingSocket::send(const void *buffer, int bufferLen)
{
if (::send(sockDesc, (raw_type *) buffer, bufferLen, 0) < 0)
{
if (::send(sockDesc, (raw_type *) buffer, bufferLen, 0) < 0) {
throw SocketException("Send failed (send())", true);
}
}
@ -210,10 +184,8 @@ int CommunicatingSocket::recv(void *buffer, int bufferLen)
int rtn;
if ((rtn = ::recv(sockDesc, (raw_type *) buffer, bufferLen, MSG_DONTWAIT)) < 0 )
{
if(errno == EWOULDBLOCK || errno == EAGAIN)
return -1;
else
throw SocketException("Received failed (recv())", true);
if(errno == EWOULDBLOCK || errno == EAGAIN) return -1;
else throw SocketException("Received failed (recv())", true);
}
return rtn;
@ -224,20 +196,17 @@ string CommunicatingSocket::getForeignAddress()
sockaddr_in addr;
unsigned int addr_len = sizeof(addr);
if (getpeername(sockDesc, (sockaddr *) &addr,(socklen_t *) &addr_len) < 0)
{
if (getpeername(sockDesc, (sockaddr *) &addr,(socklen_t *) &addr_len) < 0) {
throw SocketException("Fetch of foreign address failed (getpeername())", true);
}
return inet_ntoa(addr.sin_addr);
}
unsigned short CommunicatingSocket::getForeignPort()
{
unsigned short CommunicatingSocket::getForeignPort() {
sockaddr_in addr;
unsigned int addr_len = sizeof(addr);
if (getpeername(sockDesc, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0)
{
if (getpeername(sockDesc, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0) {
throw SocketException("Fetch of foreign port failed (getpeername())", true);
}
return ntohs(addr.sin_port);
@ -247,20 +216,16 @@ unsigned short CommunicatingSocket::getForeignPort()
TCPSocket::TCPSocket()
: CommunicatingSocket(SOCK_STREAM,
IPPROTO_TCP)
{
IPPROTO_TCP) {
}
TCPSocket::TCPSocket(const string &foreignAddress, unsigned short foreignPort, bool keepalive)
: CommunicatingSocket(SOCK_STREAM, IPPROTO_TCP)
{
: CommunicatingSocket(SOCK_STREAM, IPPROTO_TCP) {
connect(foreignAddress, foreignPort);
if(keepalive)
setKeepalive();
if(keepalive) setKeepalive();
}
TCPSocket::TCPSocket(int newConnSD) : CommunicatingSocket(newConnSD)
{
TCPSocket::TCPSocket(int newConnSD) : CommunicatingSocket(newConnSD) {
}
// TCPServerSocket Code
@ -292,18 +257,15 @@ TCPSocket* TCPServerSocket::accept()
if(newConnSD > 0)
{
newSocket = new TCPSocket(newConnSD);
if(keepalive)
newSocket->setKeepalive();
if(keepalive) newSocket->setKeepalive();
}
return newSocket;
}
void TCPServerSocket::setListen(int queueLen)
{
if (listen(sockDesc, queueLen) < 0)
{
void TCPServerSocket::setListen(int queueLen) {
if (listen(sockDesc, queueLen) < 0) {
throw SocketException("Set listening socket failed (listen())", true);
}
}
@ -311,27 +273,23 @@ void TCPServerSocket::setListen(int queueLen)
// UDPSocket Code
UDPSocket::UDPSocket() : CommunicatingSocket(SOCK_DGRAM,
IPPROTO_UDP)
{
IPPROTO_UDP) {
setBroadcast();
}
UDPSocket::UDPSocket(unsigned short localPort) :
CommunicatingSocket(SOCK_DGRAM, IPPROTO_UDP)
{
CommunicatingSocket(SOCK_DGRAM, IPPROTO_UDP) {
setLocalPort(localPort);
setBroadcast();
}
UDPSocket::UDPSocket(const string &localAddress, unsigned short localPort)
: CommunicatingSocket(SOCK_DGRAM, IPPROTO_UDP)
{
: CommunicatingSocket(SOCK_DGRAM, IPPROTO_UDP) {
setLocalAddressAndPort(localAddress, localPort);
setBroadcast();
}
void UDPSocket::setBroadcast()
{
void UDPSocket::setBroadcast() {
// If this fails, we'll hear about it when we try to send. This will allow
// system that cannot broadcast to continue if they don't plan to broadcast
int broadcastPermission = 1;
@ -339,17 +297,14 @@ void UDPSocket::setBroadcast()
(raw_type *) &broadcastPermission, sizeof(broadcastPermission));
}
void UDPSocket::disconnect()
{
void UDPSocket::disconnect() {
sockaddr_in nullAddr;
memset(&nullAddr, 0, sizeof(nullAddr));
nullAddr.sin_family = AF_UNSPEC;
// Try to disconnect
if (::connect(sockDesc, (sockaddr *) &nullAddr, sizeof(nullAddr)) < 0)
{
if (errno != EAFNOSUPPORT)
{
if (::connect(sockDesc, (sockaddr *) &nullAddr, sizeof(nullAddr)) < 0) {
if (errno != EAFNOSUPPORT) {
throw SocketException("Disconnect failed (connect())", true);
}
}
@ -363,20 +318,17 @@ void UDPSocket::sendTo(const void *buffer, int bufferLen,
// Write out the whole buffer as a single message.
if (sendto(sockDesc, (raw_type *) buffer, bufferLen, 0,
(sockaddr *) &destAddr, sizeof(destAddr)) != bufferLen)
{
(sockaddr *) &destAddr, sizeof(destAddr)) != bufferLen) {
throw SocketException("Send failed (sendto())", true);
}
}
int UDPSocket::recvFrom(void *buffer, int bufferLen, string &sourceAddress,
unsigned short &sourcePort)
{
unsigned short &sourcePort) {
sockaddr_in clntAddr;
socklen_t addrLen = sizeof(clntAddr);
int rtn;
if ((rtn = recvfrom(sockDesc, (raw_type *) buffer, bufferLen, MSG_DONTWAIT, (sockaddr *) &clntAddr,
(socklen_t *) &addrLen)) < 0)
if ((rtn = recvfrom(sockDesc, (raw_type *) buffer, bufferLen, MSG_DONTWAIT, (sockaddr *) &clntAddr, (socklen_t *) &addrLen)) < 0)
{
throw SocketException("Receive failed (recvfrom())", true);
}
@ -386,39 +338,33 @@ int UDPSocket::recvFrom(void *buffer, int bufferLen, string &sourceAddress,
return rtn;
}
void UDPSocket::setMulticastTTL(unsigned char multicastTTL)
{
void UDPSocket::setMulticastTTL(unsigned char multicastTTL) {
if (setsockopt(sockDesc, IPPROTO_IP, IP_MULTICAST_TTL,
(raw_type *) &multicastTTL, sizeof(multicastTTL)) < 0)
{
(raw_type *) &multicastTTL, sizeof(multicastTTL)) < 0) {
throw SocketException("Multicast TTL set failed (setsockopt())", true);
}
}
void UDPSocket::joinGroup(const string &multicastGroup)
{
void UDPSocket::joinGroup(const string &multicastGroup) {
struct ip_mreq multicastRequest;
multicastRequest.imr_multiaddr.s_addr = inet_addr(multicastGroup.c_str());
multicastRequest.imr_interface.s_addr = htonl(INADDR_ANY);
if (setsockopt(sockDesc, IPPROTO_IP, IP_ADD_MEMBERSHIP,
(raw_type *) &multicastRequest,
sizeof(multicastRequest)) < 0)
{
sizeof(multicastRequest)) < 0) {
throw SocketException("Multicast group join failed (setsockopt())", true);
}
}
void UDPSocket::leaveGroup(const string &multicastGroup)
{
void UDPSocket::leaveGroup(const string &multicastGroup) {
struct ip_mreq multicastRequest;
multicastRequest.imr_multiaddr.s_addr = inet_addr(multicastGroup.c_str());
multicastRequest.imr_interface.s_addr = htonl(INADDR_ANY);
if (setsockopt(sockDesc, IPPROTO_IP, IP_DROP_MEMBERSHIP,
(raw_type *) &multicastRequest,
sizeof(multicastRequest)) < 0)
{
sizeof(multicastRequest)) < 0) {
throw SocketException("Multicast group leave failed (setsockopt())", true);
}
}

18
Socket.h
View file

@ -9,8 +9,7 @@
/**
* Signals a problem with the execution of a socket call.
*/
class SocketException : public std::exception
{
class SocketException : public std::exception {
public:
/**
* Construct a SocketException with a explanatory message.
@ -38,8 +37,7 @@ private:
/**
* Base class representing basic communication endpoint
*/
class Socket
{
class Socket {
public:
/**
* Close and deallocate this socket
@ -120,8 +118,7 @@ protected:
/**
* Socket which is able to connect, send, and receive
*/
class CommunicatingSocket : public Socket
{
class CommunicatingSocket : public Socket {
public:
/**
* Establish a socket connection with the given foreign
@ -174,8 +171,7 @@ protected:
/**
* TCP socket for communication with other TCP sockets
*/
class TCPSocket : public CommunicatingSocket
{
class TCPSocket : public CommunicatingSocket {
public:
/**
* Construct a TCP socket with no connection
@ -204,8 +200,7 @@ private:
/**
* TCP socket class for servers
*/
class TCPServerSocket : public Socket
{
class TCPServerSocket : public Socket {
public:
/**
* Construct a TCP socket for use with a server, accepting connections
@ -246,8 +241,7 @@ private:
/**
* UDP socket class
*/
class UDPSocket : public CommunicatingSocket
{
class UDPSocket : public CommunicatingSocket {
public:
/**
* Construct a UDP socket

59
main.cpp
View file

@ -29,18 +29,18 @@ struct PwlPriv
struct Config config;
TCPSocket* socket = nullptr;
TCPSocket* socket;
spa_source* timer_source;
struct timespec timout_timespan;
};
static bool send_sensor(TCPSocket* socket, uint8_t id, bool state)
static void send_sensor(TCPSocket* socket, uint8_t id, bool state)
{
static int8_t lastState = -1;
if(state == lastState)
return true;
return;
lastState = state;
@ -50,48 +50,13 @@ static bool send_sensor(TCPSocket* socket, uint8_t id, bool state)
json.append(" }] }\n");
std::string sendbuf(MSG_STRING + std::to_string(json.length()) + "\n" + json);
Log(Log::DEBUG)<<"Sending: \n"<<sendbuf;
try
{
socket->send(sendbuf.c_str(), sendbuf.size());
}
catch(SocketException &e)
{
return false;
}
return true;
}
static TCPSocket* check_socket(PwlPriv *priv)
{
if(priv->socket)
return priv->socket;
try
{
priv->socket = new TCPSocket(priv->config.host, priv->config.port);
}
catch(SocketException &e)
{
Log(Log::WARN)<<e.what();
delete priv->socket;
priv->socket = nullptr;
}
return priv->socket;
}
static void off_timeout(void *data, uint64_t expirations)
{
PwlPriv *priv = reinterpret_cast<PwlPriv*>(data);
priv->socket = check_socket(priv);
if(priv->socket)
{
bool ret = send_sensor(priv->socket, priv->config.id, false);
if(!ret)
{
delete priv->socket;
priv->socket = nullptr;
}
}
send_sensor(priv->socket, priv->config.id, false);
}
static void state_changed(PwlPriv *priv, bool state)
@ -105,16 +70,7 @@ static void state_changed(PwlPriv *priv, bool state)
if(state)
{
priv->socket = check_socket(priv);
if(priv->socket)
{
bool ret = send_sensor(priv->socket, priv->config.id, false);
if(!ret)
{
delete priv->socket;
priv->socket = nullptr;
}
}
send_sensor(priv->socket, priv->config.id, true);
pw_loop_update_timer(pw_main_loop_get_loop(priv->loop), priv->timer_source, NULL, NULL, false);
}
else
@ -305,9 +261,8 @@ int main(int argc, char *argv[])
}
catch(SocketException &e)
{
Log(Log::WARN)<<e.what();
delete priv.socket;
priv.socket = nullptr;
Log(Log::ERROR)<<e.what();
return 1;
}
priv.loop = pw_main_loop_new(NULL);