wirenboard / wb-mqtt-serial / 680

Committed 05 Aug 2025 11:42AM UTC coverage: 73.001% (-0.002%) from 73.003%

Build # 680

Build Type

push

github

Committed by

web-flow

Commit Message

Use response_timeout_ms from port settings for RPC requests (#978)

Coverage Stats

6493 of 9259 branches covered (70.13%)

16 of 26 new or added lines in 6 files covered. (61.54%)

12408 of 16997 relevant lines covered (73.0%)

373.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

9.84

/src/tcp_port.cpp

#include "tcp_port.h"
#include "serial_exc.h"

#include <iostream>
#include <sstream>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/types.h>

#include "log.h"

#define LOG(logger) ::logger.Log() << "[tcp port] "

using namespace std;

namespace
{
    const int CONNECTION_TIMEOUT_S = 5;

    // Additional timeout for reading from tcp port. It is caused by intermediate hardware and internal Linux processing
    // Values are taken from old default timeouts
    const std::chrono::microseconds ResponseTCPLag = std::chrono::microseconds(500000);
    const std::chrono::microseconds FrameTCPLag = std::chrono::microseconds(150000);
}

TTcpPort::TTcpPort(const TTcpPortSettings& settings): Settings(settings)
{}

void TTcpPort::Open()
{
    if (IsOpen()) {
        throw TSerialDeviceException("port is already open");
    }

    struct hostent* server = gethostbyname(Settings.Address.c_str());
    if (!server) {
        throw TSerialDeviceException("no such host: " + Settings.Address);
    }

    Fd = socket(AF_INET, SOCK_STREAM, 0);
    if (Fd < 0) {
        throw TSerialDeviceErrnoException("cannot open tcp port: ", errno);
    }

    struct sockaddr_in serv_addr;
    memset((char*)&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    memmove((char*)&serv_addr.sin_addr.s_addr, (char*)server->h_addr, server->h_length);
    serv_addr.sin_port = htons(Settings.Port);

    // set socket to non-blocking state
    auto arg = fcntl(Fd, F_GETFL, NULL);
    arg |= O_NONBLOCK;
    fcntl(Fd, F_SETFL, arg);

    try {
        if (connect(Fd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
            if (errno != EINPROGRESS) {
                throw std::runtime_error("connect error: " + FormatErrno(errno));
            }
            timeval tv = {CONNECTION_TIMEOUT_S, 0};
            fd_set myset;
            FD_ZERO(&myset);
            FD_SET(Fd, &myset);
            auto res = select(Fd + 1, NULL, &myset, NULL, &tv);
            if (res > 0) {
                socklen_t lon = sizeof(int);
                int valopt;
                getsockopt(Fd, SOL_SOCKET, SO_ERROR, (void*)(&valopt), &lon);
                if (valopt) {
                    throw std::runtime_error("connect error: " + FormatErrno(valopt));
                }
            } else if (res < 0 && errno != EINTR) {
                throw std::runtime_error("connect error: " + FormatErrno(errno));
            } else {
                throw std::runtime_error("connect error: timeout");
            }
        }
    } catch (const std::runtime_error& e) {
        close(Fd);
        Fd = -1;
        throw TSerialDeviceException(GetDescription() + " " + e.what());
    }

    // set socket back to blocking state
    arg = fcntl(Fd, F_GETFL, NULL);
    arg &= (~O_NONBLOCK);
    fcntl(Fd, F_SETFL, arg);

    LastInteraction = std::chrono::steady_clock::now();
}

void TTcpPort::OnReadyEmptyFd()
{
    Close();
    throw TSerialDeviceTransientErrorException("socket closed");
}

void TTcpPort::WriteBytes(const uint8_t* buf, int count)
{
    if (IsOpen()) {
        Base::WriteBytes(buf, count);
    } else {
        LOG(Debug) << "Attempt to write to not open port";
    }
}

uint8_t TTcpPort::ReadByte(const std::chrono::microseconds& timeout)
{
    return Base::ReadByte(CalcResponseTimeout(timeout) + ResponseTCPLag);
}

TReadFrameResult TTcpPort::ReadFrame(uint8_t* buf,
                                     size_t count,
                                     const std::chrono::microseconds& responseTimeout,
                                     const std::chrono::microseconds& frameTimeout,
                                     TFrameCompletePred frame_complete)
{
    if (IsOpen()) {
        return Base::ReadFrame(buf,
                               count,
                               CalcResponseTimeout(responseTimeout) + ResponseTCPLag,
                               frameTimeout + FrameTCPLag,
                               frame_complete);
    }
    LOG(Debug) << "Attempt to read from not open port";
    return TReadFrameResult();
}

std::string TTcpPort::GetDescription(bool verbose) const
{
    if (verbose) {
        return Settings.ToString();
    }
    return Settings.Address + ":" + std::to_string(Settings.Port);
}

1	#include "tcp_port.h"
2	#include "serial_exc.h"
3
4	#include <iostream>
5	#include <sstream>
6	#include <stdio.h>
7	#include <stdlib.h>
8	#include <string.h>
9	#include <unistd.h>
10
11	#include <fcntl.h>
12	#include <netdb.h>
13	#include <netinet/in.h>
14	#include <sys/socket.h>
15	#include <sys/types.h>
16
17	#include "log.h"
18
19	#define LOG(logger) ::logger.Log() << "[tcp port] "
20
21	using namespace std;
22
23	namespace
24	{
25	const int CONNECTION_TIMEOUT_S = 5;
26
27	// Additional timeout for reading from tcp port. It is caused by intermediate hardware and internal Linux processing
28	// Values are taken from old default timeouts
29	const std::chrono::microseconds ResponseTCPLag = std::chrono::microseconds(500000);
30	const std::chrono::microseconds FrameTCPLag = std::chrono::microseconds(150000);
31	}
32
33	TTcpPort::TTcpPort(const TTcpPortSettings& settings): Settings(settings)	4✔
34	{}	4✔
35
36	void TTcpPort::Open()	×
37	{
38	if (IsOpen()) {	×
39	throw TSerialDeviceException("port is already open");	×
40	}
41
42	struct hostent* server = gethostbyname(Settings.Address.c_str());	×
43	if (!server) {	×
44	throw TSerialDeviceException("no such host: " + Settings.Address);	×
45	}
46
47	Fd = socket(AF_INET, SOCK_STREAM, 0);	×
48	if (Fd < 0) {	×
49	throw TSerialDeviceErrnoException("cannot open tcp port: ", errno);	×
50	}
51
52	struct sockaddr_in serv_addr;
53	memset((char*)&serv_addr, 0, sizeof(serv_addr));	×
54	serv_addr.sin_family = AF_INET;	×
55	memmove((char)&serv_addr.sin_addr.s_addr, (char)server->h_addr, server->h_length);	×
56	serv_addr.sin_port = htons(Settings.Port);	×
57
58	// set socket to non-blocking state
59	auto arg = fcntl(Fd, F_GETFL, NULL);	×
60	arg \|= O_NONBLOCK;	×
61	fcntl(Fd, F_SETFL, arg);	×
62
63	try {
64	if (connect(Fd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {	×
65	if (errno != EINPROGRESS) {	×
66	throw std::runtime_error("connect error: " + FormatErrno(errno));	×
67	}
68	timeval tv = {CONNECTION_TIMEOUT_S, 0};	×
69	fd_set myset;
70	FD_ZERO(&myset);	×
71	FD_SET(Fd, &myset);	×
72	auto res = select(Fd + 1, NULL, &myset, NULL, &tv);	×
73	if (res > 0) {	×
74	socklen_t lon = sizeof(int);	×
75	int valopt;
76	getsockopt(Fd, SOL_SOCKET, SO_ERROR, (void*)(&valopt), &lon);	×
77	if (valopt) {	×
78	throw std::runtime_error("connect error: " + FormatErrno(valopt));	×
79	}
80	} else if (res < 0 && errno != EINTR) {	×
81	throw std::runtime_error("connect error: " + FormatErrno(errno));	×
82	} else {
83	throw std::runtime_error("connect error: timeout");	×
84	}
85	}
86	} catch (const std::runtime_error& e) {	×
87	close(Fd);	×
88	Fd = -1;	×
89	throw TSerialDeviceException(GetDescription() + " " + e.what());	×
90	}
91
92	// set socket back to blocking state
93	arg = fcntl(Fd, F_GETFL, NULL);	×
94	arg &= (~O_NONBLOCK);	×
95	fcntl(Fd, F_SETFL, arg);	×
96
97	LastInteraction = std::chrono::steady_clock::now();	×
98	}
99
100	void TTcpPort::OnReadyEmptyFd()	×
101	{
102	Close();	×
103	throw TSerialDeviceTransientErrorException("socket closed");	×
104	}
105
106	void TTcpPort::WriteBytes(const uint8_t* buf, int count)	×
107	{
108	if (IsOpen()) {	×
109	Base::WriteBytes(buf, count);	×
110	} else {
111	LOG(Debug) << "Attempt to write to not open port";	×
112	}
113	}
114
115	uint8_t TTcpPort::ReadByte(const std::chrono::microseconds& timeout)	×
116	{
NEW 117	return Base::ReadByte(CalcResponseTimeout(timeout) + ResponseTCPLag);	×
118	}
119
120	TReadFrameResult TTcpPort::ReadFrame(uint8_t* buf,	×
121	size_t count,
122	const std::chrono::microseconds& responseTimeout,
123	const std::chrono::microseconds& frameTimeout,
124	TFrameCompletePred frame_complete)
125	{
126	if (IsOpen()) {	×
127	return Base::ReadFrame(buf,
128	count,
NEW 129	CalcResponseTimeout(responseTimeout) + ResponseTCPLag,	×
130	frameTimeout + FrameTCPLag,	×
131	frame_complete);	×
132	}
133	LOG(Debug) << "Attempt to read from not open port";	×
134	return TReadFrameResult();	×
135	}
136
137	std::string TTcpPort::GetDescription(bool verbose) const	6✔
138	{
139	if (verbose) {	6✔
140	return Settings.ToString();	2✔
141	}
142	return Settings.Address + ":" + std::to_string(Settings.Port);	8✔
143	}

wirenboard / wb-mqtt-serial / 680

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous