cptlobster/video-streaming-poc
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

switch to Boost for network connectivity

Browse Source
This commit is contained in:
Camryn Thomas
2025-09-26 20:03:28 -05:00
parent 5f57f4631b
commit 80690945fe
3 changed files with 98 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
client.cpp
View File

@@ -6,39 +6,55 @@
#include <unistd.h>
#include "transfer.h"
#include "logging.h"
#include <chrono>
#include <boost/asio.hpp>
using namespace std;
using boost::asio::ip::tcp;
int main() {
const int FRAME_DELAY_MS = 1000 / 30;
// create video capture
cv::VideoCapture cap = cv::VideoCapture(0);
try {
boost::asio::io_context io_context;
tcp::resolver resolver(io_context);
// create socket
int clientSocket = socket(AF_INET, SOCK_STREAM, 0);
tcp::resolver::results_type endpoints =
resolver.resolve("127.0.0.1", "8080");
// specifying address
sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_port = htons(8080);
serverAddress.sin_addr.s_addr = INADDR_ANY;
cv::Mat image = cv::Mat::zeros(cv::Size(640, 480), CV_8UC3);
cv::Mat image = cv::Mat::zeros(cv::Size(640, 480), CV_8UC3);
info("Ready to connect.");
// sending connection request
tcp::socket socket(io_context);
boost::asio::connect(socket, endpoints);
info("Ready to connect.");
// sending connection request
connect(clientSocket, reinterpret_cast<sockaddr *>(&serverAddress),
sizeof(serverAddress));
info("Connected.");
// create buffer for serialization
vector<uchar> imgbuf;
info("Connected.");
// create buffer for serialization
vector<uchar> imgbuf;
while (true) {
auto start_time = std::chrono::high_resolution_clock::now();
cap.read(image);
trace("Sending image");
sendImage(socket, image, imgbuf);
while (true) {
cap.read(image);
trace("Sending image");
sendImage(clientSocket, image, imgbuf);
auto end_time = std::chrono::high_resolution_clock::now();
auto elapsed_time = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time);
// Calculate the remaining time to sleep
auto sleep_duration = std::chrono::milliseconds(FRAME_DELAY_MS) - elapsed_time;
// Sleep for the remaining duration if positive
if (sleep_duration.count() > 0) {
std::this_thread::sleep_for(sleep_duration);
}
}
}
catch (std::exception& e) {
error(e.what());
}
close(clientSocket);
return 0;
}

64
server.cpp
View File

@@ -1,5 +1,6 @@
#include <opencv2/highgui.hpp>
#include <opencv2/core/mat.hpp>
#include <opencv2/imgproc.hpp>
#include <cstring>
#include <iostream>
#include <netinet/in.h>
@@ -7,50 +8,49 @@
#include <unistd.h>
#include "transfer.h"
#include "logging.h"
#include <boost/asio.hpp>
using namespace std;
using boost::asio::ip::tcp;
int main() {
// TODO: read image data from socket instead of VideoCapture
// creating socket
int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
try {
boost::asio::io_context io_context;
// creating socket
// specifying the address
sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_port = htons(8080);
serverAddress.sin_addr.s_addr = INADDR_ANY;
tcp::acceptor acceptor(io_context, tcp::endpoint(tcp::v4(), 8080));
// binding socket.
bind(serverSocket, reinterpret_cast<sockaddr *>(&serverAddress),
sizeof(serverAddress));
tcp::socket socket(io_context);
info("Ready to accept connections.");
// listening to the assigned socket
listen(serverSocket, 5);
info("Ready to accept connections.");
// accepting connection request
int clientSocket = accept(serverSocket, nullptr, nullptr);
info("Client connected.");
// accepting connection request
acceptor.accept(socket);
info("Client connected.");
// TODO: handle multiple images
cv::Mat image = cv::Mat::zeros(cv::Size(640, 480), CV_8UC3);
// TODO: handle multiple images
cv::Mat image = cv::Mat::zeros(cv::Size(640, 480), CV_8UC3);
bool running = true;
// TODO: make this asynchronous. probably do that in tandem with setting up networking
while (running) {
// receive data
vector<uchar> buffer;
trace("Receiving image");
recvImage(clientSocket, buffer);
bool running = true;
// TODO: make this asynchronous. probably do that in tandem with setting up networking
while (running) {
// receive data
vector<uchar> buffer;
trace("Receiving image");
int latency = recvImage(socket, buffer);
trace("Applying new data to image");
applyImage(image, &buffer);
trace("Applying new data to image");
applyImage(image, &buffer);
trace("Displaying image");
imshow("image", image);
running = cv::waitKey(30) != 27;
cv::putText(image, std::format("Latency: {}ms", latency), cv::Point(0, 480), cv::FONT_HERSHEY_PLAIN, 1, cv::Scalar(255, 0, 0), 1, cv::LINE_AA);
trace("Displaying image");
imshow("image", image);
running = cv::waitKey(30) != 27;
}
}
catch (std::exception& e) {
error(e.what());
}
close(serverSocket);
return 0;
}

51
transfer.h
View File

@@ -5,10 +5,12 @@
#include <opencv2/core/mat.hpp>
#include <opencv2/imgcodecs.hpp>
#include <sys/socket.h>
#include <logging.h>
#include "logging.h"
#include <vector>
#include <boost/asio.hpp>
using namespace std;
using boost::asio::ip::tcp;
struct imageHeader {
size_t size;
@@ -29,7 +31,9 @@ inline void serializeImage(const cv::Mat& image, std::vector<uchar>& buffer) {
cv::imencode(".jpg", image, buffer);
}
int sendImage(int socket, const cv::Mat& image, std::vector<uchar>& buffer) {
void sendImage(tcp::socket& socket, const cv::Mat& image, std::vector<uchar>& buffer) {
boost::system::error_code e;
serializeImage(image, buffer);
size_t totalSent = 0;
@@ -40,30 +44,33 @@ int sendImage(int socket, const cv::Mat& image, std::vector<uchar>& buffer) {
header.size = size;
header.timestamp = timestamp;
trace("Buffer size: ", size);
if (const ssize_t sent = send(socket, &header, sizeof(header), 0); sent == -1) {
error("Error sending data header");
return -1;
if (const ssize_t sent = boost::asio::write(socket, boost::asio::buffer(&header, sizeof(header)), e); sent == -1) {
throw boost::system::system_error(e);
}
// then start sending the serialized image
while (totalSent < size) {
const ssize_t sent = send(socket, buffer.data() + totalSent, size - totalSent, 0);
const ssize_t sent = boost::asio::write(socket, boost::asio::buffer(buffer), e);
if (sent == -1) {
error("Error sending data");
return -1;
throw boost::system::system_error(e);
}
totalSent += sent;
debug("Packet sent (", sent, " bytes, total ", totalSent, " bytes)");
debug("Packet sent (", sent, " bytes, total ", totalSent, " / ", size, " bytes)");
}
return 0;
}
int recvImage(int socket, std::vector<uchar>& buffer) {
int recvImage(tcp::socket& socket, std::vector<uchar>& buffer) {
boost::system::error_code e;
// first receive the size of the image
imageHeader header;
if (ssize_t recvd = recv(socket, &header, sizeof(imageHeader), 0); recvd <= 0) {
error("Error receiving data header");
return -1;
size_t len = 0;
while (len < sizeof(imageHeader)) {
len += socket.read_some(boost::asio::buffer(&header, sizeof(header)), e);
if (e.failed()) {
throw boost::system::system_error(e);
}
}
size_t dataSize = header.size;
@@ -77,20 +84,20 @@ int recvImage(int socket, std::vector<uchar>& buffer) {
// start receiving the image until the whole thing arrives
size_t totalReceived = 0;
while (totalReceived < dataSize) {
ssize_t bytesReceived = recv(socket, buffer.data() + totalReceived, dataSize, 0);
if (bytesReceived <= 0) {
error("Error receiving data");
buffer.clear();
return -1;
}
size_t bytesReceived = boost::asio::read(socket, boost::asio::buffer(buffer), e);
if (e == boost::asio::error::eof) // Connection closed cleanly by peer.
throw boost::system::system_error(e);
else if (e)
throw boost::system::system_error(e);
totalReceived += bytesReceived;
debug("Packet received (", bytesReceived, " bytes, total ", totalReceived, " bytes)");
debug("Packet received (", bytesReceived, " bytes, total ", totalReceived, " / ", dataSize, " bytes)");
}
chrono::milliseconds currentTime = getMillis();
chrono::milliseconds diff = currentTime - sentTime;
debug("Packet latency: ", diff.count(), "ms");
return 0;
return diff.count();
}
bool applyImage(cv::Mat& image, std::vector<uchar> *src) {